[incubator-pulsar] branch master updated: add consumer.seek in python client (#2008)

[mmerli]

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 a545c2e  add consumer.seek in python client (#2008)
a545c2e is described below

commit a545c2ef71fb1bc260a284596a3ab4283e9b528f
Author: Jia Zhai <zhaiji...@gmail.com>
AuthorDate: Fri Jun 22 00:33:48 2018 +0800

    add consumer.seek in python client (#2008)
---
 pulsar-client-cpp/python/pulsar/__init__.py | 14 ++++++++++++++
 pulsar-client-cpp/python/pulsar_test.py     | 22 ++++++++++++++++++++++
 pulsar-client-cpp/python/src/consumer.cc    | 10 ++++++++++
 3 files changed, 46 insertions(+)

diff --git a/pulsar-client-cpp/python/pulsar/__init__.py 
b/pulsar-client-cpp/python/pulsar/__init__.py
index 74d3011..76b08cd 100644
--- a/pulsar-client-cpp/python/pulsar/__init__.py
+++ b/pulsar-client-cpp/python/pulsar/__init__.py
@@ -801,6 +801,20 @@ class Consumer:
         """
         self._consumer.redeliver_unacknowledged_messages()
 
+    def seek(self, messageid):
+        """
+        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.
+
+        **Args**
+
+        * `message`:
+          The message id for seek.
+        """
+        self._consumer.seek(messageid)
+
     def close(self):
         """
         Close the consumer.
diff --git a/pulsar-client-cpp/python/pulsar_test.py 
b/pulsar-client-cpp/python/pulsar_test.py
index 8f80f44..2572557 100755
--- a/pulsar-client-cpp/python/pulsar_test.py
+++ b/pulsar-client-cpp/python/pulsar_test.py
@@ -488,6 +488,28 @@ class PulsarTest(TestCase):
         consumer2.close()
         client.close()
 
+    def test_seek(self):
+        client = Client(self.serviceUrl)
+        consumer = 
client.subscribe('persistent://sample/standalone/ns/my-python-topic-seek',
+                                    'my-sub',
+                                    consumer_type=ConsumerType.Shared)
+        producer = 
client.create_producer('persistent://sample/standalone/ns/my-python-topic-seek')
+
+        for i in range(100):
+            producer.send('hello-%d' % i)
+
+        for i in range(100):
+            msg = consumer.receive()
+            self.assertEqual(msg.data(), b'hello-%d' % i)
+            consumer.acknowledge(msg)
+
+        # seek, and after reconnect, expected receive first message.
+        consumer.seek(MessageId.earliest)
+        time.sleep(0.5)
+        msg = consumer.receive()
+        self.assertEqual(msg.data(), b'hello-0')
+        client.close()
+
     def _check_value_error(self, fun):
         try:
             fun()
diff --git a/pulsar-client-cpp/python/src/consumer.cc 
b/pulsar-client-cpp/python/src/consumer.cc
index 557536a..f4379f7 100644
--- a/pulsar-client-cpp/python/src/consumer.cc
+++ b/pulsar-client-cpp/python/src/consumer.cc
@@ -119,6 +119,15 @@ void Consumer_resumeMessageListener(Consumer& consumer) {
     CHECK_RESULT(consumer.resumeMessageListener());
 }
 
+void Consumer_seek(Consumer& consumer, const MessageId& msgId) {
+    Result res;
+    Py_BEGIN_ALLOW_THREADS
+    res = consumer.seek(msgId);
+    Py_END_ALLOW_THREADS
+
+    CHECK_RESULT(res);
+}
+
 void export_consumer() {
     using namespace boost::python;
 
@@ -137,5 +146,6 @@ void export_consumer() {
             .def("pause_message_listener", &Consumer_pauseMessageListener)
             .def("resume_message_listener", &Consumer_resumeMessageListener)
             .def("redeliver_unacknowledged_messages", 
&Consumer::redeliverUnacknowledgedMessages)
+            .def("seek", &Consumer_seek)
             ;
 }