| 53 | } |
| 54 | |
| 55 | comm::RetCode QueueSelectorDefault::GetQueueID(int &queue_id) { |
| 56 | comm::RetCode ret; |
| 57 | |
| 58 | std::shared_ptr<const config::TopicConfig> topic_config; |
| 59 | if (comm::RetCode::RET_OK != (ret = config::GlobalConfig::GetThreadInstance()->GetTopicConfigByTopicID(impl_->topic_id, topic_config))) { |
| 60 | QLErr("GetTopicConfigByTopicID ret %d", as_integer(ret)); |
| 61 | return ret; |
| 62 | } |
| 63 | |
| 64 | int queue_info_id; |
| 65 | if (comm::RetCode::RET_OK != (ret = topic_config->GetQueueInfoIDByCount(impl_->pub_id, impl_->count, queue_info_id))) { |
| 66 | QLErr("GetQueueInfoIDByCount ret %d", as_integer(ret)); |
| 67 | return ret; |
| 68 | } |
| 69 | |
| 70 | shared_ptr<const config::proto::QueueInfo> queue_info; |
| 71 | if (comm::RetCode::RET_OK != (ret = topic_config->GetQueueInfoByQueueInfoID(queue_info_id, queue_info))) { |
| 72 | QLErr("GetQueueInfoByQueueInfoID ret %d", as_integer(ret)); |
| 73 | return ret; |
| 74 | } |
| 75 | |
| 76 | size_t rd = crc32(0, Z_NULL, 0); |
| 77 | { |
| 78 | uint64_t tmp_uin = (!impl_->uin || queue_info->handle_by_random_uin()) ? comm::utils::OtherUtils::FastRand() : impl_->uin; |
| 79 | rd = crc32(rd, (const unsigned char *)&tmp_uin, sizeof(uint64_t)); |
| 80 | rd = crc32(rd, (const unsigned char *)&impl_->topic_id, sizeof(int)); |
| 81 | if (impl_->retry_switch_queue) rd += impl_->retry; |
| 82 | } |
| 83 | |
| 84 | if (0 > comm::as_integer(ret = topic_config->GetQueueByLoopRank(queue_info_id, rd, queue_id))) { |
| 85 | QLErr("GetQueueByLoopRank ret %d queue_info_id %d rd %d", ret, queue_info_id, rd); |
| 86 | } |
| 87 | |
| 88 | QLVerb("pub_id %d count %d queue_info_id %d rd %zu queue_id %d retry %d", impl_->pub_id, impl_->count, queue_info_id, rd, queue_id, impl_->retry); |
| 89 | |
| 90 | ++impl_->retry; |
| 91 | |
| 92 | return comm::RetCode::RET_OK; |
| 93 | } |
| 94 | |
| 95 | class StoreSelectorDefault::StoreSelectorDefaultImpl { |
| 96 | public: |
no test coverage detected