| 35 | } |
| 36 | |
| 37 | comm::RetCode SimpleProducer::Add(const comm::proto::AddRequest &req, |
| 38 | comm::proto::AddResponse &resp) { |
| 39 | QLVerb("Add topic_id %d store_id %d queue_id %d", |
| 40 | req.topic_id(), req.store_id(), req.queue_id()); |
| 41 | |
| 42 | comm::RetCode ret; |
| 43 | |
| 44 | shared_ptr<const config::StoreConfig> store_config; |
| 45 | if (comm::RetCode::RET_OK != (ret = config::GlobalConfig::GetThreadInstance()-> |
| 46 | GetStoreConfig(req.topic_id(), store_config))) { |
| 47 | QLErr("GetStoreConfig ret %d topic_id %d", as_integer(ret), req.topic_id()); |
| 48 | return ret; |
| 49 | } |
| 50 | |
| 51 | shared_ptr<const config::proto::Store> store; |
| 52 | if (comm::RetCode::RET_OK != (ret = store_config->GetStoreByStoreID(req.store_id(), store))) { |
| 53 | QLErr("GetStoreByStoreID ret %d store_id %d", as_integer(ret), req.store_id()); |
| 54 | return ret; |
| 55 | } |
| 56 | |
| 57 | assert(store->addrs_size() >= 2); |
| 58 | comm::proto::Addr master_addr = store->addrs(1); |
| 59 | |
| 60 | if (req.master_addr().ip() == master_addr.ip() && |
| 61 | req.master_addr().port() == master_addr.port() && |
| 62 | req.master_addr().paxos_port() == master_addr.paxos_port()) { |
| 63 | |
| 64 | resp.set_cursor_id(1); |
| 65 | return comm::RetCode::RET_OK; |
| 66 | } |
| 67 | resp.set_cursor_id(-1); |
| 68 | resp.mutable_redirect_addr()->CopyFrom(master_addr); |
| 69 | return comm::RetCode::RET_ERR_NOT_MASTER; |
| 70 | } |
| 71 | |
| 72 | |
| 73 | } // namespace test |
nothing calls this directly
no test coverage detected