This is an automated email from the ASF dual-hosted git repository.
mmerli pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new 0c1cc54 Enable C++ message listener to receive messages even if
receiver queue size is zero (#2010)
0c1cc54 is described below
commit 0c1cc5452197512d42ea45a24a20981e183301a8
Author: massakam <[email protected]>
AuthorDate: Fri Jun 22 01:52:38 2018 +0900
Enable C++ message listener to receive messages even if receiver queue size
is zero (#2010)
---
pulsar-client-cpp/lib/ClientImpl.cc | 3 +-
pulsar-client-cpp/lib/ConsumerImpl.cc | 11 ++-
pulsar-client-cpp/tests/ZeroQueueSizeTest.cc | 115 +++++++++++++++++++++++++++
3 files changed, 125 insertions(+), 4 deletions(-)
diff --git a/pulsar-client-cpp/lib/ClientImpl.cc
b/pulsar-client-cpp/lib/ClientImpl.cc
index d8d07e7..1d46cd9 100644
--- a/pulsar-client-cpp/lib/ClientImpl.cc
+++ b/pulsar-client-cpp/lib/ClientImpl.cc
@@ -82,7 +82,8 @@ ClientImpl::ClientImpl(const std::string& serviceUrl, const
ClientConfiguration&
#ifdef USE_LOG4CXX
if (!clientConfiguration.getLogConfFilePath().empty()) {
// A log4cxx log file was passed through deprecated parameter. Use
that to configure Log4CXX
-
LogUtils::setLoggerFactory(Log4CxxLogger::create(clientConfiguration.getLogConfFilePath()));
+ LogUtils::setLoggerFactory(
+
Log4CxxLoggerFactory::create(clientConfiguration.getLogConfFilePath()));
} else {
// Use default simple console logger
LogUtils::setLoggerFactory(SimpleLoggerFactory::create());
diff --git a/pulsar-client-cpp/lib/ConsumerImpl.cc
b/pulsar-client-cpp/lib/ConsumerImpl.cc
index 55ded8f..df5cae5 100644
--- a/pulsar-client-cpp/lib/ConsumerImpl.cc
+++ b/pulsar-client-cpp/lib/ConsumerImpl.cc
@@ -181,8 +181,12 @@ void ConsumerImpl::handleCreateConsumer(const
ClientConnectionPtr& cnx, Result r
}
LOG_DEBUG(getName() << "Send initial flow permits: " <<
config_.getReceiverQueueSize());
- if ((consumerTopicType_ == NonPartitioned || !firstTime) &&
config_.getReceiverQueueSize() != 0) {
- receiveMessages(cnx, config_.getReceiverQueueSize());
+ if (consumerTopicType_ == NonPartitioned || !firstTime) {
+ if (config_.getReceiverQueueSize() != 0) {
+ receiveMessages(cnx, config_.getReceiverQueueSize());
+ } else if (messageListener_) {
+ receiveMessages(cnx, 1);
+ }
}
consumerCreatedPromise_.setValue(shared_from_this());
} else {
@@ -286,7 +290,8 @@ void ConsumerImpl::messageReceived(const
ClientConnectionPtr& cnx, const proto::
numOfMessageReceived = receiveIndividualMessagesFromBatch(cnx, m);
} else {
// config_.getReceiverQueueSize() != 0 or waiting For ZeroQueueSize
Message`
- if (config_.getReceiverQueueSize() != 0) {
+ if (config_.getReceiverQueueSize() != 0 ||
+ (config_.getReceiverQueueSize() == 0 && messageListener_)) {
incomingMessages_.push(m);
} else {
Lock lock(mutex_);
diff --git a/pulsar-client-cpp/tests/ZeroQueueSizeTest.cc
b/pulsar-client-cpp/tests/ZeroQueueSizeTest.cc
new file mode 100644
index 0000000..e1cd1fc
--- /dev/null
+++ b/pulsar-client-cpp/tests/ZeroQueueSizeTest.cc
@@ -0,0 +1,115 @@
+/**
+ * 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 <gtest/gtest.h>
+#include <pulsar/Client.h>
+#include <lib/Latch.h>
+#include "ConsumerTest.h"
+DECLARE_LOG_OBJECT()
+
+using namespace pulsar;
+
+static int totalMessages = 10;
+static int globalCount = 0;
+static std::string lookupUrl = "pulsar://localhost:8885";
+static std::string contentBase = "msg-";
+
+static void messageListenerFunction(Consumer consumer, const Message& msg,
Latch& latch) {
+ ASSERT_EQ(0, ConsumerTest::getNumOfMessagesInQueue(consumer));
+ std::ostringstream ss;
+ ss << contentBase << globalCount;
+ ASSERT_EQ(ss.str(), msg.getDataAsString());
+ globalCount++;
+ latch.countdown();
+ ASSERT_EQ(0, ConsumerTest::getNumOfMessagesInQueue(consumer));
+}
+
+TEST(ZeroQueueSizeTest, testProduceConsume) {
+ Client client(lookupUrl);
+ std::string topicName = "persistent://prop/unit/ns1/zero-queue-size";
+ std::string subName = "my-sub-name";
+
+ Producer producer;
+ Result result = client.createProducer(topicName, producer);
+ ASSERT_EQ(ResultOk, result);
+
+ Consumer consumer;
+ ConsumerConfiguration consConfig;
+ consConfig.setReceiverQueueSize(0);
+ result = client.subscribe(topicName, subName, consConfig, consumer);
+ ASSERT_EQ(ResultOk, result);
+
+ for (int i = 0; i < totalMessages; i++) {
+ std::ostringstream ss;
+ ss << contentBase << i;
+ Message msg = MessageBuilder().setContent(ss.str()).build();
+ result = producer.send(msg);
+ ASSERT_EQ(ResultOk, result);
+ }
+
+ for (int i = 0; i < totalMessages; i++) {
+ ASSERT_EQ(0, ConsumerTest::getNumOfMessagesInQueue(consumer));
+ std::ostringstream ss;
+ ss << contentBase << i;
+ Message receivedMsg;
+ consumer.receive(receivedMsg);
+ ASSERT_EQ(ss.str(), receivedMsg.getDataAsString());
+ ASSERT_EQ(0, ConsumerTest::getNumOfMessagesInQueue(consumer));
+ }
+
+ consumer.unsubscribe();
+ consumer.close();
+ producer.close();
+ client.close();
+}
+
+TEST(ZeroQueueSizeTest, testMessageListener) {
+ Client client(lookupUrl);
+ std::string topicName =
"persistent://prop/unit/ns/zero-queue-size-listener";
+ std::string subName = "my-sub-name";
+
+ Producer producer;
+ Result result = client.createProducer(topicName, producer);
+ ASSERT_EQ(ResultOk, result);
+
+ Consumer consumer;
+ ConsumerConfiguration consConfig;
+ consConfig.setReceiverQueueSize(0);
+ Latch latch(totalMessages);
+ consConfig.setMessageListener(boost::bind(messageListenerFunction, _1, _2,
latch));
+ result = client.subscribe(topicName, subName, consConfig, consumer);
+ ASSERT_EQ(ResultOk, result);
+
+ globalCount = 0;
+
+ for (int i = 0; i < totalMessages; i++) {
+ std::ostringstream ss;
+ ss << contentBase << i;
+ Message msg = MessageBuilder().setContent(ss.str()).build();
+ result = producer.send(msg);
+ ASSERT_EQ(ResultOk, result);
+ }
+
+ ASSERT_TRUE(latch.wait(milliseconds(30 * 1000)));
+ ASSERT_EQ(globalCount, totalMessages);
+
+ consumer.unsubscribe();
+ consumer.close();
+ producer.close();
+ client.close();
+}
