You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by xy...@apache.org on 2023/12/04 02:25:49 UTC
(pulsar-client-cpp) branch main updated: Add the Consumer::getConsumerName API (#360)
This is an automated email from the ASF dual-hosted git repository.
xyz pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pulsar-client-cpp.git
The following commit(s) were added to refs/heads/main by this push:
new 63d494f Add the Consumer::getConsumerName API (#360)
63d494f is described below
commit 63d494f57df126cf633396c618886eeb79946408
Author: Yunze Xu <xy...@163.com>
AuthorDate: Mon Dec 4 10:25:44 2023 +0800
Add the Consumer::getConsumerName API (#360)
### Motivation
C++ client supports setting the consumer name via
`ConsumerConfiguration::setConsumerName` but it does not support getting
the consumer name for a given consumer.
### Modifications
Add the `getConsumerName` method to `ConsumerImplBase` to get the
consumer name from the config and a public method with the same name to
`Consumer`.
Add `ConsumerTest.testConsumerName` to verify the consumer name is set
correctly in the subscribe command.
---
include/pulsar/Consumer.h | 7 ++++++-
lib/Consumer.cc | 4 ++++
lib/ConsumerImpl.cc | 7 +++----
lib/ConsumerImpl.h | 1 -
lib/ConsumerImplBase.cc | 3 ++-
lib/ConsumerImplBase.h | 4 ++++
tests/ClientTest.cc | 17 +++++------------
tests/ConsumerTest.cc | 48 ++++++++++++++++++++++++++++++++++++++++++++++-
tests/PulsarAdminHelper.h | 42 +++++++++++++++++++++++++++++++++++++++++
9 files changed, 113 insertions(+), 20 deletions(-)
diff --git a/include/pulsar/Consumer.h b/include/pulsar/Consumer.h
index 6596894..63defdb 100644
--- a/include/pulsar/Consumer.h
+++ b/include/pulsar/Consumer.h
@@ -48,10 +48,15 @@ class PULSAR_PUBLIC Consumer {
const std::string& getTopic() const;
/**
- * @return the consumer name
+ * @return the subscription name
*/
const std::string& getSubscriptionName() const;
+ /**
+ * @return the consumer name
+ */
+ const std::string& getConsumerName() const;
+
/**
* Unsubscribe the current consumer from the topic.
*
diff --git a/lib/Consumer.cc b/lib/Consumer.cc
index bda7f86..031cdff 100644
--- a/lib/Consumer.cc
+++ b/lib/Consumer.cc
@@ -39,6 +39,10 @@ const std::string& Consumer::getSubscriptionName() const {
return impl_ != NULL ? impl_->getSubscriptionName() : EMPTY_STRING;
}
+const std::string& Consumer::getConsumerName() const {
+ return impl_ ? impl_->getConsumerName() : EMPTY_STRING;
+}
+
Result Consumer::unsubscribe() {
if (!impl_) {
return ResultConsumerNotInitialized;
diff --git a/lib/ConsumerImpl.cc b/lib/ConsumerImpl.cc
index b466683..955fb32 100644
--- a/lib/ConsumerImpl.cc
+++ b/lib/ConsumerImpl.cc
@@ -83,7 +83,6 @@ ConsumerImpl::ConsumerImpl(const ClientImplPtr client, const std::string& topic,
availablePermits_(0),
receiverQueueRefillThreshold_(config_.getReceiverQueueSize() / 2),
consumerId_(client->newConsumerId()),
- consumerName_(config_.getConsumerName()),
consumerStr_("[" + topic + ", " + subscriptionName + ", " + std::to_string(consumerId_) + "] "),
messageListenerRunning_(true),
negativeAcksTracker_(client, *this, conf),
@@ -249,7 +248,7 @@ Future<Result, bool> ConsumerImpl::connectionOpened(const ClientConnectionPtr& c
ClientImplPtr client = client_.lock();
uint64_t requestId = client->newRequestId();
SharedBuffer cmd = Commands::newSubscribe(
- topic(), subscription_, consumerId_, requestId, getSubType(), consumerName_, subscriptionMode_,
+ topic(), subscription_, consumerId_, requestId, getSubType(), getConsumerName(), subscriptionMode_,
subscribeMessageId, readCompacted_, config_.getProperties(), config_.getSubscriptionProperties(),
config_.getSchema(), getInitialPosition(), config_.isReplicateSubscriptionStateEnabled(),
config_.getKeySharedPolicy(), config_.getPriorityLevel());
@@ -1780,7 +1779,7 @@ void ConsumerImpl::processPossibleToDLQ(const MessageId& messageId, ProcessDLQCa
}
if (result != ResultOk) {
LOG_WARN("{" << self->topic() << "} {" << self->subscription_ << "} {"
- << self->consumerName_ << "} Failed to acknowledge the message {"
+ << self->getConsumerName() << "} Failed to acknowledge the message {"
<< originMessageId
<< "} of the original topic but send to the DLQ successfully : "
<< result);
@@ -1793,7 +1792,7 @@ void ConsumerImpl::processPossibleToDLQ(const MessageId& messageId, ProcessDLQCa
});
} else {
LOG_WARN("{" << self->topic() << "} {" << self->subscription_ << "} {"
- << self->consumerName_ << "} Failed to send DLQ message to {"
+ << self->getConsumerName() << "} Failed to send DLQ message to {"
<< self->deadLetterPolicy_.getDeadLetterTopic() << "} for message id "
<< "{" << originMessageId << "} : " << res);
cb(false);
diff --git a/lib/ConsumerImpl.h b/lib/ConsumerImpl.h
index 61d96b1..5e9ef04 100644
--- a/lib/ConsumerImpl.h
+++ b/lib/ConsumerImpl.h
@@ -216,7 +216,6 @@ class ConsumerImpl : public ConsumerImplBase {
std::atomic_int availablePermits_;
const int receiverQueueRefillThreshold_;
uint64_t consumerId_;
- std::string consumerName_;
const std::string consumerStr_;
int32_t partitionIndex_ = -1;
Promise<Result, ConsumerImplBaseWeakPtr> consumerCreatedPromise_;
diff --git a/lib/ConsumerImplBase.cc b/lib/ConsumerImplBase.cc
index 6c86aa4..3921237 100644
--- a/lib/ConsumerImplBase.cc
+++ b/lib/ConsumerImplBase.cc
@@ -33,7 +33,8 @@ ConsumerImplBase::ConsumerImplBase(ClientImplPtr client, const std::string& topi
const ConsumerConfiguration& conf, ExecutorServicePtr listenerExecutor)
: HandlerBase(client, topic, backoff),
listenerExecutor_(listenerExecutor),
- batchReceivePolicy_(conf.getBatchReceivePolicy()) {
+ batchReceivePolicy_(conf.getBatchReceivePolicy()),
+ consumerName_(conf.getConsumerName()) {
auto userBatchReceivePolicy = conf.getBatchReceivePolicy();
if (userBatchReceivePolicy.getMaxNumMessages() > conf.getReceiverQueueSize()) {
batchReceivePolicy_ =
diff --git a/lib/ConsumerImplBase.h b/lib/ConsumerImplBase.h
index 2f3420c..1b7e86e 100644
--- a/lib/ConsumerImplBase.h
+++ b/lib/ConsumerImplBase.h
@@ -82,6 +82,8 @@ class ConsumerImplBase : public HandlerBase {
virtual const std::string& getName() const override = 0;
virtual void hasMessageAvailableAsync(HasMessageAvailableCallback callback) = 0;
+ const std::string& getConsumerName() const noexcept { return consumerName_; }
+
protected:
// overrided methods from HandlerBase
Future<Result, bool> connectionOpened(const ClientConnectionPtr& cnx) override {
@@ -106,6 +108,8 @@ class ConsumerImplBase : public HandlerBase {
virtual bool hasEnoughMessagesForBatchReceive() const = 0;
private:
+ const std::string consumerName_;
+
virtual void setNegativeAcknowledgeEnabledForTesting(bool enabled) = 0;
friend class MultiTopicsConsumerImpl;
diff --git a/tests/ClientTest.cc b/tests/ClientTest.cc
index 5b4dd2e..8c5a32a 100644
--- a/tests/ClientTest.cc
+++ b/tests/ClientTest.cc
@@ -21,13 +21,11 @@
#include <pulsar/Version.h>
#include <algorithm>
-#include <boost/property_tree/json_parser.hpp>
-#include <boost/property_tree/ptree.hpp>
#include <chrono>
#include <future>
#include <sstream>
-#include "HttpHelper.h"
+#include "PulsarAdminHelper.h"
#include "PulsarFriend.h"
#include "WaitUtils.h"
#include "lib/ClientConnection.h"
@@ -335,18 +333,13 @@ class PulsarWrapper {
// When `subscription` is empty, get client versions of the producers.
// Otherwise, get client versions of the consumers under the subscribe.
static std::vector<std::string> getClientVersions(const std::string &topic, std::string subscription = "") {
- const auto url = adminUrl + "admin/v2/persistent/public/default/" + topic + "/stats";
- std::string responseData;
- int res = makeGetRequest(url, responseData);
- if (res != 200) {
- LOG_ERROR(url << " failed: " << res);
+ boost::property_tree::ptree root;
+ const auto error = getTopicStats(topic, root);
+ if (!error.empty()) {
+ LOG_ERROR(error);
return {};
}
- std::stringstream stream;
- stream << responseData;
- boost::property_tree::ptree root;
- boost::property_tree::read_json(stream, root);
std::vector<std::string> versions;
if (subscription.empty()) {
for (auto &child : root.get_child("publishers")) {
diff --git a/tests/ConsumerTest.cc b/tests/ConsumerTest.cc
index 0836fbf..3ca48ce 100644
--- a/tests/ConsumerTest.cc
+++ b/tests/ConsumerTest.cc
@@ -29,8 +29,8 @@
#include <thread>
#include <vector>
-#include "HttpHelper.h"
#include "NoOpsCryptoKeyReader.h"
+#include "PulsarAdminHelper.h"
#include "PulsarFriend.h"
#include "SynchronizedQueue.h"
#include "WaitUtils.h"
@@ -1430,4 +1430,50 @@ TEST(ConsumerTest, testCloseAgainBeforeCloseDone) {
ASSERT_TRUE(*done);
}
+inline std::string getConsumerName(const std::string& topic) {
+ boost::property_tree::ptree root;
+ const auto error = getTopicStats(topic, root);
+ if (!error.empty()) {
+ LOG_INFO(error);
+ return {};
+ }
+ return root.get_child("subscriptions")
+ .get_child("sub")
+ .get_child("consumers")
+ .front()
+ .second.get<std::string>("consumerName");
+}
+
+TEST(ConsumerTest, testConsumerName) {
+ Client client{lookupUrl};
+ Consumer consumer;
+ ASSERT_TRUE(consumer.getConsumerName().empty());
+ const auto topic1 = "consumer-test-consumer-name-1";
+ const auto topic2 = "consumer-test-consumer-name-2";
+
+ // Default consumer name
+ ASSERT_EQ(ResultOk, client.subscribe(topic1, "sub", consumer));
+ LOG_INFO("Random consumer name: " << consumer.getConsumerName());
+ ASSERT_FALSE(consumer.getConsumerName().empty()); // a random name
+ ASSERT_EQ(consumer.getConsumerName(), getConsumerName(topic1));
+ consumer.close();
+
+ // Single-topic consumer
+ ConsumerConfiguration conf;
+ const std::string consumerName = "custom-consumer";
+ conf.setConsumerName(consumerName);
+ ASSERT_EQ(ResultOk, client.subscribe(topic1, "sub", conf, consumer));
+ ASSERT_EQ(consumerName, consumer.getConsumerName());
+ ASSERT_EQ(consumerName, getConsumerName(topic1));
+ consumer.close();
+
+ // Multi-topics consumer
+ ASSERT_EQ(ResultOk, client.subscribe(std::vector<std::string>{topic1, topic2}, "sub", conf, consumer));
+ ASSERT_EQ(consumerName, consumer.getConsumerName());
+ ASSERT_EQ(consumerName, getConsumerName(topic1));
+ ASSERT_EQ(consumerName, getConsumerName(topic2));
+
+ client.close();
+}
+
} // namespace pulsar
diff --git a/tests/PulsarAdminHelper.h b/tests/PulsarAdminHelper.h
new file mode 100644
index 0000000..944c7b8
--- /dev/null
+++ b/tests/PulsarAdminHelper.h
@@ -0,0 +1,42 @@
+/**
+ * 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.
+ */
+#pragma once
+
+#include <boost/property_tree/json_parser.hpp>
+#include <boost/property_tree/ptree.hpp>
+
+#include "HttpHelper.h"
+
+namespace pulsar {
+
+inline std::string getTopicStats(const std::string& topic, boost::property_tree::ptree& root) {
+ const auto url = "http://localhost:8080/admin/v2/persistent/public/default/" + topic + "/stats";
+ std::string responseData;
+ int code = makeGetRequest(url, responseData);
+ if (code != 200) {
+ return url + " failed: " + std::to_string(code);
+ }
+
+ std::stringstream stream;
+ stream << responseData;
+ boost::property_tree::read_json(stream, root);
+ return "";
+}
+
+} // namespace pulsar