cpp/source/rocketmq/SimpleConsumerImpl.cpp - rocketmq-clients - Git at Google

 /*
  * 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 "SimpleConsumerImpl.h"

 #include "Signature.h"
 #include "google/protobuf/util/time_util.h"
 #include "rocketmq/ErrorCode.h"

 ROCKETMQ_NAMESPACE_BEGIN

 SimpleConsumerImpl::SimpleConsumerImpl(std::string group) : ClientImpl(group) {
   client_config_.subscriber.polling_timeout = absl::FromChrono(MixAll::DefaultReceiveMessageTimeout);
 }

 SimpleConsumerImpl::~SimpleConsumerImpl() {
   shutdown();
 }

 void SimpleConsumerImpl::prepareHeartbeatData(rmq::HeartbeatRequest& request) {
   request.set_client_type(rmq::ClientType::SIMPLE_CONSUMER);
   request.mutable_group()->CopyFrom(client_config_.subscriber.group);
 }

 void SimpleConsumerImpl::buildClientSettings(rmq::Settings& settings) {
   settings.set_client_type(rmq::ClientType::SIMPLE_CONSUMER);

   settings.mutable_subscription()->mutable_group()->CopyFrom(client_config_.subscriber.group);

   auto subscriptions = settings.mutable_subscription()->mutable_subscriptions();

   settings.mutable_access_point()->CopyFrom(accessPoint());
   {
     absl::MutexLock lk(&subscriptions_mtx_);
     for (const auto& entry : subscriptions_) {
       auto subscription_entry = new rmq::SubscriptionEntry;
       subscription_entry->mutable_topic()->set_name(entry.first);
       subscription_entry->mutable_topic()->set_resource_namespace(resourceNamespace());

       switch (entry.second.type_) {
         case ExpressionType::TAG: {
           subscription_entry->mutable_expression()->set_type(rmq::FilterType::TAG);
           break;
         }
         case ExpressionType::SQL92: {
           subscription_entry->mutable_expression()->set_type(rmq::FilterType::SQL);
           break;
         }
       }
       subscription_entry->mutable_expression()->set_expression(entry.second.content_);
       subscriptions->AddAllocated(subscription_entry);
     }
   }
 }

 void SimpleConsumerImpl::topicsOfInterest(std::vector<std::string> topics) {
   absl::MutexLock lk(&subscriptions_mtx_);
   for (const auto& entry : subscriptions_) {
     if (std::find(topics.begin(), topics.end(), entry.first) == topics.end()) {
       topics.push_back(entry.first);
     }
   }
 }

 /**
  * @brief Start SimpleConsumer
  *
  * During start, we need synchronously fetch routes and query assignments
  */
 void SimpleConsumerImpl::start() {
   ClientImpl::start();
   State expected = State::STARTING;
   if (state_.compare_exchange_strong(expected, State::STARTED, std::memory_order_relaxed)) {
     refreshAssignments();

     std::weak_ptr<SimpleConsumerImpl> consumer(shared_from_this());
     auto refresh_assignment_task = [consumer]() {
       auto simple_consumer = consumer.lock();
       if (simple_consumer) {
         simple_consumer->refreshAssignments0();
       }
     };
     refresh_assignment_task_ = manager()->getScheduler()->schedule(refresh_assignment_task, "RefreshAssignmentTask",
                                                                    std::chrono::seconds(3), std::chrono::seconds(3));
   }
 }

 void SimpleConsumerImpl::shutdown() {
   State expected = State::STARTED;

   if (state_.compare_exchange_strong(expected, State::STOPPING, std::memory_order_relaxed)) {
     manager()->getScheduler()->cancel(refresh_assignment_task_);
     ClientImpl::shutdown();
   }
 }

 void SimpleConsumerImpl::subscribe(std::string topic, FilterExpression expression) {
   absl::MutexLock lk(&subscriptions_mtx_);
   subscriptions_.insert_or_assign(topic, expression);
 }

 void SimpleConsumerImpl::unsubscribe(const std::string& topic) {
   {
     absl::MutexLock lk(&subscriptions_mtx_);
     subscriptions_.erase(topic);
   }

   removeAssignmentsByTopic(topic);
 }

 void SimpleConsumerImpl::removeAssignmentsByTopic(const std::string& topic) {
   std::vector<rmq::Assignment> assignments;
   {
     absl::MutexLock lk(&topic_assignments_mtx_);
     if (!topic_assignments_.contains(topic)) {
       return;
     }

     const auto& items = topic_assignments_[topic];
     assignments.insert(assignments.end(), items.begin(), items.end());
     topic_assignments_.erase(topic);
   }

   {
     absl::MutexLock lk(&assignments_mtx_);
     auto it = std::remove_if(assignments_.begin(), assignments_.end(), [&](const rmq::Assignment& assignment) {
       return std::find_if(assignments.begin(), assignments.end(),
                           [&](const rmq::Assignment& e) { return e == assignment; }) != assignments.end();
     });
     assignments_.erase(it, assignments_.end());
   }
 }

 void SimpleConsumerImpl::refreshAssignments0() {
   std::vector<std::string> topics;
   {
     absl::MutexLock lk(&subscriptions_mtx_);
     for (const auto& entry : subscriptions_) {
       topics.push_back(entry.first);
     }
   }

   auto no_op = [](const std::error_code& ec) {};
   for (const auto& topic : topics) {
     refreshAssignment(topic, no_op);
   }
 }

 void SimpleConsumerImpl::updateAssignments(const std::string& topic, const std::vector<rmq::Assignment>& assignments) {
   bool changed = false;
   {
     absl::MutexLock lk(&topic_assignments_mtx_);
     if (!topic_assignments_.contains(topic)) {
       changed = true;
       topic_assignments_.insert({topic, assignments});
       {
         absl::MutexLock assignment_lk(&assignments_mtx_);
         assignments_.insert(assignments_.begin(), assignments.begin(), assignments.end());
       }
     } else if (!assignments.empty()) {
       const auto& prev = topic_assignments_[topic];
       std::vector<rmq::Assignment> to_remove;
       std::vector<rmq::Assignment> to_add;
       for (const auto& item : prev) {
         if (std::find_if(assignments.begin(), assignments.end(), [&](const rmq::Assignment& e) { return item == e; }) ==
             assignments.end()) {
           to_remove.push_back(item);
         }
       }

       for (const auto& entry : assignments) {
         if (std::find_if(prev.begin(), prev.end(), [&](const rmq::Assignment e) { return e == entry; }) == prev.end()) {
           to_add.push_back(entry);
         }
       }

       if (!to_remove.empty() || !to_add.empty()) {
         changed = true;
         absl::MutexLock lk(&assignments_mtx_);
         for (const auto& item : to_remove) {
           std::remove_if(assignments_.begin(), assignments_.end(), [&](const rmq::Assignment& e) { return e == item; });
         }

         for (const auto& item : to_add) {
           assignments_.push_back(item);
         }
         topic_assignments_.insert_or_assign(topic, assignments);
       }
     }
   }

   if (changed) {
     SPDLOG_DEBUG("Assignments for topic={} change to: {}", topic,
                  absl::StrJoin(assignments.begin(), assignments.end(), ",",
                                [](std::string* out, const rmq::Assignment& assignment) {
                                  out->append(assignment.DebugString());
                                }));
   }
 }

 thread_local std::size_t SimpleConsumerImpl::assignment_index_ = 0;

 void SimpleConsumerImpl::refreshAssignment(const std::string& topic, std::function<void(const std::error_code&)> cb) {
   absl::flat_hash_set<std::string> endpoints;
   endpointsInUse(endpoints);
   if (endpoints.empty()) {
     SPDLOG_WARN("No broker is available");
     return;
   }

   rmq::QueryAssignmentRequest query_assignment_request;
   query_assignment_request.mutable_topic()->set_name(topic);
   query_assignment_request.mutable_topic()->set_resource_namespace(resourceNamespace());

   query_assignment_request.mutable_group()->CopyFrom(client_config_.subscriber.group);
   query_assignment_request.mutable_endpoints()->CopyFrom(accessPoint());

   Metadata metadata;
   Signature::sign(client_config_, metadata);

   std::weak_ptr<SimpleConsumerImpl> consumer(shared_from_this());
   auto callback = [consumer, topic, cb](const std::error_code& ec, const rmq::QueryAssignmentResponse& response) {
     auto simple_consumer = consumer.lock();
     const auto& assignments = response.assignments();
     if (assignments.empty()) {
       cb(ec);
       return;
     }

     std::vector<rmq::Assignment> assigns;
     assigns.insert(assigns.begin(), assignments.begin(), assignments.end());
     simple_consumer->updateAssignments(topic, assigns);
     cb(ec);
   };

   manager()->queryAssignment(*endpoints.begin(), metadata, query_assignment_request,
                              absl::ToChronoMilliseconds(client_config_.request_timeout), callback);
 }

 void SimpleConsumerImpl::refreshAssignments() {
   std::vector<std::string> topics;
   {
     absl::MutexLock lk(&subscriptions_mtx_);
     for (const auto& entry : subscriptions_) {
       topics.push_back(entry.first);
     }
   }

   auto mtx = std::make_shared<absl::Mutex>();
   auto cv = std::make_shared<absl::CondVar>();
   bool completed;
   for (const auto& topic : topics) {
     completed = false;
     auto callback = [&, mtx, cv](const std::error_code& ec) {
       absl::MutexLock lk(mtx.get());
       completed = true;
       cv->Signal();
     };

     refreshAssignment(topic, callback);

     {
       absl::MutexLock lk(mtx.get());
       if (!completed) {
         cv->Wait(mtx.get());
       }
     }
     SPDLOG_INFO("Assignments for {} received", topic);
   }
 }

 void SimpleConsumerImpl::receive(std::size_t limit,
                                  std::chrono::milliseconds invisible_duration,
                                  ReceiveCallback callback) {
   rmq::Assignment assignment;
   {
     absl::MutexLock lk(&assignments_mtx_);
     if (assignments_.empty()) {
       std::error_code ec = ErrorCode::NotFound;
       std::vector<MessageConstSharedPtr> messages;
       callback(ec, messages);
       return;
     }
     std::size_t idx = ++assignment_index_ % assignments_.size();
     assignment.CopyFrom(assignments_[idx]);
   }

   const auto& target = urlOf(assignment.message_queue());
   Metadata metadata;
   Signature::sign(client_config_, metadata);

   rmq::ReceiveMessageRequest request;
   request.set_auto_renew(false);
   request.mutable_group()->CopyFrom(config().subscriber.group);
   request.mutable_message_queue()->CopyFrom(assignment.message_queue());
   request.set_batch_size(limit);

   auto duration = google::protobuf::util::TimeUtil::MillisecondsToDuration(invisible_duration.count());

   request.mutable_invisible_duration()->set_nanos(duration.nanos());
   request.mutable_invisible_duration()->set_seconds(duration.seconds());

   auto cb = [callback](const std::error_code& ec, const ReceiveMessageResult& result) {
     std::vector<MessageConstSharedPtr> messages;
     if (ec) {
       callback(ec, messages);
       return;
     }

     callback(ec, result.messages);
   };

   auto timeout = absl::ToChronoMilliseconds(config().subscriber.polling_timeout);
   SPDLOG_DEBUG("ReceiveMessage.polling_timeout: {}ms", timeout.count());
   manager()->receiveMessage(target, metadata, request, timeout, cb);
 }

 void SimpleConsumerImpl::wrapAckRequest(const Message& message, AckMessageRequest& request) {
   request.mutable_group()->CopyFrom(client_config_.subscriber.group);

   request.mutable_topic()->set_resource_namespace(resourceNamespace());
   request.mutable_topic()->set_name(message.topic());

   auto entry = new rmq::AckMessageEntry();
   entry->set_message_id(message.id());
   entry->set_receipt_handle(message.extension().receipt_handle);

   request.mutable_entries()->AddAllocated(entry);
 }

 void SimpleConsumerImpl::ack(const Message& message, std::error_code& ec) {
   Metadata metadata;
   Signature::sign(client_config_, metadata);
   AckMessageRequest request;
   wrapAckRequest(message, request);

   auto mtx = std::make_shared<absl::Mutex>();
   auto cv = std::make_shared<absl::CondVar>();
   bool completed = false;

   auto callback = [&, mtx, cv](const std::error_code& err) {
     absl::MutexLock lk(mtx.get());
     completed = true;
     ec = err;
     cv->Signal();
   };

   manager()->ack(message.extension().target_endpoint, metadata, request,
                  absl::ToChronoMilliseconds(client_config_.request_timeout), callback);

   {
     absl::MutexLock lk(mtx.get());
     if (!completed) {
       cv->Wait(mtx.get());
     }
   }
 }

 void SimpleConsumerImpl::ackAsync(const Message& message, AckCallback callback) {
   Metadata metadata;
   Signature::sign(client_config_, metadata);
   AckMessageRequest request;
   wrapAckRequest(message, request);
   manager()->ack(message.extension().target_endpoint, metadata, request,
                  absl::ToChronoMilliseconds(client_config_.request_timeout), callback);
 }

 void SimpleConsumerImpl::changeInvisibleDuration(const Message& message,
                                                  std::chrono::milliseconds duration,
                                                  ChangeInvisibleDurationCallback callback) {
   Metadata metadata;
   Signature::sign(client_config_, metadata);

   rmq::ChangeInvisibleDurationRequest request;
   request.mutable_group()->CopyFrom(client_config_.subscriber.group);
   request.mutable_topic()->set_resource_namespace(resourceNamespace());
   request.mutable_topic()->set_name(message.topic());
   request.set_message_id(message.id());
   request.set_receipt_handle(message.extension().receipt_handle);
   auto d = google::protobuf::util::TimeUtil::MillisecondsToDuration(duration.count());
   request.mutable_invisible_duration()->CopyFrom(d);

   manager()->changeInvisibleDuration(message.extension().target_endpoint, metadata, request, duration, callback);
 }

 ROCKETMQ_NAMESPACE_END
	/*
	* 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 "SimpleConsumerImpl.h"

	#include "Signature.h"
	#include "google/protobuf/util/time_util.h"
	#include "rocketmq/ErrorCode.h"

	ROCKETMQ_NAMESPACE_BEGIN

	SimpleConsumerImpl::SimpleConsumerImpl(std::string group) : ClientImpl(group) {
	client_config_.subscriber.polling_timeout = absl::FromChrono(MixAll::DefaultReceiveMessageTimeout);
	}

	SimpleConsumerImpl::~SimpleConsumerImpl() {
	shutdown();
	}

	void SimpleConsumerImpl::prepareHeartbeatData(rmq::HeartbeatRequest& request) {
	request.set_client_type(rmq::ClientType::SIMPLE_CONSUMER);
	request.mutable_group()->CopyFrom(client_config_.subscriber.group);
	}

	void SimpleConsumerImpl::buildClientSettings(rmq::Settings& settings) {
	settings.set_client_type(rmq::ClientType::SIMPLE_CONSUMER);

	settings.mutable_subscription()->mutable_group()->CopyFrom(client_config_.subscriber.group);

	auto subscriptions = settings.mutable_subscription()->mutable_subscriptions();

	settings.mutable_access_point()->CopyFrom(accessPoint());
	{
	absl::MutexLock lk(&subscriptions_mtx_);
	for (const auto& entry : subscriptions_) {
	auto subscription_entry = new rmq::SubscriptionEntry;
	subscription_entry->mutable_topic()->set_name(entry.first);
	subscription_entry->mutable_topic()->set_resource_namespace(resourceNamespace());

	switch (entry.second.type_) {
	case ExpressionType::TAG: {
	subscription_entry->mutable_expression()->set_type(rmq::FilterType::TAG);
	break;
	}
	case ExpressionType::SQL92: {
	subscription_entry->mutable_expression()->set_type(rmq::FilterType::SQL);
	break;
	}
	}
	subscription_entry->mutable_expression()->set_expression(entry.second.content_);
	subscriptions->AddAllocated(subscription_entry);
	}
	}
	}

	void SimpleConsumerImpl::topicsOfInterest(std::vector<std::string> topics) {
	absl::MutexLock lk(&subscriptions_mtx_);
	for (const auto& entry : subscriptions_) {
	if (std::find(topics.begin(), topics.end(), entry.first) == topics.end()) {
	topics.push_back(entry.first);
	}
	}
	}

	/**
	* @brief Start SimpleConsumer
	*
	* During start, we need synchronously fetch routes and query assignments
	*/
	void SimpleConsumerImpl::start() {
	ClientImpl::start();
	State expected = State::STARTING;
	if (state_.compare_exchange_strong(expected, State::STARTED, std::memory_order_relaxed)) {
	refreshAssignments();

	std::weak_ptr<SimpleConsumerImpl> consumer(shared_from_this());
	auto refresh_assignment_task = [consumer]() {
	auto simple_consumer = consumer.lock();
	if (simple_consumer) {
	simple_consumer->refreshAssignments0();
	}
	};
	refresh_assignment_task_ = manager()->getScheduler()->schedule(refresh_assignment_task, "RefreshAssignmentTask",
	std::chrono::seconds(3), std::chrono::seconds(3));
	}
	}

	void SimpleConsumerImpl::shutdown() {
	State expected = State::STARTED;

	if (state_.compare_exchange_strong(expected, State::STOPPING, std::memory_order_relaxed)) {
	manager()->getScheduler()->cancel(refresh_assignment_task_);
	ClientImpl::shutdown();
	}
	}

	void SimpleConsumerImpl::subscribe(std::string topic, FilterExpression expression) {
	absl::MutexLock lk(&subscriptions_mtx_);
	subscriptions_.insert_or_assign(topic, expression);
	}

	void SimpleConsumerImpl::unsubscribe(const std::string& topic) {
	{
	absl::MutexLock lk(&subscriptions_mtx_);
	subscriptions_.erase(topic);
	}

	removeAssignmentsByTopic(topic);
	}

	void SimpleConsumerImpl::removeAssignmentsByTopic(const std::string& topic) {
	std::vector<rmq::Assignment> assignments;
	{
	absl::MutexLock lk(&topic_assignments_mtx_);
	if (!topic_assignments_.contains(topic)) {
	return;
	}

	const auto& items = topic_assignments_[topic];
	assignments.insert(assignments.end(), items.begin(), items.end());
	topic_assignments_.erase(topic);
	}

	{
	absl::MutexLock lk(&assignments_mtx_);
	auto it = std::remove_if(assignments_.begin(), assignments_.end(), [&](const rmq::Assignment& assignment) {
	return std::find_if(assignments.begin(), assignments.end(),
	[&](const rmq::Assignment& e) { return e == assignment; }) != assignments.end();
	});
	assignments_.erase(it, assignments_.end());
	}
	}

	void SimpleConsumerImpl::refreshAssignments0() {
	std::vector<std::string> topics;
	{
	absl::MutexLock lk(&subscriptions_mtx_);
	for (const auto& entry : subscriptions_) {
	topics.push_back(entry.first);
	}
	}

	auto no_op = [](const std::error_code& ec) {};
	for (const auto& topic : topics) {
	refreshAssignment(topic, no_op);
	}
	}

	void SimpleConsumerImpl::updateAssignments(const std::string& topic, const std::vector<rmq::Assignment>& assignments) {
	bool changed = false;
	{
	absl::MutexLock lk(&topic_assignments_mtx_);
	if (!topic_assignments_.contains(topic)) {
	changed = true;
	topic_assignments_.insert({topic, assignments});
	{
	absl::MutexLock assignment_lk(&assignments_mtx_);
	assignments_.insert(assignments_.begin(), assignments.begin(), assignments.end());
	}
	} else if (!assignments.empty()) {
	const auto& prev = topic_assignments_[topic];
	std::vector<rmq::Assignment> to_remove;
	std::vector<rmq::Assignment> to_add;
	for (const auto& item : prev) {
	if (std::find_if(assignments.begin(), assignments.end(), [&](const rmq::Assignment& e) { return item == e; }) ==
	assignments.end()) {
	to_remove.push_back(item);
	}
	}

	for (const auto& entry : assignments) {
	if (std::find_if(prev.begin(), prev.end(), [&](const rmq::Assignment e) { return e == entry; }) == prev.end()) {
	to_add.push_back(entry);
	}
	}

	if (!to_remove.empty() \|\| !to_add.empty()) {
	changed = true;
	absl::MutexLock lk(&assignments_mtx_);
	for (const auto& item : to_remove) {
	std::remove_if(assignments_.begin(), assignments_.end(), [&](const rmq::Assignment& e) { return e == item; });
	}

	for (const auto& item : to_add) {
	assignments_.push_back(item);
	}
	topic_assignments_.insert_or_assign(topic, assignments);
	}
	}
	}

	if (changed) {
	SPDLOG_DEBUG("Assignments for topic={} change to: {}", topic,
	absl::StrJoin(assignments.begin(), assignments.end(), ",",
	[](std::string* out, const rmq::Assignment& assignment) {
	out->append(assignment.DebugString());
	}));
	}
	}

	thread_local std::size_t SimpleConsumerImpl::assignment_index_ = 0;

	void SimpleConsumerImpl::refreshAssignment(const std::string& topic, std::function<void(const std::error_code&)> cb) {
	absl::flat_hash_set<std::string> endpoints;
	endpointsInUse(endpoints);
	if (endpoints.empty()) {
	SPDLOG_WARN("No broker is available");
	return;
	}

	rmq::QueryAssignmentRequest query_assignment_request;
	query_assignment_request.mutable_topic()->set_name(topic);
	query_assignment_request.mutable_topic()->set_resource_namespace(resourceNamespace());

	query_assignment_request.mutable_group()->CopyFrom(client_config_.subscriber.group);
	query_assignment_request.mutable_endpoints()->CopyFrom(accessPoint());

	Metadata metadata;
	Signature::sign(client_config_, metadata);

	std::weak_ptr<SimpleConsumerImpl> consumer(shared_from_this());
	auto callback = [consumer, topic, cb](const std::error_code& ec, const rmq::QueryAssignmentResponse& response) {
	auto simple_consumer = consumer.lock();
	const auto& assignments = response.assignments();
	if (assignments.empty()) {
	cb(ec);
	return;
	}

	std::vector<rmq::Assignment> assigns;
	assigns.insert(assigns.begin(), assignments.begin(), assignments.end());
	simple_consumer->updateAssignments(topic, assigns);
	cb(ec);
	};

	manager()->queryAssignment(*endpoints.begin(), metadata, query_assignment_request,
	absl::ToChronoMilliseconds(client_config_.request_timeout), callback);
	}

	void SimpleConsumerImpl::refreshAssignments() {
	std::vector<std::string> topics;
	{
	absl::MutexLock lk(&subscriptions_mtx_);
	for (const auto& entry : subscriptions_) {
	topics.push_back(entry.first);
	}
	}

	auto mtx = std::make_shared<absl::Mutex>();
	auto cv = std::make_shared<absl::CondVar>();
	bool completed;
	for (const auto& topic : topics) {
	completed = false;
	auto callback = [&, mtx, cv](const std::error_code& ec) {
	absl::MutexLock lk(mtx.get());
	completed = true;
	cv->Signal();
	};

	refreshAssignment(topic, callback);

	{
	absl::MutexLock lk(mtx.get());
	if (!completed) {
	cv->Wait(mtx.get());
	}
	}
	SPDLOG_INFO("Assignments for {} received", topic);
	}
	}

	void SimpleConsumerImpl::receive(std::size_t limit,
	std::chrono::milliseconds invisible_duration,
	ReceiveCallback callback) {
	rmq::Assignment assignment;
	{
	absl::MutexLock lk(&assignments_mtx_);
	if (assignments_.empty()) {
	std::error_code ec = ErrorCode::NotFound;
	std::vector<MessageConstSharedPtr> messages;
	callback(ec, messages);
	return;
	}
	std::size_t idx = ++assignment_index_ % assignments_.size();
	assignment.CopyFrom(assignments_[idx]);
	}

	const auto& target = urlOf(assignment.message_queue());
	Metadata metadata;
	Signature::sign(client_config_, metadata);

	rmq::ReceiveMessageRequest request;
	request.set_auto_renew(false);
	request.mutable_group()->CopyFrom(config().subscriber.group);
	request.mutable_message_queue()->CopyFrom(assignment.message_queue());
	request.set_batch_size(limit);

	auto duration = google::protobuf::util::TimeUtil::MillisecondsToDuration(invisible_duration.count());

	request.mutable_invisible_duration()->set_nanos(duration.nanos());
	request.mutable_invisible_duration()->set_seconds(duration.seconds());

	auto cb = [callback](const std::error_code& ec, const ReceiveMessageResult& result) {
	std::vector<MessageConstSharedPtr> messages;
	if (ec) {
	callback(ec, messages);
	return;
	}

	callback(ec, result.messages);
	};

	auto timeout = absl::ToChronoMilliseconds(config().subscriber.polling_timeout);
	SPDLOG_DEBUG("ReceiveMessage.polling_timeout: {}ms", timeout.count());
	manager()->receiveMessage(target, metadata, request, timeout, cb);
	}

	void SimpleConsumerImpl::wrapAckRequest(const Message& message, AckMessageRequest& request) {
	request.mutable_group()->CopyFrom(client_config_.subscriber.group);

	request.mutable_topic()->set_resource_namespace(resourceNamespace());
	request.mutable_topic()->set_name(message.topic());

	auto entry = new rmq::AckMessageEntry();
	entry->set_message_id(message.id());
	entry->set_receipt_handle(message.extension().receipt_handle);

	request.mutable_entries()->AddAllocated(entry);
	}

	void SimpleConsumerImpl::ack(const Message& message, std::error_code& ec) {
	Metadata metadata;
	Signature::sign(client_config_, metadata);
	AckMessageRequest request;
	wrapAckRequest(message, request);

	auto mtx = std::make_shared<absl::Mutex>();
	auto cv = std::make_shared<absl::CondVar>();
	bool completed = false;

	auto callback = [&, mtx, cv](const std::error_code& err) {
	absl::MutexLock lk(mtx.get());
	completed = true;
	ec = err;
	cv->Signal();
	};

	manager()->ack(message.extension().target_endpoint, metadata, request,
	absl::ToChronoMilliseconds(client_config_.request_timeout), callback);

	{
	absl::MutexLock lk(mtx.get());
	if (!completed) {
	cv->Wait(mtx.get());
	}
	}
	}

	void SimpleConsumerImpl::ackAsync(const Message& message, AckCallback callback) {
	Metadata metadata;
	Signature::sign(client_config_, metadata);
	AckMessageRequest request;
	wrapAckRequest(message, request);
	manager()->ack(message.extension().target_endpoint, metadata, request,
	absl::ToChronoMilliseconds(client_config_.request_timeout), callback);
	}

	void SimpleConsumerImpl::changeInvisibleDuration(const Message& message,
	std::chrono::milliseconds duration,
	ChangeInvisibleDurationCallback callback) {
	Metadata metadata;
	Signature::sign(client_config_, metadata);

	rmq::ChangeInvisibleDurationRequest request;
	request.mutable_group()->CopyFrom(client_config_.subscriber.group);
	request.mutable_topic()->set_resource_namespace(resourceNamespace());
	request.mutable_topic()->set_name(message.topic());
	request.set_message_id(message.id());
	request.set_receipt_handle(message.extension().receipt_handle);
	auto d = google::protobuf::util::TimeUtil::MillisecondsToDuration(duration.count());
	request.mutable_invisible_duration()->CopyFrom(d);

	manager()->changeInvisibleDuration(message.extension().target_endpoint, metadata, request, duration, callback);
	}

	ROCKETMQ_NAMESPACE_END