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 6b84809 Cpp client: add seek support in consumer for cpp client
(#1863)
6b84809 is described below
commit 6b84809406d52c26cf7faa1a5543db209b803cc3
Author: Jia Zhai <[email protected]>
AuthorDate: Tue Jun 5 13:30:07 2018 +0800
Cpp client: add seek support in consumer for cpp client (#1863)
* add seek support in consumer for cpp client
* add c wrapper
---
pulsar-client-cpp/include/pulsar/Consumer.h | 24 +++++++++
pulsar-client-cpp/include/pulsar/c/consumer.h | 5 ++
pulsar-client-cpp/lib/Commands.cc | 13 +++++
pulsar-client-cpp/lib/Commands.h | 2 +
pulsar-client-cpp/lib/Consumer.cc | 21 ++++++++
pulsar-client-cpp/lib/ConsumerImpl.cc | 40 +++++++++++++++
pulsar-client-cpp/lib/ConsumerImpl.h | 2 +
pulsar-client-cpp/lib/ConsumerImplBase.h | 1 +
pulsar-client-cpp/lib/PartitionedConsumerImpl.cc | 5 ++
pulsar-client-cpp/lib/PartitionedConsumerImpl.h | 1 +
pulsar-client-cpp/lib/c/c_Consumer.cc | 10 ++++
pulsar-client-cpp/tests/BasicEndToEndTest.cc | 65 ++++++++++++++++++++++++
12 files changed, 189 insertions(+)
diff --git a/pulsar-client-cpp/include/pulsar/Consumer.h
b/pulsar-client-cpp/include/pulsar/Consumer.h
index ff222ee..4272166 100644
--- a/pulsar-client-cpp/include/pulsar/Consumer.h
+++ b/pulsar-client-cpp/include/pulsar/Consumer.h
@@ -211,6 +211,30 @@ class Consumer {
*/
void getBrokerConsumerStatsAsync(BrokerConsumerStatsCallback callback);
+ /**
+ * Reset the subscription associated with this consumer to a specific
message id.
+ * The message id can either be a specific message or represent the first
or last messages in the topic.
+ *
+ * Note: this operation can only be done on non-partitioned topics. For
these, one can rather perform the
+ * seek() on the individual partitions.
+ *
+ * @param messageId
+ * the message id where to reposition the subscription
+ */
+ Result seek(const MessageId& msgId);
+
+ /**
+ * Asynchronously reset the subscription associated with this consumer to
a specific message id.
+ * The message id can either be a specific message or represent the first
or last messages in the topic.
+ *
+ * Note: this operation can only be done on non-partitioned topics. For
these, one can rather perform the
+ * seek() on the individual partitions.
+ *
+ * @param messageId
+ * the message id where to reposition the subscription
+ */
+ virtual void seekAsync(const MessageId& msgId, ResultCallback callback);
+
private:
typedef boost::shared_ptr<ConsumerImplBase> ConsumerImplBasePtr;
ConsumerImplBasePtr impl_;
diff --git a/pulsar-client-cpp/include/pulsar/c/consumer.h
b/pulsar-client-cpp/include/pulsar/c/consumer.h
index e3e683b..f350ee0 100644
--- a/pulsar-client-cpp/include/pulsar/c/consumer.h
+++ b/pulsar-client-cpp/include/pulsar/c/consumer.h
@@ -192,6 +192,11 @@ pulsar_result resume_message_listener(pulsar_consumer_t
*consumer);
*/
void pulsar_consumer_redeliver_unacknowledged_messages(pulsar_consumer_t
*consumer);
+void pulsar_consumer_seek_async(pulsar_consumer_t *consumer,
pulsar_message_id_t *messageId,
+ pulsar_result_callback callback, void *ctx);
+
+pulsar_result pulsar_consumer_seek(pulsar_consumer_t *consumer,
pulsar_message_id_t *messageId);
+
#pragma GCC visibility pop
#ifdef __cplusplus
diff --git a/pulsar-client-cpp/lib/Commands.cc
b/pulsar-client-cpp/lib/Commands.cc
index 27297b5..677fce0 100644
--- a/pulsar-client-cpp/lib/Commands.cc
+++ b/pulsar-client-cpp/lib/Commands.cc
@@ -298,6 +298,19 @@ SharedBuffer
Commands::newRedeliverUnacknowledgedMessages(uint64_t consumerId) {
return writeMessageWithSize(cmd);
}
+SharedBuffer Commands::newSeek(uint64_t consumerId, uint64_t requestId, const
MessageId& messageId) {
+ BaseCommand cmd;
+ cmd.set_type(BaseCommand::SEEK);
+ CommandSeek* commandSeek = cmd.mutable_seek();
+ commandSeek->set_consumer_id(consumerId);
+ commandSeek->set_request_id(requestId);
+
+ MessageIdData& messageIdData = *commandSeek->mutable_message_id();
+ messageIdData.set_ledgerid(messageId.ledgerId());
+ messageIdData.set_entryid(messageId.entryId());
+ return writeMessageWithSize(cmd);
+}
+
std::string Commands::messageType(BaseCommand_Type type) {
switch (type) {
case BaseCommand::CONNECT:
diff --git a/pulsar-client-cpp/lib/Commands.h b/pulsar-client-cpp/lib/Commands.h
index 3746dc5..38c498a 100644
--- a/pulsar-client-cpp/lib/Commands.h
+++ b/pulsar-client-cpp/lib/Commands.h
@@ -109,6 +109,8 @@ class Commands {
static SharedBuffer newConsumerStats(uint64_t consumerId, uint64_t
requestId);
+ static SharedBuffer newSeek(uint64_t consumerId, uint64_t requestId, const
MessageId& messageId);
+
private:
Commands();
diff --git a/pulsar-client-cpp/lib/Consumer.cc
b/pulsar-client-cpp/lib/Consumer.cc
index 726676d..d89dd73 100644
--- a/pulsar-client-cpp/lib/Consumer.cc
+++ b/pulsar-client-cpp/lib/Consumer.cc
@@ -189,4 +189,25 @@ void
Consumer::getBrokerConsumerStatsAsync(BrokerConsumerStatsCallback callback)
}
impl_->getBrokerConsumerStatsAsync(callback);
}
+
+void Consumer::seekAsync(const MessageId& msgId, ResultCallback callback) {
+ if (!impl_) {
+ callback(ResultConsumerNotInitialized);
+ return;
+ }
+ impl_->seekAsync(msgId, callback);
+}
+
+Result Consumer::seek(const MessageId& msgId) {
+ if (!impl_) {
+ return ResultConsumerNotInitialized;
+ }
+
+ Promise<bool, Result> promise;
+ impl_->seekAsync(msgId, WaitForCallback(promise));
+ Result result;
+ promise.getFuture().get(result);
+ return result;
+}
+
} // namespace pulsar
diff --git a/pulsar-client-cpp/lib/ConsumerImpl.cc
b/pulsar-client-cpp/lib/ConsumerImpl.cc
index 2abbdd0..0425238 100644
--- a/pulsar-client-cpp/lib/ConsumerImpl.cc
+++ b/pulsar-client-cpp/lib/ConsumerImpl.cc
@@ -879,4 +879,44 @@ void ConsumerImpl::brokerConsumerStatsListener(Result res,
BrokerConsumerStatsIm
}
}
+void ConsumerImpl::handleSeek(Result result, ResultCallback callback) {
+ if (result == ResultOk) {
+ LOG_INFO(getName() << "Seek successfully");
+ } else {
+ LOG_ERROR(getName() << "Failed to seek: " << strResult(result));
+ }
+ callback(result);
+}
+
+void ConsumerImpl::seekAsync(const MessageId& msgId, ResultCallback callback) {
+ Lock lock(mutex_);
+ if (state_ == Closed || state_ == Closing) {
+ lock.unlock();
+ LOG_ERROR(getName() << "Client connection already closed.");
+ if (!callback.empty()) {
+ callback(ResultAlreadyClosed);
+ }
+ return;
+ }
+ lock.unlock();
+
+ ClientConnectionPtr cnx = getCnx().lock();
+ if (cnx) {
+ ClientImplPtr client = client_.lock();
+ uint64_t requestId = client->newRequestId();
+ LOG_DEBUG(getName() << " Sending seek Command for Consumer - " <<
getConsumerId() << ", requestId - "
+ << requestId);
+ Future<Result, ResponseData> future =
+ cnx->sendRequestWithId(Commands::newSeek(consumerId_, requestId,
msgId), requestId);
+
+ if (!callback.empty()) {
+ future.addListener(boost::bind(&ConsumerImpl::handleSeek,
shared_from_this(), _1, callback));
+ }
+ return;
+ }
+
+ LOG_ERROR(getName() << " Client Connection not ready for Consumer");
+ callback(ResultNotConnected);
+}
+
} /* namespace pulsar */
diff --git a/pulsar-client-cpp/lib/ConsumerImpl.h
b/pulsar-client-cpp/lib/ConsumerImpl.h
index 45a12ef..e35f444 100644
--- a/pulsar-client-cpp/lib/ConsumerImpl.h
+++ b/pulsar-client-cpp/lib/ConsumerImpl.h
@@ -102,6 +102,8 @@ class ConsumerImpl : public ConsumerImplBase,
virtual Result resumeMessageListener();
virtual void redeliverUnacknowledgedMessages();
virtual void getBrokerConsumerStatsAsync(BrokerConsumerStatsCallback
callback);
+ void handleSeek(Result result, ResultCallback callback);
+ virtual void seekAsync(const MessageId& msgId, ResultCallback callback);
protected:
void connectionOpened(const ClientConnectionPtr& cnx);
diff --git a/pulsar-client-cpp/lib/ConsumerImplBase.h
b/pulsar-client-cpp/lib/ConsumerImplBase.h
index a1f3bd5..11f2fc6 100644
--- a/pulsar-client-cpp/lib/ConsumerImplBase.h
+++ b/pulsar-client-cpp/lib/ConsumerImplBase.h
@@ -49,6 +49,7 @@ class ConsumerImplBase {
virtual const std::string& getName() const = 0;
virtual int getNumOfPrefetchedMessages() const = 0;
virtual void getBrokerConsumerStatsAsync(BrokerConsumerStatsCallback
callback) = 0;
+ virtual void seekAsync(const MessageId& msgId, ResultCallback callback) =
0;
};
} // namespace pulsar
#endif // PULSAR_CONSUMER_IMPL_BASE_HEADER
diff --git a/pulsar-client-cpp/lib/PartitionedConsumerImpl.cc
b/pulsar-client-cpp/lib/PartitionedConsumerImpl.cc
index dc218ef..199edb2 100644
--- a/pulsar-client-cpp/lib/PartitionedConsumerImpl.cc
+++ b/pulsar-client-cpp/lib/PartitionedConsumerImpl.cc
@@ -407,4 +407,9 @@ void PartitionedConsumerImpl::handleGetConsumerStats(Result
res, BrokerConsumerS
callback(ResultOk, BrokerConsumerStats(statsPtr));
}
}
+
+void PartitionedConsumerImpl::seekAsync(const MessageId& msgId, ResultCallback
callback) {
+ callback(ResultOperationNotSupported);
+}
+
} // namespace pulsar
diff --git a/pulsar-client-cpp/lib/PartitionedConsumerImpl.h
b/pulsar-client-cpp/lib/PartitionedConsumerImpl.h
index 56d04205..606c007 100644
--- a/pulsar-client-cpp/lib/PartitionedConsumerImpl.h
+++ b/pulsar-client-cpp/lib/PartitionedConsumerImpl.h
@@ -68,6 +68,7 @@ class PartitionedConsumerImpl : public ConsumerImplBase,
virtual void getBrokerConsumerStatsAsync(BrokerConsumerStatsCallback
callback);
void handleGetConsumerStats(Result, BrokerConsumerStats, LatchPtr,
PartitionedBrokerConsumerStatsPtr,
size_t, BrokerConsumerStatsCallback);
+ virtual void seekAsync(const MessageId& msgId, ResultCallback callback);
private:
const ClientImplPtr client_;
diff --git a/pulsar-client-cpp/lib/c/c_Consumer.cc
b/pulsar-client-cpp/lib/c/c_Consumer.cc
index dae824a..f9a211d 100644
--- a/pulsar-client-cpp/lib/c/c_Consumer.cc
+++ b/pulsar-client-cpp/lib/c/c_Consumer.cc
@@ -122,3 +122,13 @@ pulsar_result resume_message_listener(pulsar_consumer_t
*consumer) {
void pulsar_consumer_redeliver_unacknowledged_messages(pulsar_consumer_t
*consumer) {
return consumer->consumer.redeliverUnacknowledgedMessages();
}
+
+void pulsar_consumer_seek_async(pulsar_consumer_t *consumer,
pulsar_message_id_t *messageId,
+ pulsar_result_callback callback, void *ctx) {
+ consumer->consumer.seekAsync(messageId->messageId,
+ boost::bind(handle_result_callback, _1,
callback, ctx));
+}
+
+pulsar_result pulsar_consumer_seek(pulsar_consumer_t *consumer,
pulsar_message_id_t *messageId) {
+ return (pulsar_result)consumer->consumer.seek(messageId->messageId);
+}
diff --git a/pulsar-client-cpp/tests/BasicEndToEndTest.cc
b/pulsar-client-cpp/tests/BasicEndToEndTest.cc
index f9a6592..47713e9 100644
--- a/pulsar-client-cpp/tests/BasicEndToEndTest.cc
+++ b/pulsar-client-cpp/tests/BasicEndToEndTest.cc
@@ -1329,3 +1329,68 @@ TEST(BasicEndToEndTest, testEventTime) {
consumer.close();
producer.close();
}
+
+TEST(BasicEndToEndTest, testSeek) {
+ ClientConfiguration config;
+ Client client(lookupUrl);
+ std::string topicName = "persistent://prop/unit/ns1/testSeek";
+ std::string subName = "sub-testSeek";
+ Producer producer;
+
+ Promise<Result, Producer> producerPromise;
+ client.createProducerAsync(topicName,
WaitForCallbackValue<Producer>(producerPromise));
+ Future<Result, Producer> producerFuture = producerPromise.getFuture();
+ Result result = producerFuture.get(producer);
+ ASSERT_EQ(ResultOk, result);
+
+ Consumer consumer;
+ ConsumerConfiguration consConfig;
+ consConfig.setReceiverQueueSize(1);
+ Promise<Result, Consumer> consumerPromise;
+ client.subscribeAsync(topicName, subName, consConfig,
WaitForCallbackValue<Consumer>(consumerPromise));
+ Future<Result, Consumer> consumerFuture = consumerPromise.getFuture();
+ result = consumerFuture.get(consumer);
+ ASSERT_EQ(ResultOk, result);
+ std::string temp = producer.getTopic();
+ ASSERT_EQ(temp, topicName);
+ temp = consumer.getTopic();
+ ASSERT_EQ(temp, topicName);
+ ASSERT_EQ(consumer.getSubscriptionName(), subName);
+
+ // Send 1000 messages synchronously
+ std::string msgContent = "msg-content";
+ LOG_INFO("Publishing 100 messages synchronously");
+ int msgNum = 0;
+ for (; msgNum < 100; msgNum++) {
+ std::stringstream stream;
+ stream << msgContent << msgNum;
+ Message msg = MessageBuilder().setContent(stream.str()).build();
+ ASSERT_EQ(ResultOk, producer.send(msg));
+ }
+
+ LOG_INFO("Trying to receive 100 messages");
+ Message msgReceived;
+ for (msgNum = 0; msgNum < 100; msgNum++) {
+ consumer.receive(msgReceived, 100);
+ LOG_DEBUG("Received message :" << msgReceived.getMessageId());
+ std::stringstream expected;
+ expected << msgContent << msgNum;
+ ASSERT_EQ(expected.str(), msgReceived.getDataAsString());
+ ASSERT_EQ(ResultOk, consumer.acknowledge(msgReceived));
+ }
+
+ // seek to earliest, expected receive first message.
+ result = consumer.seek(MessageId::earliest());
+ ASSERT_EQ(ResultOk, result);
+ consumer.receive(msgReceived, 100);
+ LOG_ERROR("Received message :" << msgReceived.getMessageId());
+ std::stringstream expected;
+ msgNum = 0;
+ expected << msgContent << msgNum;
+ ASSERT_EQ(expected.str(), msgReceived.getDataAsString());
+
+ ASSERT_EQ(ResultOk, consumer.unsubscribe());
+ ASSERT_EQ(ResultAlreadyClosed, consumer.close());
+ ASSERT_EQ(ResultOk, producer.close());
+ ASSERT_EQ(ResultOk, client.close());
+}
--
To stop receiving notification emails like this one, please contact
[email protected].
