Merge branch 'main' into bewaremypower/tableview

Author: BewareMyPower

pulsar/__init__.py

@@ -49,7 +49,7 @@
 
 from _pulsar import Result, CompressionType, ConsumerType, InitialPosition, PartitionsRoutingMode, BatchingType, \
     LoggerLevel, BatchReceivePolicy, KeySharedPolicy, KeySharedMode, ProducerAccessMode, RegexSubscriptionMode, \
-    DeadLetterPolicyBuilder  # noqa: F401
+    DeadLetterPolicyBuilder, ConsumerCryptoFailureAction  # noqa: F401
 
 from pulsar.__about__ import __version__
 
@@ -82,7 +82,7 @@ class MessageId:
     """
 
     def __init__(self, partition=-1, ledger_id=-1, entry_id=-1, batch_index=-1):
-        self._msg_id = _pulsar.MessageId(partition, ledger_id, entry_id, batch_index)
+        self._msg_id: _pulsar.MessageId = _pulsar.MessageId(partition, ledger_id, entry_id, batch_index)
 
     earliest = _pulsar.MessageId.earliest
     latest = _pulsar.MessageId.latest
@@ -112,6 +112,24 @@ def __str__(self) -> str:
         """
         return str(self._msg_id)
 
+    def __eq__(self, other) -> bool:
+        return self._msg_id == other._msg_id
+
+    def __ne__(self, other) -> bool:
+        return self._msg_id != other._msg_id
+
+    def __le__(self, other) -> bool:
+        return self._msg_id <= other._msg_id
+
+    def __lt__(self, other) -> bool:
+        return self._msg_id < other._msg_id
+
+    def __ge__(self, other) -> bool:
+        return self._msg_id >= other._msg_id
+
+    def __gt__(self, other) -> bool:
+        return self._msg_id > other._msg_id
+
     @staticmethod
     def deserialize(message_id_bytes):
         """
@@ -120,6 +138,14 @@ def deserialize(message_id_bytes):
         """
         return _pulsar.MessageId.deserialize(message_id_bytes)
 
+    @classmethod
+    def wrap(cls, msg_id: _pulsar.MessageId):
+        """
+        Wrap the underlying MessageId type from the C extension to the Python type.
+        """
+        self = cls()
+        self._msg_id = msg_id
+        return self
 
 class Message:
     """
@@ -171,9 +197,13 @@ def event_timestamp(self):
         """
         return self._message.event_timestamp()
 
-    def message_id(self):
+    def message_id(self) -> _pulsar.MessageId:
         """
         The message ID that can be used to refer to this particular message.
+
+        Returns
+        ----------
+        A `_pulsar.MessageId` object that represents where the message is persisted.
         """
         return self._message.message_id()
 
@@ -490,7 +520,9 @@ def __init__(self, service_url,
                  tls_validate_hostname=False,
                  logger=None,
                  connection_timeout_ms=10000,
-                 listener_name=None
+                 listener_name=None,
+                 tls_private_key_file_path: Optional[str] = None,
+                 tls_certificate_file_path: Optional[str] = None,
                  ):
         """
         Create a new Pulsar client instance.
@@ -556,6 +588,10 @@ def __init__(self, service_url,
             Listener name for lookup. Clients can use listenerName to choose one of the listeners as
             the service URL to create a connection to the broker as long as the network is accessible.
             ``advertisedListeners`` must be enabled in broker side.
+        tls_private_key_file_path: str, optional
+            The path to the TLS private key file
+        tls_certificate_file_path: str, optional
+            The path to the TLS certificate file.
         """
         _check_type(str, service_url, 'service_url')
         _check_type_or_none(Authentication, authentication, 'authentication')
@@ -571,6 +607,8 @@ def __init__(self, service_url,
         _check_type(bool, tls_allow_insecure_connection, 'tls_allow_insecure_connection')
         _check_type(bool, tls_validate_hostname, 'tls_validate_hostname')
         _check_type_or_none(str, listener_name, 'listener_name')
+        _check_type_or_none(str, tls_private_key_file_path, 'tls_private_key_file_path')
+        _check_type_or_none(str, tls_certificate_file_path, 'tls_certificate_file_path')
 
         conf = _pulsar.ClientConfiguration()
         if authentication:
@@ -602,6 +640,10 @@ def __init__(self, service_url,
             conf.tls_trust_certs_file_path(certifi.where())
         conf.tls_allow_insecure_connection(tls_allow_insecure_connection)
         conf.tls_validate_hostname(tls_validate_hostname)
+        if tls_private_key_file_path is not None:
+            conf.tls_private_key_file_path(tls_private_key_file_path)
+        if tls_certificate_file_path is not None:
+            conf.tls_certificate_file_path(tls_certificate_file_path)
         self._client = _pulsar.Client(service_url, conf)
         self._consumers = []
 
@@ -835,6 +877,7 @@ def subscribe(self, topic, subscription_name,
                   batch_index_ack_enabled=False,
                   regex_subscription_mode: RegexSubscriptionMode = RegexSubscriptionMode.PersistentOnly,
                   dead_letter_policy: Union[None, ConsumerDeadLetterPolicy] = None,
+                  crypto_failure_action: ConsumerCryptoFailureAction = ConsumerCryptoFailureAction.FAIL,
                   ):
         """
         Subscribe to the given topic and subscription combination.
@@ -938,6 +981,19 @@ def my_listener(consumer, message):
           stopped. By using the dead letter mechanism, messages have the max redelivery count, when they're
           exceeding the maximum number of redeliveries. Messages are sent to dead letter topics and acknowledged
           automatically.
+        crypto_failure_action: ConsumerCryptoFailureAction, default=ConsumerCryptoFailureAction.FAIL
+          Set the behavior when the decryption fails. The default is to fail the message.
+
+          Supported actions:
+
+          * ConsumerCryptoFailureAction.FAIL: Fail consume until crypto succeeds
+          * ConsumerCryptoFailureAction.DISCARD:
+            Message is silently acknowledged and not delivered to the application.
+          * ConsumerCryptoFailureAction.CONSUME:
+            Deliver the encrypted message to the application. It's the application's responsibility
+            to decrypt the message. If message is also compressed, decompression will fail. If the
+            message contains batch messages, client will not be able to retrieve individual messages
+            in the batch.
         """
         _check_type(str, subscription_name, 'subscription_name')
         _check_type(ConsumerType, consumer_type, 'consumer_type')
@@ -961,6 +1017,7 @@ def my_listener(consumer, message):
         _check_type_or_none(ConsumerKeySharedPolicy, key_shared_policy, 'key_shared_policy')
         _check_type(bool, batch_index_ack_enabled, 'batch_index_ack_enabled')
         _check_type(RegexSubscriptionMode, regex_subscription_mode, 'regex_subscription_mode')
+        _check_type(ConsumerCryptoFailureAction, crypto_failure_action, 'crypto_failure_action')
 
         conf = _pulsar.ConsumerConfiguration()
         conf.consumer_type(consumer_type)
@@ -999,6 +1056,7 @@ def my_listener(consumer, message):
         conf.batch_index_ack_enabled(batch_index_ack_enabled)
         if dead_letter_policy:
             conf.dead_letter_policy(dead_letter_policy.policy())
+        conf.crypto_failure_action(crypto_failure_action)
 
         c = Consumer()
         if isinstance(topic, str):
@@ -1027,7 +1085,8 @@ def create_reader(self, topic, start_message_id,
                       subscription_role_prefix=None,
                       is_read_compacted=False,
                       crypto_key_reader: Union[None, CryptoKeyReader] = None,
-                      start_message_id_inclusive=False
+                      start_message_id_inclusive=False,
+                      crypto_failure_action: ConsumerCryptoFailureAction = ConsumerCryptoFailureAction.FAIL,
                       ):
         """
         Create a reader on a particular topic
@@ -1088,6 +1147,19 @@ def my_listener(reader, message):
             and private key decryption messages for the consumer
         start_message_id_inclusive: bool, default=False
             Set the reader to include the startMessageId or given position of any reset operation like Reader.seek
+        crypto_failure_action: ConsumerCryptoFailureAction, default=ConsumerCryptoFailureAction.FAIL
+          Set the behavior when the decryption fails. The default is to fail the message.
+
+          Supported actions:
+
+          * ConsumerCryptoFailureAction.FAIL: Fail consume until crypto succeeds
+          * ConsumerCryptoFailureAction.DISCARD:
+            Message is silently acknowledged and not delivered to the application.
+          * ConsumerCryptoFailureAction.CONSUME:
+            Deliver the encrypted message to the application. It's the application's responsibility
+            to decrypt the message. If message is also compressed, decompression will fail. If the
+            message contains batch messages, client will not be able to retrieve individual messages
+            in the batch.
         """
 
         # If a pulsar.MessageId object is passed, access the _pulsar.MessageId object
@@ -1103,6 +1175,7 @@ def my_listener(reader, message):
         _check_type(bool, is_read_compacted, 'is_read_compacted')
         _check_type_or_none(CryptoKeyReader, crypto_key_reader, 'crypto_key_reader')
         _check_type(bool, start_message_id_inclusive, 'start_message_id_inclusive')
+        _check_type(ConsumerCryptoFailureAction, crypto_failure_action, 'crypto_failure_action')
 
         conf = _pulsar.ReaderConfiguration()
         if reader_listener:
@@ -1117,6 +1190,7 @@ def my_listener(reader, message):
         if crypto_key_reader:
             conf.crypto_key_reader(crypto_key_reader.cryptoKeyReader)
         conf.start_message_id_inclusive(start_message_id_inclusive)
+        conf.crypto_failure_action(crypto_failure_action)
 
         c = Reader()
         c._reader = self._client.create_reader(topic, start_message_id, conf)
@@ -1256,7 +1330,7 @@ def send(self, content,
              event_timestamp=None,
              deliver_at=None,
              deliver_after=None,
-             ):
+             ) -> _pulsar.MessageId:
         """
         Publish a message on the topic. Blocks until the message is acknowledged
 
@@ -1289,6 +1363,10 @@ def send(self, content,
             The timestamp is milliseconds and based on UTC
         deliver_after: optional
             Specify a delay in timedelta for the delivery of the messages.
+
+        Returns
+        ----------
+        A `_pulsar.MessageId` object that represents where the message is persisted.
         """
         msg = self._build_msg(content, properties, partition_key, ordering_key, sequence_id,
                               replication_clusters, disable_replication, event_timestamp,
@@ -1527,7 +1605,7 @@ def batch_receive(self):
             messages.append(m)
         return messages
 
-    def acknowledge(self, message):
+    def acknowledge(self, message: Union[Message, MessageId, _pulsar.Message, _pulsar.MessageId]):
         """
         Acknowledge the reception of a single message.
 
@@ -1536,7 +1614,7 @@ def acknowledge(self, message):
 
         Parameters
         ----------
-        message : Message, _pulsar.Message, _pulsar.MessageId
+        message : Message, MessageId, _pulsar.Message, _pulsar.MessageId
             The received message or message id.
 
         Raises
@@ -1546,10 +1624,12 @@ def acknowledge(self, message):
         """
         if isinstance(message, Message):
             self._consumer.acknowledge(message._message)
+        elif isinstance(message, MessageId):
+            self._consumer.acknowledge(message._msg_id)
         else:
             self._consumer.acknowledge(message)
 
-    def acknowledge_cumulative(self, message):
+    def acknowledge_cumulative(self, message: Union[Message, MessageId, _pulsar.Message, _pulsar.MessageId]):
         """
         Acknowledge the reception of all the messages in the stream up to (and
         including) the provided message.
@@ -1570,6 +1650,8 @@ def acknowledge_cumulative(self, message):
         """
         if isinstance(message, Message):
             self._consumer.acknowledge_cumulative(message._message)
+        elif isinstance(message, MessageId):
+            self._consumer.acknowledge_cumulative(message._msg_id)
         else:
             self._consumer.acknowledge_cumulative(message)
 

src/config.cc

@@ -166,6 +166,14 @@ void export_config(py::module_& m) {
              return_value_policy::copy)
         .def("tls_trust_certs_file_path", &ClientConfiguration::setTlsTrustCertsFilePath,
              return_value_policy::reference)
+        .def("tls_private_key_file_path", &ClientConfiguration::getTlsPrivateKeyFilePath,
+             return_value_policy::copy)
+        .def("tls_private_key_file_path", &ClientConfiguration::setTlsPrivateKeyFilePath,
+             return_value_policy::reference)
+        .def("tls_certificate_file_path", &ClientConfiguration::getTlsCertificateFilePath,
+             return_value_policy::copy)
+        .def("tls_certificate_file_path", &ClientConfiguration::setTlsCertificateFilePath,
+             return_value_policy::reference)
         .def("tls_allow_insecure_connection", &ClientConfiguration::isTlsAllowInsecureConnection)
         .def("tls_allow_insecure_connection", &ClientConfiguration::setTlsAllowInsecureConnection,
              return_value_policy::reference)
@@ -305,7 +313,11 @@ void export_config(py::module_& m) {
         .def("batch_index_ack_enabled", &ConsumerConfiguration::setBatchIndexAckEnabled,
              return_value_policy::reference)
         .def("dead_letter_policy", &ConsumerConfiguration::setDeadLetterPolicy)
-        .def("dead_letter_policy", &ConsumerConfiguration::getDeadLetterPolicy, return_value_policy::copy);
+        .def("dead_letter_policy", &ConsumerConfiguration::getDeadLetterPolicy, return_value_policy::copy)
+        .def("crypto_failure_action", &ConsumerConfiguration::getCryptoFailureAction,
+            return_value_policy::copy)
+        .def("crypto_failure_action", &ConsumerConfiguration::setCryptoFailureAction,
+            return_value_policy::reference);
 
     class_<ReaderConfiguration, std::shared_ptr<ReaderConfiguration>>(m, "ReaderConfiguration")
         .def(init<>())
@@ -323,5 +335,9 @@ void export_config(py::module_& m) {
         .def("read_compacted", &ReaderConfiguration::setReadCompacted)
         .def("crypto_key_reader", &ReaderConfiguration::setCryptoKeyReader, return_value_policy::reference)
         .def("start_message_id_inclusive", &ReaderConfiguration::isStartMessageIdInclusive)
-        .def("start_message_id_inclusive", &ReaderConfiguration::setStartMessageIdInclusive, return_value_policy::reference);
+        .def("start_message_id_inclusive", &ReaderConfiguration::setStartMessageIdInclusive, return_value_policy::reference)
+        .def("crypto_failure_action", &ReaderConfiguration::getCryptoFailureAction,
+            return_value_policy::copy)
+        .def("crypto_failure_action", &ReaderConfiguration::setCryptoFailureAction,
+            return_value_policy::reference);
 }

src/enums.cc

@@ -19,6 +19,7 @@
 #include "utils.h"
 #include <pulsar/CompressionType.h>
 #include <pulsar/ConsumerConfiguration.h>
+#include <pulsar/ConsumerCryptoFailureAction.h>
 #include <pulsar/ProducerConfiguration.h>
 #include <pulsar/KeySharedPolicy.h>
 #include <pybind11/pybind11.h>
@@ -140,4 +141,9 @@ void export_enums(py::module_& m) {
         .value("Info", Logger::LEVEL_INFO)
         .value("Warn", Logger::LEVEL_WARN)
         .value("Error", Logger::LEVEL_ERROR);
+    
+    enum_<ConsumerCryptoFailureAction>(m, "ConsumerCryptoFailureAction")
+        .value("FAIL", ConsumerCryptoFailureAction::FAIL)
+        .value("DISCARD", ConsumerCryptoFailureAction::DISCARD)
+        .value("CONSUME", ConsumerCryptoFailureAction::CONSUME);
 }