sijie closed pull request #2011: Cpp client: python client wrapper for reader
hasMessageAvaliable
URL: https://github.com/apache/incubator-pulsar/pull/2011
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/Reader.h
b/pulsar-client-cpp/include/pulsar/Reader.h
index 9ce9f44b0b..214a468d43 100644
--- a/pulsar-client-cpp/include/pulsar/Reader.h
+++ b/pulsar-client-cpp/include/pulsar/Reader.h
@@ -73,8 +73,14 @@ class Reader {
void closeAsync(ResultCallback callback);
+ /**
+ * Asynchronously check if there is any message available to read from the
current position.
+ */
void hasMessageAvailableAsync(HasMessageAvailableCallback callback);
+ /**
+ * Check if there is any message available to read from the current
position.
+ */
Result hasMessageAvailable(bool& hasMessageAvailable);
private:
diff --git a/pulsar-client-cpp/include/pulsar/c/reader.h
b/pulsar-client-cpp/include/pulsar/c/reader.h
index 72c02a6370..49f38fcade 100644
--- a/pulsar-client-cpp/include/pulsar/c/reader.h
+++ b/pulsar-client-cpp/include/pulsar/c/reader.h
@@ -66,6 +66,8 @@ void pulsar_reader_close_async(pulsar_reader_t *reader,
pulsar_result_callback c
void pulsar_reader_free(pulsar_reader_t *reader);
+pulsar_result pulsar_reader_has_message_available(pulsar_reader_t *reader, int
*available);
+
#pragma GCC visibility pop
#ifdef __cplusplus
diff --git a/pulsar-client-cpp/lib/c/c_Reader.cc
b/pulsar-client-cpp/lib/c/c_Reader.cc
index 3f7849def2..334c8616e9 100644
--- a/pulsar-client-cpp/lib/c/c_Reader.cc
+++ b/pulsar-client-cpp/lib/c/c_Reader.cc
@@ -52,3 +52,10 @@ void pulsar_reader_close_async(pulsar_reader_t *reader,
pulsar_result_callback c
}
void pulsar_reader_free(pulsar_reader_t *reader) { delete reader; }
+
+pulsar_result pulsar_reader_has_message_available(pulsar_reader_t *reader, int
*available) {
+ bool isAvailable;
+ pulsar_result result =
(pulsar_result)reader->reader.hasMessageAvailable(isAvailable);
+ *available = isAvailable;
+ return result;
+}
diff --git a/pulsar-client-cpp/python/pulsar/__init__.py
b/pulsar-client-cpp/python/pulsar/__init__.py
index 76b08cd7b7..434fb07a41 100644
--- a/pulsar-client-cpp/python/pulsar/__init__.py
+++ b/pulsar-client-cpp/python/pulsar/__init__.py
@@ -853,6 +853,12 @@ def read_next(self, timeout_millis=None):
_check_type(int, timeout_millis, 'timeout_millis')
return self._reader.read_next(timeout_millis)
+ def has_message_available(self):
+ """
+ Check if there is any message available to read from the current
position.
+ """
+ return self._reader.has_message_available();
+
def close(self):
"""
Close the reader.
diff --git a/pulsar-client-cpp/python/pulsar_test.py
b/pulsar-client-cpp/python/pulsar_test.py
index 2572557070..b07abbc998 100755
--- a/pulsar-client-cpp/python/pulsar_test.py
+++ b/pulsar-client-cpp/python/pulsar_test.py
@@ -488,6 +488,39 @@ def test_publish_compact_and_consume(self):
consumer2.close()
client.close()
+ def test_reader_has_message_available(self):
+ # create client, producer, reader
+ client = Client(self.serviceUrl)
+ producer =
client.create_producer('persistent://sample/standalone/ns/my-python-topic-reader-has-message-available')
+ reader =
client.create_reader('persistent://sample/standalone/ns/my-python-topic-reader-has-message-available',
+ MessageId.latest)
+
+ # before produce data, expected not has message available
+ self.assertFalse(reader.has_message_available());
+
+ for i in range(10):
+ producer.send('hello-%d' % i)
+
+ # produced data, expected has message available
+ self.assertTrue(reader.has_message_available());
+
+ for i in range(10):
+ msg = reader.read_next()
+ self.assertTrue(msg)
+ self.assertEqual(msg.data(), b'hello-%d' % i)
+
+ # consumed all data, expected not has message available
+ self.assertFalse(reader.has_message_available());
+
+ for i in range(10, 20):
+ producer.send('hello-%d' % i)
+
+ # produced data again, expected has message available
+ self.assertTrue(reader.has_message_available());
+ reader.close()
+ producer.close()
+ client.close()
+
def test_seek(self):
client = Client(self.serviceUrl)
consumer =
client.subscribe('persistent://sample/standalone/ns/my-python-topic-seek',
diff --git a/pulsar-client-cpp/python/src/reader.cc
b/pulsar-client-cpp/python/src/reader.cc
index 81aa78b3da..705c698482 100644
--- a/pulsar-client-cpp/python/src/reader.cc
+++ b/pulsar-client-cpp/python/src/reader.cc
@@ -57,6 +57,17 @@ Message Reader_readNextTimeout(Reader& reader, int
timeoutMs) {
return msg;
}
+bool Reader_hasMessageAvailable(Reader& reader) {
+ bool available = false;
+ Result res;
+ Py_BEGIN_ALLOW_THREADS
+ res = reader.hasMessageAvailable(available);
+ Py_END_ALLOW_THREADS
+
+ CHECK_RESULT(res);
+ return available;
+}
+
void Reader_close(Reader& reader) {
Result res;
Py_BEGIN_ALLOW_THREADS
@@ -73,6 +84,7 @@ void export_reader() {
.def("topic", &Reader::getTopic,
return_value_policy<copy_const_reference>())
.def("read_next", &Reader_readNext)
.def("read_next", &Reader_readNextTimeout)
+ .def("has_message_available", &Reader_hasMessageAvailable)
.def("close", &Reader_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
