| 147 | } |
| 148 | |
| 149 | comm::RetCode BatchTask::Process(bool is_timeout) { |
| 150 | auto now_timestamp_ms = comm::utils::Time::GetSteadyClockMS(); |
| 151 | uint64_t time_wait_ms = now_timestamp_ms - start_timestamp_ms_; |
| 152 | |
| 153 | comm::proto::AddRequest batch_req; |
| 154 | comm::proto::AddResponse batch_resp; |
| 155 | |
| 156 | for (auto &task : tasks_) { |
| 157 | auto req = task->GetReq(); |
| 158 | if (nullptr == req) continue; |
| 159 | if (0 == batch_req.items_size()) { |
| 160 | batch_req.CopyFrom(*req); |
| 161 | } else { |
| 162 | for (int i{0}; i < req->items_size(); ++i) { |
| 163 | batch_req.add_items()->CopyFrom(req->items(i)); |
| 164 | } |
| 165 | } |
| 166 | } |
| 167 | |
| 168 | auto retcode = producer_->RawAdd(batch_req, batch_resp); |
| 169 | |
| 170 | comm::ProducerBP::GetThreadInstance()->OnBatchStat(batch_req, retcode, time_wait_ms, is_timeout); |
| 171 | //printf("batch %d time_wait_ms %" PRIu64 " is_timeout %d\n", batch_req.items_size(), time_wait_ms, is_timeout); |
| 172 | |
| 173 | for (auto &task : tasks_) { |
| 174 | task->Notify(retcode); |
| 175 | } |
| 176 | return retcode; |
| 177 | } |
| 178 | |
| 179 | |
| 180 | //////////////////////////////////////////////////////////////////////////////////// |
no test coverage detected