| 1112 | } |
| 1113 | |
| 1114 | Status ReplicationThread::fetchFiles(int sock_fd, const std::string &dir, const std::vector<std::string> &files, |
| 1115 | const std::vector<uint32_t> &crcs, const FetchFileCallback &fn, ssl_st *ssl) { |
| 1116 | std::string files_str; |
| 1117 | for (const auto &file : files) { |
| 1118 | files_str += file; |
| 1119 | files_str.push_back(','); |
| 1120 | } |
| 1121 | files_str.pop_back(); |
| 1122 | |
| 1123 | const auto fetch_command = redis::ArrayOfBulkStrings({"_fetch_file", files_str}); |
| 1124 | auto s = util::SockSend(sock_fd, fetch_command, ssl); |
| 1125 | if (!s.IsOK()) return s.Prefixed("send fetch file command"); |
| 1126 | |
| 1127 | UniqueEvbuf evbuf; |
| 1128 | for (unsigned i = 0; i < files.size(); i++) { |
| 1129 | DEBUG("[fetch] Start to fetch file {}", files[i]); |
| 1130 | s = fetchFile(sock_fd, evbuf.get(), dir, files[i], crcs[i], fn, ssl); |
| 1131 | if (!s.IsOK()) { |
| 1132 | s = Status(Status::NotOK, "fetch file err: " + s.Msg()); |
| 1133 | WARN("[fetch] Fail to fetch file {}, err: {}", files[i], s.Msg()); |
| 1134 | break; |
| 1135 | } |
| 1136 | DEBUG("[fetch] Succeed fetching file {}", files[i]); |
| 1137 | |
| 1138 | // Just for tests |
| 1139 | if (srv_->GetConfig()->fullsync_recv_file_delay) { |
| 1140 | sleep(srv_->GetConfig()->fullsync_recv_file_delay); |
| 1141 | } |
| 1142 | } |
| 1143 | return s; |
| 1144 | } |
| 1145 | |
| 1146 | // Check if stop_flag_ is set, when do, tear down replication |
| 1147 | void ReplicationThread::TimerCB(int, int16_t) { |
no test coverage detected