diff --git a/pulsar-client-cpp/README.md b/pulsar-client-cpp/README.md index f5265577e458e..2af50a0ffafbe 100644 --- a/pulsar-client-cpp/README.md +++ b/pulsar-client-cpp/README.md @@ -280,7 +280,7 @@ ${PULSAR_PATH}/pulsar-test-service-stop.sh ## Requirements for Contributors -It's recommended to install [LLVM](https://llvm.org/builds/) for `clang-tidy` and `clang-format`. Pulsar C++ client use `clang-format` 6.0+ to format files. +It's required to install [LLVM](https://llvm.org/builds/) for `clang-tidy` and `clang-format`. Pulsar C++ client use `clang-format` 6.0+ to format files. `make format` automatically formats the files. Use `pulsar-client-cpp/docker-format.sh` to ensure the C++ sources are correctly formatted. diff --git a/pulsar-client-cpp/include/pulsar/c/consumer.h b/pulsar-client-cpp/include/pulsar/c/consumer.h index 03f80f3239479..37fd2acf5b8a6 100644 --- a/pulsar-client-cpp/include/pulsar/c/consumer.h +++ b/pulsar-client-cpp/include/pulsar/c/consumer.h @@ -236,6 +236,9 @@ PULSAR_PUBLIC pulsar_result pulsar_consumer_seek(pulsar_consumer_t *consumer, pu PULSAR_PUBLIC int pulsar_consumer_is_connected(pulsar_consumer_t *consumer); +PULSAR_PUBLIC pulsar_result pulsar_consumer_get_last_message_id(pulsar_consumer_t *consumer, + pulsar_message_id_t *messageId); + #ifdef __cplusplus } #endif diff --git a/pulsar-client-cpp/lib/c/c_Consumer.cc b/pulsar-client-cpp/lib/c/c_Consumer.cc index 9917e8cfad6e3..00d8311f13278 100644 --- a/pulsar-client-cpp/lib/c/c_Consumer.cc +++ b/pulsar-client-cpp/lib/c/c_Consumer.cc @@ -143,3 +143,8 @@ pulsar_result pulsar_consumer_seek(pulsar_consumer_t *consumer, pulsar_message_i } int pulsar_consumer_is_connected(pulsar_consumer_t *consumer) { return consumer->consumer.isConnected(); } + +pulsar_result pulsar_consumer_get_last_message_id(pulsar_consumer_t *consumer, + pulsar_message_id_t *messageId) { + return (pulsar_result)consumer->consumer.getLastMessageId(messageId->messageId); +} diff --git a/pulsar-client-cpp/python/pulsar/__init__.py b/pulsar-client-cpp/python/pulsar/__init__.py index e79955b57dd79..3832c3e69e23f 100644 --- a/pulsar-client-cpp/python/pulsar/__init__.py +++ b/pulsar-client-cpp/python/pulsar/__init__.py @@ -1253,7 +1253,12 @@ def is_connected(self): Check if the consumer is connected or not. """ return self._consumer.is_connected() - + + def get_last_message_id(self): + """ + Get the last message id. + """ + return self._consumer.get_last_message_id() class Reader: diff --git a/pulsar-client-cpp/python/pulsar_test.py b/pulsar-client-cpp/python/pulsar_test.py index dbdd6be59c7a6..127ecc4247cca 100755 --- a/pulsar-client-cpp/python/pulsar_test.py +++ b/pulsar-client-cpp/python/pulsar_test.py @@ -753,6 +753,18 @@ def test_reader_argument_errors(self): self._check_value_error(lambda: client.create_reader(topic, MessageId.earliest, reader_name=5)) client.close() + def test_get_last_message_id(self): + client = Client(self.serviceUrl) + consumer = client.subscribe( + "persistent://public/default/topic_name_test", "topic_name_test_sub", consumer_type=ConsumerType.Shared + ) + producer = client.create_producer("persistent://public/default/topic_name_test") + msg_id = producer.send(b"hello") + + msg = consumer.receive(TM) + self.assertEqual(msg.message_id(), msg_id) + client.close() + def test_publish_compact_and_consume(self): client = Client(self.serviceUrl) topic = "compaction_%s" % (uuid.uuid4()) diff --git a/pulsar-client-cpp/python/src/consumer.cc b/pulsar-client-cpp/python/src/consumer.cc index 10ffd07496f78..811ceb3ddf553 100644 --- a/pulsar-client-cpp/python/src/consumer.cc +++ b/pulsar-client-cpp/python/src/consumer.cc @@ -83,6 +83,16 @@ void Consumer_seek_timestamp(Consumer& consumer, uint64_t timestamp) { bool Consumer_is_connected(Consumer& consumer) { return consumer.isConnected(); } +MessageId Consumer_get_last_message_id(Consumer& consumer) { + MessageId msgId; + Result res; + Py_BEGIN_ALLOW_THREADS res = consumer.getLastMessageId(msgId); + Py_END_ALLOW_THREADS + + CHECK_RESULT(res); + return msgId; +} + void export_consumer() { using namespace boost::python; @@ -105,5 +115,6 @@ void export_consumer() { .def("redeliver_unacknowledged_messages", &Consumer::redeliverUnacknowledgedMessages) .def("seek", &Consumer_seek) .def("seek", &Consumer_seek_timestamp) - .def("is_connected", &Consumer_is_connected); + .def("is_connected", &Consumer_is_connected) + .def("get_last_message_id", &Consumer_get_last_message_id); }