| 52 | } |
| 53 | |
| 54 | int ReplicationTransfer::Process() { |
| 55 | ctx_->Wait(); |
| 56 | |
| 57 | ColorLogInfo("dc process begin"); |
| 58 | |
| 59 | if (ctx_->GetSlaveFD() < 0) { |
| 60 | ColorLogError("%s slave has closed, exit", __func__); |
| 61 | return SLAVE_FAIL; |
| 62 | } |
| 63 | |
| 64 | int ret = OK; |
| 65 | //init the gtid to get the data pos |
| 66 | while (1) { |
| 67 | vector < string > recv_buff; |
| 68 | ret = ReadDataFromDC(&recv_buff); |
| 69 | if (ret != OK && ret != DATA_EMPTY) |
| 70 | break; |
| 71 | |
| 72 | if (ctx_->GetSlaveFD() < 0) { |
| 73 | ColorLogError("%s slave has closed, exit", __func__); |
| 74 | break; |
| 75 | } |
| 76 | |
| 77 | if (recv_buff.size() == 0) |
| 78 | continue; |
| 79 | |
| 80 | for (size_t i = 0; i < recv_buff.size(); ++i) { |
| 81 | int seq = ctx_->GetSeq(); |
| 82 | ret = NetIO::SendWithSeq(ctx_->GetSlaveFD(), recv_buff[i], seq++); |
| 83 | ctx_->SetSeq(seq); |
| 84 | if (ret) { |
| 85 | ColorLogError("%s send buffer fail len %zu ret %d seq %d", "slave", recv_buff[i].size(), ret, seq); |
| 86 | STATISTICS(ReplSendDataFail()); |
| 87 | break; |
| 88 | } |
| 89 | } |
| 90 | |
| 91 | if (ret) |
| 92 | break; |
| 93 | |
| 94 | LogVerbose("%s recv buffer list size %zu ret %d", "dc", recv_buff.size(), ret); |
| 95 | } |
| 96 | |
| 97 | ColorLogInfo("%s transfer done, ret %d", __func__, ret); |
| 98 | return ret; |
| 99 | } |
| 100 | |
| 101 | int ReplicationTransfer::ReadDataFromDC(vector<string> *buffer) { |
| 102 | return GetEvents(buffer); |
nothing calls this directly
no test coverage detected