merlimat closed pull request #1863: Cpp client: add seek support in consumer
for cpp client
URL: https://github.com/apache/incubator-pulsar/pull/1863
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/pulsar-client-cpp/include/pulsar/Consumer.h
b/pulsar-client-cpp/include/pulsar/Consumer.h
index ff222ee7bd..4272166864 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 e3e683b173..f350ee00a2 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 27297b5193..677fce0e80 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 3746dc5a2f..38c498a463 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 726676da21..d89dd73c8a 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 2abbdd05d2..0425238567 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 45a12efbea..e35f44464b 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 a1f3bd560c..11f2fc6bb9 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 dc218efe1e..199edb26e1 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 56d0420529..606c007005 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 dae824a447..f9a211d6a8 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 f9a6592569..47713e9b42 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());
+}
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
