| 118 | } |
| 119 | |
| 120 | comm::RetCode SyncCtrl::AdjustNextCursorID(const int sub_id, const int queue_id, |
| 121 | uint64_t &prev_cursor_id, uint64_t &next_cursor_id) { |
| 122 | auto opt = impl_->store->GetStoreOption(); |
| 123 | |
| 124 | if (!sub_id || sub_id > opt->nsub || queue_id >= opt->nqueue) |
| 125 | return comm::RetCode::RET_ERR_ARG; |
| 126 | |
| 127 | size_t idx; |
| 128 | GetIdx(opt->nsub, sub_id, queue_id, idx); |
| 129 | |
| 130 | std::lock_guard<mutex> lock_guard(impl_->locks[idx]); |
| 131 | |
| 132 | auto &&item = impl_->buf[idx]; |
| 133 | |
| 134 | if (item.next_magic != SYNCCTRL_MAGIC || item.next_cursor_id != next_cursor_id || |
| 135 | (-1 != prev_cursor_id && prev_cursor_id > next_cursor_id)) { |
| 136 | if (item.prev_magic != SYNCCTRL_MAGIC) return comm::RetCode::RET_ERR_CURSOR_NOT_FOUND; |
| 137 | prev_cursor_id = next_cursor_id = item.prev_cursor_id; |
| 138 | } |
| 139 | |
| 140 | if (item.prev_magic == SYNCCTRL_MAGIC && |
| 141 | (prev_cursor_id == -1 || item.prev_cursor_id > prev_cursor_id)) { |
| 142 | prev_cursor_id = item.prev_cursor_id; |
| 143 | } |
| 144 | |
| 145 | return comm::RetCode::RET_OK; |
| 146 | } |
| 147 | |
| 148 | comm::RetCode SyncCtrl::UpdateCursorID(const int sub_id, const int queue_id, |
| 149 | const uint64_t cursor_id, const bool is_prev) { |
no test coverage detected