| /* |
| * 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 "PushConsumerImpl.h" |
| |
| #include <atomic> |
| #include <cassert> |
| #include <chrono> |
| #include <cstdlib> |
| #include <system_error> |
| |
| #include "apache/rocketmq/v1/definition.pb.h" |
| |
| #include "AsyncReceiveMessageCallback.h" |
| #include "ClientManagerFactory.h" |
| #include "ConsumeFifoMessageService.h" |
| #include "ConsumeStandardMessageService.h" |
| #include "MessageAccessor.h" |
| #include "MixAll.h" |
| #include "ProcessQueueImpl.h" |
| #include "RpcClient.h" |
| #include "Signature.h" |
| #include "rocketmq/MQClientException.h" |
| #include "rocketmq/MessageListener.h" |
| #include "rocketmq/MessageModel.h" |
| |
| ROCKETMQ_NAMESPACE_BEGIN |
| |
| PushConsumerImpl::PushConsumerImpl(absl::string_view group_name) : ClientImpl(group_name) { |
| } |
| |
| PushConsumerImpl::~PushConsumerImpl() { |
| SPDLOG_DEBUG("DefaultMQPushConsumerImpl is destructed"); |
| } |
| |
| void PushConsumerImpl::start() { |
| ClientImpl::start(); |
| |
| State expecting = State::STARTING; |
| if (!state_.compare_exchange_strong(expecting, State::STARTED)) { |
| SPDLOG_ERROR("Unexpected consumer state. Expecting: {}, Actual: {}", State::STARTING, |
| state_.load(std::memory_order_relaxed)); |
| return; |
| } |
| |
| if (!message_listener_) { |
| SPDLOG_ERROR("Required message listener is nullptr"); |
| abort(); |
| return; |
| } |
| |
| client_manager_->addClientObserver(shared_from_this()); |
| |
| fetchRoutes(); |
| |
| if (message_listener_->listenerType() == MessageListenerType::FIFO) { |
| SPDLOG_INFO("start orderly consume service: {}", group_name_); |
| consume_message_service_ = |
| std::make_shared<ConsumeFifoMessageService>(shared_from_this(), consume_thread_pool_size_, message_listener_); |
| consume_batch_size_ = 1; |
| } else { |
| // For backward compatibility, by default, ConsumeMessageConcurrentlyService is assumed. |
| SPDLOG_INFO("start concurrently consume service: {}", group_name_); |
| consume_message_service_ = std::make_shared<ConsumeStandardMessageService>( |
| shared_from_this(), consume_thread_pool_size_, message_listener_); |
| } |
| consume_message_service_->start(); |
| |
| { |
| // Set consumer throttling |
| absl::MutexLock lock(&throttle_table_mtx_); |
| for (const auto& item : throttle_table_) { |
| consume_message_service_->throttle(item.first, item.second); |
| } |
| } |
| |
| // Heartbeat depends on initialization of consume-message-service |
| heartbeat(); |
| |
| std::weak_ptr<PushConsumerImpl> consumer_weak_ptr(shared_from_this()); |
| auto scan_assignment_functor = [consumer_weak_ptr]() { |
| std::shared_ptr<PushConsumerImpl> consumer = consumer_weak_ptr.lock(); |
| if (consumer) { |
| consumer->scanAssignments(); |
| } |
| }; |
| |
| scan_assignment_handle_ = client_manager_->getScheduler().schedule( |
| scan_assignment_functor, SCAN_ASSIGNMENT_TASK_NAME, std::chrono::milliseconds(100), std::chrono::seconds(5)); |
| |
| SPDLOG_INFO("PushConsumer started, groupName={}", group_name_); |
| } |
| |
| const char* PushConsumerImpl::SCAN_ASSIGNMENT_TASK_NAME = "scan-assignment-task"; |
| |
| void PushConsumerImpl::shutdown() { |
| State expecting = State::STARTED; |
| if (state_.compare_exchange_strong(expecting, State::STOPPING)) { |
| if (scan_assignment_handle_) { |
| client_manager_->getScheduler().cancel(scan_assignment_handle_); |
| SPDLOG_DEBUG("Scan assignment periodic task cancelled"); |
| } |
| |
| { |
| absl::MutexLock lock(&process_queue_table_mtx_); |
| process_queue_table_.clear(); |
| } |
| |
| if (consume_message_service_) { |
| consume_message_service_->shutdown(); |
| } |
| |
| // Shutdown services started by parent |
| ClientImpl::shutdown(); |
| |
| SPDLOG_INFO("PushConsumerImpl stopped"); |
| } else { |
| SPDLOG_ERROR("Shutdown with unexpected state. Expecting: {}, Actual: {}", State::STARTED, |
| state_.load(std::memory_order_relaxed)); |
| } |
| } |
| |
| void PushConsumerImpl::subscribe(const std::string& topic, const std::string& expression, |
| ExpressionType expression_type) { |
| absl::MutexLock lock(&topic_filter_expression_table_mtx_); |
| FilterExpression filter_expression{expression, expression_type}; |
| topic_filter_expression_table_.emplace(topic, filter_expression); |
| } |
| |
| void PushConsumerImpl::unsubscribe(const std::string& topic) { |
| absl::MutexLock lock(&topic_filter_expression_table_mtx_); |
| topic_filter_expression_table_.erase(topic); |
| } |
| |
| absl::optional<FilterExpression> PushConsumerImpl::getFilterExpression(const std::string& topic) const { |
| { |
| absl::MutexLock lock(&topic_filter_expression_table_mtx_); |
| if (topic_filter_expression_table_.contains(topic)) { |
| return absl::make_optional(topic_filter_expression_table_.at(topic)); |
| } else { |
| return absl::optional<FilterExpression>(); |
| } |
| } |
| } |
| |
| void PushConsumerImpl::setConsumeFromWhere(ConsumeFromWhere consume_from_where) { |
| consume_from_where_ = consume_from_where; |
| } |
| |
| void PushConsumerImpl::scanAssignments() { |
| SPDLOG_DEBUG("Start of assignment scanning"); |
| if (!active()) { |
| SPDLOG_INFO("Client has stopped. Abort scanning immediately."); |
| return; |
| } |
| |
| { |
| absl::MutexLock lk(&topic_filter_expression_table_mtx_); |
| for (auto& entry : topic_filter_expression_table_) { |
| std::string topic = entry.first; |
| const auto& filter_expression = entry.second; |
| SPDLOG_DEBUG("Scan assignments for {}", topic); |
| auto callback = [this, topic, filter_expression](const std::error_code& ec, |
| const TopicAssignmentPtr& assignments) { |
| if (ec) { |
| SPDLOG_WARN("Failed to acquire assignments for topic={} from load balancer. Cause: {}", topic, ec.message()); |
| } else if (assignments && !assignments->assignmentList().empty()) { |
| syncProcessQueue(topic, assignments, filter_expression); |
| } |
| }; |
| queryAssignment(topic, callback); |
| } // end of for-loop |
| } |
| SPDLOG_DEBUG("End of assignment scanning."); |
| } |
| |
| bool PushConsumerImpl::selectBroker(const TopicRouteDataPtr& topic_route_data, std::string& broker_host) { |
| if (topic_route_data && !topic_route_data->partitions().empty()) { |
| uint32_t index = TopicAssignment::getAndIncreaseQueryWhichBroker(); |
| for (uint32_t i = index; i < index + topic_route_data->partitions().size(); i++) { |
| auto partition = topic_route_data->partitions().at(i % topic_route_data->partitions().size()); |
| if (MixAll::MASTER_BROKER_ID != partition.broker().id() || Permission::NONE == partition.permission()) { |
| continue; |
| } |
| |
| if (partition.broker()) { |
| broker_host = partition.broker().serviceAddress(); |
| return true; |
| } |
| } |
| } |
| return false; |
| } |
| |
| void PushConsumerImpl::wrapQueryAssignmentRequest(const std::string& topic, const std::string& consumer_group, |
| const std::string& client_id, const std::string& strategy_name, |
| QueryAssignmentRequest& request) { |
| request.mutable_topic()->set_name(topic); |
| request.mutable_topic()->set_resource_namespace(resourceNamespace()); |
| request.mutable_group()->set_name(consumer_group); |
| request.mutable_group()->set_resource_namespace(resourceNamespace()); |
| request.set_client_id(client_id); |
| } |
| |
| void PushConsumerImpl::queryAssignment( |
| const std::string& topic, const std::function<void(const std::error_code&, const TopicAssignmentPtr&)>& cb) { |
| |
| auto callback = [this, topic, cb](const std::error_code& ec, const TopicRouteDataPtr& topic_route) { |
| TopicAssignmentPtr topic_assignment; |
| if (MessageModel::BROADCASTING == message_model_) { |
| if (ec) { |
| SPDLOG_WARN("Failed to get valid route entries for topic={}. Cause: {}", topic, ec.message()); |
| cb(ec, topic_assignment); |
| } |
| |
| std::vector<Assignment> assignments; |
| assignments.reserve(topic_route->partitions().size()); |
| for (const auto& partition : topic_route->partitions()) { |
| assignments.emplace_back(Assignment(partition.asMessageQueue())); |
| } |
| topic_assignment = std::make_shared<TopicAssignment>(std::move(assignments)); |
| cb(ec, topic_assignment); |
| return; |
| } |
| |
| std::string broker_host; |
| if (!selectBroker(topic_route, broker_host)) { |
| SPDLOG_WARN("Failed to select a broker to query assignment for group={}, topic={}", group_name_, topic); |
| } |
| |
| QueryAssignmentRequest request; |
| setAccessPoint(request.mutable_endpoints()); |
| wrapQueryAssignmentRequest(topic, group_name_, clientId(), MixAll::DEFAULT_LOAD_BALANCER_STRATEGY_NAME_, request); |
| SPDLOG_DEBUG("QueryAssignmentRequest: {}", request.DebugString()); |
| |
| absl::flat_hash_map<std::string, std::string> metadata; |
| Signature::sign(this, metadata); |
| auto assignment_callback = [this, cb, topic, broker_host](const std::error_code& ec, |
| const QueryAssignmentResponse& response) { |
| if (ec) { |
| SPDLOG_WARN("Failed to acquire queue assignment of topic={} from brokerAddress={}", topic, broker_host); |
| cb(ec, nullptr); |
| } else { |
| SPDLOG_DEBUG("Query topic assignment OK. Topic={}, group={}, assignment-size={}", topic, group_name_, |
| response.assignments().size()); |
| SPDLOG_TRACE("Query assignment response for {} is: {}", topic, response.DebugString()); |
| cb(ec, std::make_shared<TopicAssignment>(response)); |
| } |
| }; |
| |
| client_manager_->queryAssignment(broker_host, metadata, request, absl::ToChronoMilliseconds(io_timeout_), |
| assignment_callback); |
| }; |
| getRouteFor(topic, callback); |
| } |
| |
| /** |
| * |
| * @param topic Topic to process |
| * @param assignment Latest assignment from load balancer |
| * @param filter_expression Filter expression |
| */ |
| void PushConsumerImpl::syncProcessQueue(const std::string& topic, |
| const std::shared_ptr<TopicAssignment>& topic_assignment, |
| const FilterExpression& filter_expression) { |
| const std::vector<Assignment>& assignment_list = topic_assignment->assignmentList(); |
| std::vector<MQMessageQueue> message_queue_list; |
| message_queue_list.reserve(assignment_list.size()); |
| for (const auto& assignment : assignment_list) { |
| message_queue_list.push_back(assignment.messageQueue()); |
| } |
| |
| std::vector<MQMessageQueue> current; |
| { |
| absl::MutexLock lock(&process_queue_table_mtx_); |
| for (auto it = process_queue_table_.begin(); it != process_queue_table_.end();) { |
| if (topic != it->first.getTopic()) { |
| it++; |
| continue; |
| } |
| |
| if (std::none_of(message_queue_list.cbegin(), message_queue_list.cend(), |
| [&](const MQMessageQueue& message_queue) { return it->first == message_queue; })) { |
| SPDLOG_INFO("Stop receiving messages from {} as it is not assigned to current client according to latest " |
| "assignment result from load balancer", |
| it->first.simpleName()); |
| process_queue_table_.erase(it++); |
| } else { |
| if (!it->second || it->second->expired()) { |
| SPDLOG_WARN("ProcessQueue={} is expired. Remove it for now.", it->first.simpleName()); |
| process_queue_table_.erase(it++); |
| continue; |
| } |
| current.push_back(it->first); |
| it++; |
| } |
| } |
| } |
| |
| for (const auto& message_queue : message_queue_list) { |
| if (std::none_of(current.cbegin(), current.cend(), |
| [&](const MQMessageQueue& item) { return item == message_queue; })) { |
| SPDLOG_INFO("Start to receive message from {} according to latest assignment info from load balancer", |
| message_queue.simpleName()); |
| if (!receiveMessage(message_queue, filter_expression)) { |
| if (!active()) { |
| SPDLOG_WARN("Failed to initiate receive message request-response-cycle for {}", message_queue.simpleName()); |
| // TODO: remove it from current assignment such that a second attempt will be made again in the next round. |
| } |
| } |
| } |
| } |
| } |
| |
| ProcessQueueSharedPtr PushConsumerImpl::getOrCreateProcessQueue(const MQMessageQueue& message_queue, |
| const FilterExpression& filter_expression) { |
| ProcessQueueSharedPtr process_queue; |
| { |
| absl::MutexLock lock(&process_queue_table_mtx_); |
| if (!active()) { |
| SPDLOG_INFO("PushConsumer has stopped. Drop creation of ProcessQueue"); |
| return process_queue; |
| } |
| |
| if (process_queue_table_.contains(message_queue)) { |
| process_queue = process_queue_table_.at(message_queue); |
| } else { |
| SPDLOG_INFO("Create ProcessQueue for message queue[{}]", message_queue.simpleName()); |
| // create ProcessQueue |
| process_queue = |
| std::make_shared<ProcessQueueImpl>(message_queue, filter_expression, shared_from_this(), client_manager_); |
| std::shared_ptr<AsyncReceiveMessageCallback> receive_callback = |
| std::make_shared<AsyncReceiveMessageCallback>(process_queue); |
| process_queue->callback(receive_callback); |
| process_queue_table_.emplace(std::make_pair(message_queue, process_queue)); |
| } |
| } |
| return process_queue; |
| } |
| |
| bool PushConsumerImpl::receiveMessage(const MQMessageQueue& message_queue, const FilterExpression& filter_expression) { |
| if (!active()) { |
| SPDLOG_INFO("PushConsumer has stopped. Drop further receive message request"); |
| return false; |
| } |
| |
| ProcessQueueSharedPtr process_queue_ptr = getOrCreateProcessQueue(message_queue, filter_expression); |
| if (!process_queue_ptr) { |
| SPDLOG_INFO("Consumer has stopped. Stop creating processQueue"); |
| return false; |
| } |
| |
| const std::string& broker_host = message_queue.serviceAddress(); |
| if (broker_host.empty()) { |
| SPDLOG_ERROR("Failed to resolve address for brokerName={}", message_queue.getBrokerName()); |
| return false; |
| } |
| |
| switch (message_model_) { |
| case MessageModel::BROADCASTING: { |
| int64_t offset = -1; |
| if (!offset_store_ || !offset_store_->readOffset(message_queue, offset)) { |
| // Query latest offset from server. |
| QueryOffsetRequest request; |
| request.mutable_partition()->mutable_topic()->set_resource_namespace(resource_namespace_); |
| request.mutable_partition()->mutable_topic()->set_name(message_queue.getTopic()); |
| request.mutable_partition()->set_id(message_queue.getQueueId()); |
| request.mutable_partition()->mutable_broker()->set_name(message_queue.getBrokerName()); |
| request.set_policy(rmq::QueryOffsetPolicy::END); |
| absl::flat_hash_map<std::string, std::string> metadata; |
| Signature::sign(this, metadata); |
| auto callback = [broker_host, message_queue, process_queue_ptr](const std::error_code& ec, |
| const QueryOffsetResponse& response) { |
| if (ec) { |
| SPDLOG_WARN("Failed to acquire latest offset for partition[{}] from server[host={}]. Cause: {}", |
| message_queue.simpleName(), broker_host, ec.message()); |
| } else { |
| assert(response.offset() >= 0); |
| process_queue_ptr->nextOffset(response.offset()); |
| process_queue_ptr->receiveMessage(); |
| } |
| }; |
| client_manager_->queryOffset(broker_host, metadata, request, absl::ToChronoMilliseconds(io_timeout_), callback); |
| } |
| break; |
| } |
| case MessageModel::CLUSTERING: |
| process_queue_ptr->receiveMessage(); |
| break; |
| } |
| return true; |
| } |
| |
| std::shared_ptr<ConsumeMessageService> PushConsumerImpl::getConsumeMessageService() { |
| return consume_message_service_; |
| } |
| |
| void PushConsumerImpl::ack(const MQMessageExt& msg, const std::function<void(const std::error_code&)>& callback) { |
| const std::string& target_host = MessageAccessor::targetEndpoint(msg); |
| assert(!target_host.empty()); |
| SPDLOG_DEBUG("Prepare to send ack to broker. BrokerAddress={}, topic={}, queueId={}, msgId={}", target_host, |
| msg.getTopic(), msg.getQueueId(), msg.getMsgId()); |
| AckMessageRequest request; |
| wrapAckMessageRequest(msg, request); |
| absl::flat_hash_map<std::string, std::string> metadata; |
| Signature::sign(this, metadata); |
| client_manager_->ack(target_host, metadata, request, absl::ToChronoMilliseconds(io_timeout_), callback); |
| } |
| |
| void PushConsumerImpl::nack(const MQMessageExt& msg, const std::function<void(const std::error_code&)>& callback) { |
| std::string target_host = MessageAccessor::targetEndpoint(msg); |
| |
| absl::flat_hash_map<std::string, std::string> metadata; |
| Signature::sign(this, metadata); |
| |
| rmq::NackMessageRequest request; |
| |
| // Group |
| request.mutable_group()->set_resource_namespace(resource_namespace_); |
| request.mutable_group()->set_name(group_name_); |
| // Topic |
| request.mutable_topic()->set_resource_namespace(resource_namespace_); |
| request.mutable_topic()->set_name(msg.getTopic()); |
| request.set_client_id(clientId()); |
| request.set_receipt_handle(msg.receiptHandle()); |
| request.set_message_id(msg.getMsgId()); |
| request.set_delivery_attempt(msg.getDeliveryAttempt() + 1); |
| request.set_max_delivery_attempts(max_delivery_attempts_); |
| |
| client_manager_->nack(target_host, metadata, request, absl::ToChronoMilliseconds(io_timeout_), callback); |
| SPDLOG_DEBUG("Send message nack to broker server[host={}]", target_host); |
| } |
| |
| void PushConsumerImpl::forwardToDeadLetterQueue(const MQMessageExt& message, const std::function<void(bool)>& cb) { |
| std::string target_host = MessageAccessor::targetEndpoint(message); |
| |
| absl::flat_hash_map<std::string, std::string> metadata; |
| Signature::sign(this, metadata); |
| |
| ForwardMessageToDeadLetterQueueRequest request; |
| request.mutable_group()->set_resource_namespace(resource_namespace_); |
| request.mutable_group()->set_name(group_name_); |
| |
| request.mutable_topic()->set_resource_namespace(resource_namespace_); |
| request.mutable_topic()->set_name(message.getTopic()); |
| |
| request.set_client_id(clientId()); |
| request.set_message_id(message.getMsgId()); |
| |
| request.set_delivery_attempt(message.getDeliveryAttempt()); |
| request.set_max_delivery_attempts(max_delivery_attempts_); |
| |
| client_manager_->forwardMessageToDeadLetterQueue(target_host, metadata, request, |
| absl::ToChronoMilliseconds(io_timeout_), cb); |
| } |
| |
| void PushConsumerImpl::wrapAckMessageRequest(const MQMessageExt& msg, AckMessageRequest& request) { |
| request.mutable_group()->set_resource_namespace(resource_namespace_); |
| request.mutable_group()->set_name(group_name_); |
| request.mutable_topic()->set_resource_namespace(resource_namespace_); |
| request.mutable_topic()->set_name(msg.getTopic()); |
| request.set_client_id(clientId()); |
| request.set_message_id(msg.getMsgId()); |
| request.set_receipt_handle(msg.receiptHandle()); |
| } |
| |
| uint32_t PushConsumerImpl::consumeThreadPoolSize() const { |
| return consume_thread_pool_size_; |
| } |
| |
| void PushConsumerImpl::consumeThreadPoolSize(int thread_pool_size) { |
| if (thread_pool_size >= 1) { |
| consume_thread_pool_size_ = thread_pool_size; |
| } |
| } |
| |
| uint32_t PushConsumerImpl::consumeBatchSize() const { |
| return consume_batch_size_; |
| } |
| |
| void PushConsumerImpl::consumeBatchSize(uint32_t consume_batch_size) { |
| |
| // For FIFO messages, consume batch size should always be 1. |
| if (message_listener_ && message_listener_->listenerType() == MessageListenerType::FIFO) { |
| return; |
| } |
| |
| if (consume_batch_size >= 1) { |
| consume_batch_size_ = consume_batch_size; |
| } |
| } |
| |
| void PushConsumerImpl::registerMessageListener(MessageListener* message_listener) { |
| message_listener_ = message_listener; |
| } |
| |
| std::size_t PushConsumerImpl::getProcessQueueTableSize() { |
| absl::MutexLock lock(&process_queue_table_mtx_); |
| return process_queue_table_.size(); |
| } |
| |
| void PushConsumerImpl::setThrottle(const std::string& topic, uint32_t threshold) { |
| absl::MutexLock lock(&throttle_table_mtx_); |
| throttle_table_.emplace(topic, threshold); |
| // If consumer has started, update it dynamically. |
| if (getConsumeMessageService()) { |
| getConsumeMessageService()->throttle(topic, threshold); |
| } |
| } |
| |
| void PushConsumerImpl::iterateProcessQueue(const std::function<void(ProcessQueueSharedPtr)>& callback) { |
| absl::MutexLock lock(&process_queue_table_mtx_); |
| for (const auto& item : process_queue_table_) { |
| if (item.second->hasPendingMessages()) { |
| callback(item.second); |
| } |
| } |
| } |
| |
| void PushConsumerImpl::fetchRoutes() { |
| std::vector<std::string> topics; |
| { |
| absl::MutexLock lk(&topic_filter_expression_table_mtx_); |
| for (const auto& item : topic_filter_expression_table_) { |
| topics.emplace_back(item.first); |
| } |
| } |
| |
| if (topics.empty()) { |
| return; |
| } |
| |
| std::vector<std::string>::size_type countdown = topics.size(); |
| absl::Mutex mtx; |
| absl::CondVar cv; |
| int acquired = 0; |
| auto callback = [&](const std::error_code& ec, const TopicRouteDataPtr& route) { |
| absl::MutexLock lk(&mtx); |
| countdown--; |
| cv.SignalAll(); |
| if (!ec) { |
| acquired++; |
| } |
| }; |
| |
| for (const auto& topic : topics) { |
| getRouteFor(topic, callback); |
| } |
| |
| while (countdown) { |
| absl::MutexLock lk(&mtx); |
| cv.Wait(&mtx); |
| } |
| SPDLOG_INFO("Fetched route for {} out of {} topics", acquired, topics.size()); |
| } |
| |
| void PushConsumerImpl::prepareHeartbeatData(HeartbeatRequest& request) { |
| request.set_client_id(clientId()); |
| |
| if (message_listener_) { |
| switch (message_listener_->listenerType()) { |
| case MessageListenerType::FIFO: |
| request.set_fifo_flag(true); |
| break; |
| case MessageListenerType::STANDARD: |
| request.set_fifo_flag(false); |
| break; |
| } |
| } |
| |
| auto consumer_data = request.mutable_consumer_data(); |
| consumer_data->mutable_group()->set_name(group_name_); |
| consumer_data->mutable_group()->set_resource_namespace(resource_namespace_); |
| |
| switch (message_model_) { |
| case MessageModel::BROADCASTING: |
| consumer_data->set_consume_model(rmq::ConsumeModel::BROADCASTING); |
| break; |
| case MessageModel::CLUSTERING: |
| consumer_data->set_consume_model(rmq::ConsumeModel::CLUSTERING); |
| break; |
| default: |
| break; |
| } |
| |
| auto subscriptions = consumer_data->mutable_subscriptions(); |
| { |
| absl::MutexLock lk(&topic_filter_expression_table_mtx_); |
| for (const auto& entry : topic_filter_expression_table_) { |
| auto subscription = new rmq::SubscriptionEntry; |
| subscription->mutable_topic()->set_resource_namespace(resource_namespace_); |
| subscription->mutable_topic()->set_name(entry.first); |
| subscription->mutable_expression()->set_expression(entry.second.content_); |
| switch (entry.second.type_) { |
| case ExpressionType::TAG: |
| subscription->mutable_expression()->set_type(rmq::FilterType::TAG); |
| break; |
| case ExpressionType::SQL92: |
| subscription->mutable_expression()->set_type(rmq::FilterType::SQL); |
| break; |
| } |
| subscriptions->AddAllocated(subscription); |
| } |
| } |
| |
| assert(consume_message_service_); |
| switch (consume_message_service_->messageListenerType()) { |
| case MessageListenerType::FIFO: { |
| // TODO: Use enumeration in the protocol buffer specification? |
| request.set_fifo_flag(true); |
| break; |
| } |
| case MessageListenerType::STANDARD: { |
| request.set_fifo_flag(false); |
| break; |
| } |
| } |
| |
| consumer_data->set_consume_policy(rmq::ConsumePolicy::RESUME); |
| consumer_data->mutable_dead_letter_policy()->set_max_delivery_attempts(maxDeliveryAttempts()); |
| consumer_data->set_consume_type(rmq::ConsumeMessageType::PASSIVE); |
| } |
| |
| ClientResourceBundle PushConsumerImpl::resourceBundle() { |
| auto&& resource_bundle = ClientImpl::resourceBundle(); |
| resource_bundle.group_type = GroupType::SUBSCRIBER; |
| { |
| absl::MutexLock lk(&topic_filter_expression_table_mtx_); |
| for (auto& item : topic_filter_expression_table_) { |
| resource_bundle.topics.emplace_back(item.first); |
| } |
| } |
| return std::move(resource_bundle); |
| } |
| |
| void PushConsumerImpl::notifyClientTermination() { |
| NotifyClientTerminationRequest request; |
| request.mutable_consumer_group()->set_resource_namespace(resource_namespace_); |
| request.mutable_consumer_group()->set_name(group_name_); |
| request.set_client_id(clientId()); |
| ClientImpl::notifyClientTermination(request); |
| } |
| |
| ROCKETMQ_NAMESPACE_END |
