diff --git a/pulsar/__init__.py b/pulsar/__init__.py index c1600f2..3deeb42 100644 --- a/pulsar/__init__.py +++ b/pulsar/__init__.py @@ -856,6 +856,7 @@ def __init__(self, service_url, listener_name=None, tls_private_key_file_path: Optional[str] = None, tls_certificate_file_path: Optional[str] = None, + keep_alive_interval_seconds: int = 30, ): """ Create a new Pulsar client instance. @@ -925,6 +926,11 @@ def __init__(self, service_url, The path to the TLS private key file tls_certificate_file_path: str, optional The path to the TLS certificate file. + keep_alive_interval_seconds: int, default=30 + Set the interval in seconds between keep-alive pings on each broker connection. A connection + that has received nothing from the broker since its last ping when the next one is due is + closed and reconnected. Raise it on slow or congested links, where the broker's replies can + wait behind queued messages for longer than this. """ if not isinstance(service_url, (str, AutoClusterFailover, ServiceInfoProvider)): raise ValueError( @@ -961,12 +967,16 @@ def __init__(self, service_url, _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') + _check_type(int, keep_alive_interval_seconds, 'keep_alive_interval_seconds') + if keep_alive_interval_seconds <= 0: + raise ValueError("Argument keep_alive_interval_seconds is expected to be positive") conf = _pulsar.ClientConfiguration() if authentication: conf.authentication(authentication.auth) conf.operation_timeout_seconds(operation_timeout_seconds) conf.connection_timeout(connection_timeout_ms) + conf.keep_alive_interval_in_seconds(keep_alive_interval_seconds) conf.io_threads(io_threads) conf.message_listener_threads(message_listener_threads) conf.concurrent_lookup_requests(concurrent_lookup_requests) diff --git a/src/config.cc b/src/config.cc index ec16b52..ef7c810 100644 --- a/src/config.cc +++ b/src/config.cc @@ -184,6 +184,9 @@ void export_config(py::module_& m) { return_value_policy::reference) .def("connection_timeout", &ClientConfiguration::getConnectionTimeout) .def("connection_timeout", &ClientConfiguration::setConnectionTimeout, return_value_policy::reference) + .def("keep_alive_interval_in_seconds", &ClientConfiguration::getKeepAliveIntervalInSeconds) + .def("keep_alive_interval_in_seconds", &ClientConfiguration::setKeepAliveIntervalInSeconds, + return_value_policy::reference) .def("io_threads", &ClientConfiguration::getIOThreads) .def("io_threads", &ClientConfiguration::setIOThreads, return_value_policy::reference) .def("message_listener_threads", &ClientConfiguration::getMessageListenerThreads) diff --git a/tests/pulsar_test.py b/tests/pulsar_test.py index 68eb499..d4d5b68 100755 --- a/tests/pulsar_test.py +++ b/tests/pulsar_test.py @@ -964,6 +964,8 @@ def test_client_argument_errors(self): self._check_value_error(lambda: Client(self.serviceUrl, use_tls="test")) self._check_value_error(lambda: Client(self.serviceUrl, tls_trust_certs_file_path=5)) self._check_value_error(lambda: Client(self.serviceUrl, tls_allow_insecure_connection="test")) + self._check_value_error(lambda: Client(self.serviceUrl, keep_alive_interval_seconds="test")) + self._check_value_error(lambda: Client(self.serviceUrl, keep_alive_interval_seconds=0)) def test_service_info_argument_errors(self): self._check_value_error(lambda: ServiceInfo(None))