| 325 | |
| 326 | |
| 327 | comm::RetCode SyncCtrl::SyncCursorID(const proto::SyncCtrlInfo &sync_ctrl_info) { |
| 328 | comm::RetCode ret; |
| 329 | |
| 330 | uint64_t cur_prev_cursor_id; |
| 331 | |
| 332 | for (size_t i{0}; i < sync_ctrl_info.queue_details_size(); ++i) { |
| 333 | const proto::SyncCtrlInfo::QueueDetail &queue_detail = sync_ctrl_info.queue_details(i); |
| 334 | int queue_id = queue_detail.queue_id(); |
| 335 | |
| 336 | uint64_t max_prev_cursor_id = -1; |
| 337 | for (size_t j{0}; j < queue_detail.sub_details_size(); ++j) { |
| 338 | const proto::SyncCtrlInfo::QueueDetail::SubDetail & |
| 339 | sub_detail(queue_detail.sub_details(j)); |
| 340 | int sub_id = sub_detail.sub_id(); |
| 341 | uint64_t prev_cursor_id = sub_detail.prev_cursor_id(); |
| 342 | if (-1 == max_prev_cursor_id || prev_cursor_id > max_prev_cursor_id) |
| 343 | max_prev_cursor_id = prev_cursor_id; |
| 344 | |
| 345 | if (0 > as_integer(ret = GetCursorID(sub_id, queue_id, cur_prev_cursor_id))) { |
| 346 | QLErr("GetCursorID ret %d sub_id %d queue %d", |
| 347 | as_integer(ret), sub_id, queue_id); |
| 348 | } else if (!as_integer(ret) && cur_prev_cursor_id >= prev_cursor_id) { |
| 349 | continue; |
| 350 | } |
| 351 | |
| 352 | QLVerb("sync prev_cursor_id. sub_id %d queue_id %d", sub_id, queue_id); |
| 353 | |
| 354 | if (comm::RetCode::RET_OK != |
| 355 | (ret = UpdateCursorID(sub_id, queue_id, prev_cursor_id))) { |
| 356 | QLErr("UpdateCursorID ret %d sub_id %d queue_id %d", |
| 357 | as_integer(ret), sub_id, queue_id); |
| 358 | return ret; |
| 359 | } |
| 360 | |
| 361 | if (comm::RetCode::RET_OK != |
| 362 | (ret = UpdateCursorID(sub_id, queue_id, prev_cursor_id, false))) { |
| 363 | QLErr("UpdateCursorID ret %d sub_id %d queue_id %d", |
| 364 | as_integer(ret), sub_id, queue_id); |
| 365 | return ret; |
| 366 | } |
| 367 | } |
| 368 | |
| 369 | auto meta_queue(impl_->store->GetBaseMgr()->GetMetaQueue(queue_id)); |
| 370 | if (-1 == max_prev_cursor_id) meta_queue->ClearDequeueStat(); |
| 371 | else meta_queue->UpdateDeuqueStat(StoreMeta(max_prev_cursor_id)); |
| 372 | } |
| 373 | |
| 374 | return comm::RetCode::RET_OK; |
| 375 | } |
| 376 | |
| 377 | |
| 378 | } // namespace store |
no test coverage detected