| /* |
| * Licensed to the Apache Software Foundation (ASF) under one or more |
| * contributor license agreements. See the NOTICE file distributed with |
| * this work for additional information regarding copyright ownership. |
| * The ASF licenses this file to You under the Apache License, Version 2.0 |
| * (the "License"); you may not use this file except in compliance with |
| * the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, software |
| * distributed under the License is distributed on an "AS IS" BASIS, |
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| * See the License for the specific language governing permissions and |
| * limitations under the License. |
| */ |
| #include "DefaultMQPushConsumerImpl.h" |
| |
| #ifndef WIN32 |
| #include <signal.h> |
| #endif |
| |
| #include "CommunicationMode.h" |
| #include "ConsumeMsgService.h" |
| #include "protocol/body/ConsumerRunningInfo.h" |
| #include "FilterAPI.hpp" |
| #include "Logging.h" |
| #include "MQAdminImpl.h" |
| #include "MQClientAPIImpl.h" |
| #include "MQClientInstance.h" |
| #include "MQClientManager.h" |
| #include "MQProtos.h" |
| #include "NamespaceUtil.h" |
| #include "LocalFileOffsetStore.h" |
| #include "PullAPIWrapper.h" |
| #include "PullMessageService.hpp" |
| #include "PullSysFlag.h" |
| #include "RebalancePushImpl.h" |
| #include "RemoteBrokerOffsetStore.h" |
| #include "SocketUtil.h" |
| #include "UtilAll.h" |
| #include "Validators.h" |
| |
| static const long BROKER_SUSPEND_MAX_TIME_MILLIS = 1000 * 15; |
| static const long CONSUMER_TIMEOUT_MILLIS_WHEN_SUSPEND = 1000 * 30; |
| |
| namespace rocketmq { |
| |
| class DefaultMQPushConsumerImpl::AsyncPullCallback : public AutoDeletePullCallback { |
| public: |
| AsyncPullCallback(DefaultMQPushConsumerImplPtr pushConsumer, |
| PullRequestPtr request, |
| SubscriptionData* subscriptionData) |
| : default_mq_push_consumer_(pushConsumer), pull_request_(request), subscription_data_(subscriptionData) {} |
| |
| ~AsyncPullCallback() = default; |
| |
| void onSuccess(std::unique_ptr<PullResult> pull_result) override { |
| auto consumer = default_mq_push_consumer_.lock(); |
| if (nullptr == consumer) { |
| LOG_WARN_NEW("AsyncPullCallback::onSuccess: DefaultMQPushConsumerImpl is released."); |
| return; |
| } |
| |
| pull_result = consumer->pull_api_wrapper_->processPullResult(pull_request_->message_queue(), std::move(pull_result), |
| subscription_data_); |
| switch (pull_result->pull_status()) { |
| case FOUND: { |
| int64_t prev_request_offset = pull_request_->next_offset(); |
| pull_request_->set_next_offset(pull_result->next_begin_offset()); |
| |
| int64_t first_msg_offset = (std::numeric_limits<int64_t>::max)(); |
| if (!pull_result->msg_found_list().empty()) { |
| first_msg_offset = pull_result->msg_found_list()[0]->queue_offset(); |
| |
| pull_request_->process_queue()->putMessage(pull_result->msg_found_list()); |
| consumer->consume_service_->submitConsumeRequest( |
| pull_result->msg_found_list(), pull_request_->process_queue(), pull_request_->message_queue(), true); |
| } |
| |
| consumer->executePullRequestImmediately(pull_request_); |
| |
| if (pull_result->next_begin_offset() < prev_request_offset || first_msg_offset < prev_request_offset) { |
| LOG_WARN_NEW( |
| "[BUG] pull message result maybe data wrong, nextBeginOffset:{} firstMsgOffset:{} prevRequestOffset:{}", |
| pull_result->next_begin_offset(), first_msg_offset, prev_request_offset); |
| } |
| } break; |
| case NO_NEW_MSG: |
| case NO_MATCHED_MSG: |
| pull_request_->set_next_offset(pull_result->next_begin_offset()); |
| consumer->correctTagsOffset(pull_request_); |
| consumer->executePullRequestImmediately(pull_request_); |
| break; |
| case NO_LATEST_MSG: |
| pull_request_->set_next_offset(pull_result->next_begin_offset()); |
| consumer->correctTagsOffset(pull_request_); |
| consumer->executePullRequestLater( |
| pull_request_, consumer->getDefaultMQPushConsumerConfig()->pull_time_delay_millis_when_exception()); |
| break; |
| case OFFSET_ILLEGAL: { |
| LOG_WARN_NEW("the pull request offset illegal, {} {}", pull_request_->toString(), pull_result->toString()); |
| |
| pull_request_->set_next_offset(pull_result->next_begin_offset()); |
| pull_request_->process_queue()->set_dropped(true); |
| |
| // update and persist offset, then removeProcessQueue |
| auto pull_request = pull_request_; |
| consumer->executeTaskLater( |
| [consumer, pull_request]() { |
| try { |
| consumer->getOffsetStore()->updateOffset(pull_request->message_queue(), pull_request->next_offset(), |
| false); |
| consumer->getOffsetStore()->persist(pull_request->message_queue()); |
| consumer->getRebalanceImpl()->removeProcessQueue(pull_request->message_queue()); |
| |
| LOG_WARN_NEW("fix the pull request offset, {}", pull_request->toString()); |
| } catch (std::exception& e) { |
| LOG_ERROR_NEW("executeTaskLater Exception: {}", e.what()); |
| } |
| }, |
| 10000); |
| } break; |
| default: |
| break; |
| } |
| } |
| |
| void onException(MQException& e) noexcept override { |
| auto consumer = default_mq_push_consumer_.lock(); |
| if (nullptr == consumer) { |
| LOG_WARN_NEW("AsyncPullCallback::onException: DefaultMQPushConsumerImpl is released."); |
| return; |
| } |
| |
| if (!UtilAll::isRetryTopic(pull_request_->message_queue().topic())) { |
| LOG_WARN_NEW("execute the pull request exception: {}", e.what()); |
| } |
| |
| // TODO |
| consumer->executePullRequestLater( |
| pull_request_, consumer->getDefaultMQPushConsumerConfig()->pull_time_delay_millis_when_exception()); |
| } |
| |
| private: |
| std::weak_ptr<DefaultMQPushConsumerImpl> default_mq_push_consumer_; |
| PullRequestPtr pull_request_; |
| SubscriptionData* subscription_data_; |
| }; |
| |
| DefaultMQPushConsumerImpl::DefaultMQPushConsumerImpl(DefaultMQPushConsumerConfigPtr config) |
| : DefaultMQPushConsumerImpl(config, nullptr) {} |
| |
| DefaultMQPushConsumerImpl::DefaultMQPushConsumerImpl(DefaultMQPushConsumerConfigPtr config, RPCHookPtr rpcHook) |
| : MQClientImpl(config, rpcHook), |
| start_time_(UtilAll::currentTimeMillis()), |
| pause_(false), |
| consume_orderly_(false), |
| message_listener_(nullptr), |
| consume_service_(nullptr), |
| rebalance_impl_(new RebalancePushImpl(this)), |
| pull_api_wrapper_(nullptr), |
| offset_store_(nullptr) {} |
| |
| DefaultMQPushConsumerImpl::~DefaultMQPushConsumerImpl() = default; |
| |
| void DefaultMQPushConsumerImpl::start() { |
| #ifndef WIN32 |
| /* Ignore the SIGPIPE */ |
| struct sigaction sa; |
| memset(&sa, 0, sizeof(struct sigaction)); |
| sa.sa_handler = SIG_IGN; |
| sa.sa_flags = 0; |
| ::sigaction(SIGPIPE, &sa, 0); |
| #endif |
| |
| switch (service_state_) { |
| case CREATE_JUST: { |
| // wrap namespace |
| client_config_->set_group_name( |
| NamespaceUtil::wrapNamespace(client_config_->name_space(), client_config_->group_name())); |
| |
| LOG_INFO_NEW("the consumer [{}] start beginning.", client_config_->group_name()); |
| |
| service_state_ = START_FAILED; |
| |
| checkConfig(); |
| |
| copySubscription(); |
| |
| if (messageModel() == MessageModel::CLUSTERING) { |
| client_config_->changeInstanceNameToPID(); |
| } |
| |
| // init client_instance_ |
| MQClientImpl::start(); |
| |
| // init rebalance_impl_ |
| rebalance_impl_->set_consumer_group(client_config_->group_name()); |
| rebalance_impl_->set_message_model(getDefaultMQPushConsumerConfig()->message_model()); |
| rebalance_impl_->set_allocate_mq_strategy(getDefaultMQPushConsumerConfig()->allocate_mq_strategy()); |
| rebalance_impl_->set_client_instance(client_instance_.get()); |
| |
| // init pull_api_wrapper_ |
| pull_api_wrapper_.reset(new PullAPIWrapper(client_instance_.get(), client_config_->group_name())); |
| // TODO: registerFilterMessageHook |
| |
| // init offset_store_ |
| switch (getDefaultMQPushConsumerConfig()->message_model()) { |
| case MessageModel::BROADCASTING: |
| offset_store_.reset(new LocalFileOffsetStore(client_instance_.get(), client_config_->group_name())); |
| break; |
| case MessageModel::CLUSTERING: |
| offset_store_.reset(new RemoteBrokerOffsetStore(client_instance_.get(), client_config_->group_name())); |
| break; |
| } |
| offset_store_->load(); |
| |
| // checkConfig() guarantee message_listener_ is not nullptr |
| if (message_listener_->getMessageListenerType() == messageListenerOrderly) { |
| LOG_INFO_NEW("start orderly consume service: {}", client_config_->group_name()); |
| consume_orderly_ = true; |
| consume_service_.reset(new ConsumeMessageOrderlyService( |
| this, getDefaultMQPushConsumerConfig()->consume_thread_nums(), message_listener_)); |
| } else { |
| // for backward compatible, defaultly and concurrently listeners are allocating |
| // ConsumeMessageConcurrentlyService |
| LOG_INFO_NEW("start concurrently consume service: {}", client_config_->group_name()); |
| consume_orderly_ = false; |
| consume_service_.reset(new ConsumeMessageConcurrentlyService( |
| this, getDefaultMQPushConsumerConfig()->consume_thread_nums(), message_listener_)); |
| } |
| consume_service_->start(); |
| |
| // register consumer |
| bool registerOK = client_instance_->registerConsumer(client_config_->group_name(), this); |
| if (!registerOK) { |
| service_state_ = CREATE_JUST; |
| consume_service_->shutdown(); |
| THROW_MQEXCEPTION(MQClientException, "The cousumer group[" + client_config_->group_name() + |
| "] has been created before, specify another name please.", |
| -1); |
| } |
| |
| client_instance_->start(); |
| |
| LOG_INFO_NEW("the consumer [{}] start OK", client_config_->group_name()); |
| service_state_ = RUNNING; |
| break; |
| } |
| case RUNNING: |
| case START_FAILED: |
| case SHUTDOWN_ALREADY: |
| THROW_MQEXCEPTION(MQClientException, "The PushConsumer service state not OK, maybe started once", -1); |
| break; |
| default: |
| break; |
| } |
| |
| updateTopicSubscribeInfoWhenSubscriptionChanged(); |
| client_instance_->sendHeartbeatToAllBrokerWithLock(); |
| client_instance_->rebalanceImmediately(); |
| } |
| |
| void DefaultMQPushConsumerImpl::checkConfig() { |
| std::string groupname = client_config_->group_name(); |
| |
| // check consumerGroup |
| Validators::checkGroup(groupname); |
| |
| // consumerGroup |
| if (DEFAULT_CONSUMER_GROUP == groupname) { |
| THROW_MQEXCEPTION(MQClientException, |
| "consumerGroup can not equal " + DEFAULT_CONSUMER_GROUP + ", please specify another one.", -1); |
| } |
| |
| if (getDefaultMQPushConsumerConfig()->message_model() != BROADCASTING && |
| getDefaultMQPushConsumerConfig()->message_model() != CLUSTERING) { |
| THROW_MQEXCEPTION(MQClientException, "messageModel is valid", -1); |
| } |
| |
| // allocateMessageQueueStrategy |
| if (nullptr == getDefaultMQPushConsumerConfig()->allocate_mq_strategy()) { |
| THROW_MQEXCEPTION(MQClientException, "allocateMessageQueueStrategy is null", -1); |
| } |
| |
| // subscription |
| if (subscription_.empty()) { |
| THROW_MQEXCEPTION(MQClientException, "subscription is empty", -1); |
| } |
| |
| // messageListener |
| if (message_listener_ == nullptr) { |
| THROW_MQEXCEPTION(MQClientException, "messageListener is null", -1); |
| } |
| } |
| |
| void DefaultMQPushConsumerImpl::copySubscription() { |
| for (const auto& it : subscription_) { |
| LOG_INFO_NEW("buildSubscriptionData: {}, {}", it.first, it.second); |
| rebalance_impl_->setSubscriptionData(it.first, FilterAPI::buildSubscriptionData(it.first, it.second)); |
| } |
| |
| switch (getDefaultMQPushConsumerConfig()->message_model()) { |
| case BROADCASTING: |
| break; |
| case CLUSTERING: { |
| // auto subscript retry topic |
| std::string retryTopic = UtilAll::getRetryTopic(client_config_->group_name()); |
| rebalance_impl_->setSubscriptionData(retryTopic, FilterAPI::buildSubscriptionData(retryTopic, SUB_ALL)); |
| break; |
| } |
| default: |
| break; |
| } |
| } |
| |
| void DefaultMQPushConsumerImpl::updateTopicSubscribeInfoWhenSubscriptionChanged() { |
| auto& subTable = rebalance_impl_->getSubscriptionInner(); |
| for (const auto& it : subTable) { |
| const auto& topic = it.first; |
| auto topic_route_data = client_instance_->getTopicRouteData(topic); |
| if (topic_route_data != nullptr) { |
| std::vector<MQMessageQueue> subscribeInfo = |
| MQClientInstance::topicRouteData2TopicSubscribeInfo(topic, topic_route_data); |
| updateTopicSubscribeInfo(topic, subscribeInfo); |
| } else { |
| bool ret = client_instance_->updateTopicRouteInfoFromNameServer(topic); |
| if (!ret) { |
| LOG_WARN_NEW("The topic[{}] not exist, or its route data not changed", topic); |
| } |
| } |
| } |
| } |
| |
| void DefaultMQPushConsumerImpl::shutdown() { |
| switch (service_state_) { |
| case RUNNING: { |
| consume_service_->shutdown(); |
| persistConsumerOffset(); |
| client_instance_->unregisterConsumer(client_config_->group_name()); |
| client_instance_->shutdown(); |
| rebalance_impl_->destroy(); |
| service_state_ = SHUTDOWN_ALREADY; |
| LOG_INFO_NEW("the consumer [{}] shutdown OK", client_config_->group_name()); |
| break; |
| } |
| case CREATE_JUST: |
| case SHUTDOWN_ALREADY: |
| break; |
| default: |
| break; |
| } |
| } |
| |
| void DefaultMQPushConsumerImpl::suspend() { |
| pause_ = true; |
| LOG_INFO_NEW("suspend this consumer, {}", client_config_->group_name()); |
| } |
| |
| void DefaultMQPushConsumerImpl::resume() { |
| pause_ = false; |
| doRebalance(); |
| LOG_INFO_NEW("resume this consumer, {}", client_config_->group_name()); |
| } |
| |
| MQMessageListener* DefaultMQPushConsumerImpl::getMessageListener() const { |
| return message_listener_; |
| } |
| |
| void DefaultMQPushConsumerImpl::registerMessageListener(MessageListenerConcurrently* message_listener) { |
| if (nullptr != message_listener) { |
| message_listener_ = message_listener; |
| } |
| } |
| |
| void DefaultMQPushConsumerImpl::registerMessageListener(MessageListenerOrderly* message_listener) { |
| if (nullptr != message_listener) { |
| message_listener_ = message_listener; |
| } |
| } |
| |
| void DefaultMQPushConsumerImpl::subscribe(const std::string& topic, const std::string& subExpression) { |
| // TODO: change substation after start |
| subscription_[topic] = subExpression; |
| } |
| |
| std::vector<SubscriptionData> DefaultMQPushConsumerImpl::subscriptions() const { |
| std::vector<SubscriptionData> result; |
| auto& subTable = rebalance_impl_->getSubscriptionInner(); |
| for (const auto& it : subTable) { |
| result.push_back(*(it.second)); |
| } |
| return result; |
| } |
| |
| void DefaultMQPushConsumerImpl::updateTopicSubscribeInfo(const std::string& topic, std::vector<MQMessageQueue>& info) { |
| rebalance_impl_->setTopicSubscribeInfo(topic, info); |
| } |
| |
| void DefaultMQPushConsumerImpl::doRebalance() { |
| if (!pause_) { |
| rebalance_impl_->doRebalance(consume_orderly()); |
| } |
| } |
| |
| void DefaultMQPushConsumerImpl::executePullRequestLater(PullRequestPtr pullRequest, long timeDelay) { |
| client_instance_->getPullMessageService()->executePullRequestLater(pullRequest, timeDelay); |
| } |
| |
| void DefaultMQPushConsumerImpl::executePullRequestImmediately(PullRequestPtr pullRequest) { |
| client_instance_->getPullMessageService()->executePullRequestImmediately(pullRequest); |
| } |
| |
| void DefaultMQPushConsumerImpl::pullMessage(PullRequestPtr pull_request) { |
| if (nullptr == pull_request) { |
| LOG_ERROR("PullRequest is NULL, return"); |
| return; |
| } |
| |
| auto process_queue = pull_request->process_queue(); |
| if (process_queue->dropped()) { |
| LOG_WARN_NEW("the pull request[{}] is dropped.", pull_request->toString()); |
| return; |
| } |
| |
| process_queue->set_last_pull_timestamp(UtilAll::currentTimeMillis()); |
| |
| int cachedMessageCount = process_queue->getCacheMsgCount(); |
| if (cachedMessageCount > getDefaultMQPushConsumerConfig()->pull_threshold_for_queue()) { |
| // too many message in cache, wait to process |
| executePullRequestLater(pull_request, 1000); |
| return; |
| } |
| |
| if (consume_orderly()) { |
| if (process_queue->locked()) { |
| if (!pull_request->locked_first()) { |
| const auto offset = rebalance_impl_->computePullFromWhere(pull_request->message_queue()); |
| bool brokerBusy = offset < pull_request->next_offset(); |
| LOG_INFO_NEW( |
| "the first time to pull message, so fix offset from broker. pullRequest: {} NewOffset: {} brokerBusy: {}", |
| pull_request->toString(), offset, UtilAll::to_string(brokerBusy)); |
| if (brokerBusy) { |
| LOG_INFO_NEW( |
| "[NOTIFYME] the first time to pull message, but pull request offset larger than broker consume offset. " |
| "pullRequest: {} NewOffset: {}", |
| pull_request->toString(), offset); |
| } |
| |
| pull_request->set_locked_first(true); |
| pull_request->set_next_offset(offset); |
| } |
| } else { |
| executePullRequestLater(pull_request, getDefaultMQPushConsumerConfig()->pull_time_delay_millis_when_exception()); |
| LOG_INFO_NEW("pull message later because not locked in broker, {}", pull_request->toString()); |
| return; |
| } |
| } |
| |
| const auto& message_queue = pull_request->message_queue(); |
| SubscriptionData* subscription_data = rebalance_impl_->getSubscriptionData(message_queue.topic()); |
| if (nullptr == subscription_data) { |
| executePullRequestLater(pull_request, getDefaultMQPushConsumerConfig()->pull_time_delay_millis_when_exception()); |
| LOG_WARN_NEW("find the consumer's subscription failed, {}", pull_request->toString()); |
| return; |
| } |
| |
| bool commitOffsetEnable = false; |
| int64_t commitOffsetValue = 0; |
| if (CLUSTERING == getDefaultMQPushConsumerConfig()->message_model()) { |
| commitOffsetValue = offset_store_->readOffset(message_queue, READ_FROM_MEMORY); |
| if (commitOffsetValue > 0) { |
| commitOffsetEnable = true; |
| } |
| } |
| |
| const auto& subExpression = subscription_data->sub_string(); |
| |
| int sysFlag = PullSysFlag::buildSysFlag(commitOffsetEnable, // commitOffset |
| true, // suspend |
| !subExpression.empty(), // subscription |
| false); // class filter |
| |
| try { |
| std::unique_ptr<AsyncPullCallback> callback( |
| new AsyncPullCallback(shared_from_this(), pull_request, subscription_data)); |
| |
| pull_api_wrapper_->pullKernelImpl(message_queue, // mq |
| subExpression, // subExpression |
| subscription_data->expression_type(), // expressionType |
| subscription_data->sub_version(), // subVersion |
| pull_request->next_offset(), // offset |
| getDefaultMQPushConsumerConfig()->pull_batch_size(), // maxNums |
| sysFlag, // sysFlag |
| commitOffsetValue, // commitOffset |
| BROKER_SUSPEND_MAX_TIME_MILLIS, // brokerSuspendMaxTimeMillis |
| CONSUMER_TIMEOUT_MILLIS_WHEN_SUSPEND, // timeoutMillis |
| CommunicationMode::ASYNC, // communicationMode |
| callback.get()); // pullCallback |
| |
| (void)callback.release(); |
| } catch (MQException& e) { |
| LOG_ERROR_NEW("pullKernelImpl exception: {}", e.what()); |
| executePullRequestLater(pull_request, getDefaultMQPushConsumerConfig()->pull_time_delay_millis_when_exception()); |
| } |
| } |
| |
| void DefaultMQPushConsumerImpl::correctTagsOffset(PullRequestPtr pullRequest) { |
| if (0L == pullRequest->process_queue()->getCacheMsgCount()) { |
| offset_store_->updateOffset(pullRequest->message_queue(), pullRequest->next_offset(), true); |
| } |
| } |
| |
| void DefaultMQPushConsumerImpl::executeTaskLater(const handler_type& task, long timeDelay) { |
| client_instance_->getPullMessageService()->executeTaskLater(task, timeDelay); |
| } |
| |
| void DefaultMQPushConsumerImpl::resetRetryAndNamespace(const std::vector<MessageExtPtr>& msgs) { |
| std::string retry_topic = UtilAll::getRetryTopic(groupName()); |
| for (auto& msg : msgs) { |
| std::string group_topic = msg->getProperty(MQMessageConst::PROPERTY_RETRY_TOPIC); |
| if (!group_topic.empty() && retry_topic == msg->topic()) { |
| msg->set_topic(group_topic); |
| } |
| } |
| |
| const auto& name_space = client_config_->name_space(); |
| if (!name_space.empty()) { |
| for (auto& msg : msgs) { |
| msg->set_topic(NamespaceUtil::withoutNamespace(msg->topic(), name_space)); |
| } |
| } |
| } |
| |
| bool DefaultMQPushConsumerImpl::sendMessageBack(MessageExtPtr msg, int delay_level) { |
| return sendMessageBack(msg, delay_level, null); |
| } |
| |
| bool DefaultMQPushConsumerImpl::sendMessageBack(MessageExtPtr msg, int delay_level, const std::string& brokerName) { |
| try { |
| msg->set_topic(NamespaceUtil::wrapNamespace(client_config_->name_space(), msg->topic())); |
| |
| std::string brokerAddr = |
| brokerName.empty() ? msg->store_host_string() : client_instance_->findBrokerAddressInPublish(brokerName); |
| |
| client_instance_->getMQClientAPIImpl()->consumerSendMessageBack( |
| brokerAddr, msg, getDefaultMQPushConsumerConfig()->group_name(), delay_level, 5000, |
| getDefaultMQPushConsumerConfig()->max_reconsume_times()); |
| return true; |
| } catch (const std::exception& e) { |
| LOG_ERROR_NEW("sendMessageBack exception, group: {}, msg: {}. {}", getDefaultMQPushConsumerConfig()->group_name(), |
| msg->toString(), e.what()); |
| } |
| return false; |
| } |
| |
| void DefaultMQPushConsumerImpl::persistConsumerOffset() { |
| if (isServiceStateOk()) { |
| std::vector<MQMessageQueue> mqs = rebalance_impl_->getAllocatedMQ(); |
| offset_store_->persistAll(mqs); |
| } |
| } |
| |
| void DefaultMQPushConsumerImpl::updateConsumeOffset(const MQMessageQueue& mq, int64_t offset) { |
| if (offset >= 0) { |
| offset_store_->updateOffset(mq, offset, false); |
| } else { |
| LOG_ERROR_NEW("updateConsumeOffset of mq:{} error", mq.toString()); |
| } |
| } |
| |
| std::unique_ptr<ConsumerRunningInfo> DefaultMQPushConsumerImpl::consumerRunningInfo() { |
| std::unique_ptr<ConsumerRunningInfo> info(new ConsumerRunningInfo()); |
| |
| info->setProperty(ConsumerRunningInfo::PROP_CONSUME_ORDERLY, UtilAll::to_string(consume_orderly_)); |
| info->setProperty(ConsumerRunningInfo::PROP_THREADPOOL_CORE_SIZE, |
| UtilAll::to_string(getDefaultMQPushConsumerConfig()->consume_thread_nums())); |
| info->setProperty(ConsumerRunningInfo::PROP_CONSUMER_START_TIMESTAMP, UtilAll::to_string(start_time_)); |
| |
| auto sub_set = subscriptions(); |
| info->setSubscriptionSet(sub_set); |
| |
| auto processQueueTable = rebalance_impl_->getProcessQueueTable(); |
| for (const auto& it : processQueueTable) { |
| const auto& mq = it.first; |
| const auto& pq = it.second; |
| |
| ProcessQueueInfo pq_info; |
| pq_info.setCommitOffset(offset_store_->readOffset(mq, MEMORY_FIRST_THEN_STORE)); |
| pq->fillProcessQueueInfo(pq_info); |
| info->setMqTable(mq, pq_info); |
| } |
| |
| // TODO: ConsumeStatus |
| |
| return info; |
| } |
| |
| const std::string& DefaultMQPushConsumerImpl::groupName() const { |
| return client_config_->group_name(); |
| } |
| |
| MessageModel DefaultMQPushConsumerImpl::messageModel() const { |
| return getDefaultMQPushConsumerConfig()->message_model(); |
| }; |
| |
| ConsumeType DefaultMQPushConsumerImpl::consumeType() const { |
| return CONSUME_PASSIVELY; |
| } |
| |
| ConsumeFromWhere DefaultMQPushConsumerImpl::consumeFromWhere() const { |
| return getDefaultMQPushConsumerConfig()->consume_from_where(); |
| } |
| |
| } // namespace rocketmq |
