| 439 | |
| 440 | |
| 441 | comm::RetCode Store::Add(const comm::proto::AddRequest &req, comm::proto::AddResponse &resp) { |
| 442 | QLVerb("Add"); |
| 443 | |
| 444 | comm::RetCode ret; |
| 445 | |
| 446 | comm::StoreBP::GetThreadInstance()->OnAdd(req); |
| 447 | if (0 == req.items_size()) return comm::RetCode::RET_OK; |
| 448 | |
| 449 | if (comm::RetCode::RET_OK != (ret = CheckAddRequest(req))) { |
| 450 | comm::StoreBP::GetThreadInstance()->OnAddRequestInvalid(req); |
| 451 | QLErr("CheckAddRequest ret %d", as_integer(ret)); |
| 452 | return ret; |
| 453 | } |
| 454 | |
| 455 | int paxos_group_id = req.queue_id() % impl_->opt.ngroup; |
| 456 | if (comm::RetCode::RET_OK != |
| 457 | (ret = CheckMaster(paxos_group_id, *resp.mutable_redirect_addr()))) { |
| 458 | comm::StoreBP::GetThreadInstance()->OnAddCheckMasterUnpass(req); |
| 459 | if (as_integer(ret) < 0) { |
| 460 | QLErr("CheckMaster ret %d paxos_group_id %d", as_integer(ret), paxos_group_id); |
| 461 | } |
| 462 | return ret; |
| 463 | } |
| 464 | comm::StoreBP::GetThreadInstance()->OnAddCheckMasterPass(req); |
| 465 | |
| 466 | ret = PaxosAdd(req, resp); |
| 467 | if (comm::RetCode::RET_OK != ret) { |
| 468 | QLErr("PaxosAdd ret %d", as_integer(ret)); |
| 469 | } |
| 470 | |
| 471 | QLVerb("Add succ. req: topic_id %d store_id %d queue_id %d " |
| 472 | "items.size() %d. resp: cursor_id %" PRIu64, |
| 473 | req.topic_id(), req.store_id(), req.queue_id(), req.items_size(), |
| 474 | resp.cursor_id()); |
| 475 | for (int i{0}; i < req.items_size(); i++) { |
| 476 | auto &&item = req.items(i); |
| 477 | QLVerb("item: topic_id %d uin %" PRIu64, item.meta().topic_id(), item.meta().uin()); |
| 478 | } |
| 479 | |
| 480 | return ret; |
| 481 | } |
| 482 | |
| 483 | comm::RetCode Store::PaxosAdd(const comm::proto::AddRequest &req, comm::proto::AddResponse &resp) { |
| 484 | comm::StoreBP::GetThreadInstance()->OnPaxosAdd(req); |
no test coverage detected