From 40260d950362674778827eb7bb3d3c2a11eb7635 Mon Sep 17 00:00:00 2001 From: "Patrick M. Niedzielski" Date: Tue, 1 Sep 2026 15:18:35 -0400 Subject: [PATCH 01/26] Refactor: Extract `PyRef` for GIL-safe Python references `SessionEventHandler` hand-rolls reference counting for its three Python callbacks: three `Py_INCREF`s under a `GilAcquireGuard` in the constructor, and three matching `Py_DECREF`s in the destructor. As we intend to add an additional callback to provide authentication credentials, we will have to repeat this same structure wherever that is held and make sure to get the GIL right. This patch adds a new C++ class `pybmq::PyRef`, an owning reference type that acquires the GIL whenever it adjusts a reference count. Because copying and destroying it are GIL-safe, it can be held by objects the SDK copies on its own IO threads. By doing this, we no longer need to manually maintain the reference count with `Py_INCREF` and `Py_DECREF`s under the GIL. This patch also ports the callbacks held by `SessionEventHandler` to use it. These changes do result in a few more locks and unlocks of the GIL, but this only happens once while constructing and once while destructing a `Session`, so the downside is small. Signed-off-by: Patrick M. Niedzielski --- src/cpp/pybmq_pyref.h | 114 ++++++++++++++++++++++++++ src/cpp/pybmq_sessioneventhandler.cpp | 22 ++--- src/cpp/pybmq_sessioneventhandler.h | 10 +-- 3 files changed, 124 insertions(+), 22 deletions(-) create mode 100644 src/cpp/pybmq_pyref.h diff --git a/src/cpp/pybmq_pyref.h b/src/cpp/pybmq_pyref.h new file mode 100644 index 0000000..b817032 --- /dev/null +++ b/src/cpp/pybmq_pyref.h @@ -0,0 +1,114 @@ +// Copyright 2019-2026 Bloomberg Finance L.P. +// SPDX-License-Identifier: Apache-2.0 +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +#ifndef INCLUDED_PYBMQ_PYREF +#define INCLUDED_PYBMQ_PYREF + +#define PY_SSIZE_T_CLEAN +#include + +#include + +namespace BloombergLP { +namespace pybmq { + +// =========== +// class PyRef +// =========== + +/// A value-semantic owning reference to a Python object. +/// +/// Copying or destroying a `PyRef` adjusts the referent's reference count and +/// acquires the GIL to do so, so instances may be copied and destroyed on +/// threads that do not hold it. This is what the BlazingMQ SDK does to +/// callbacks it has been handed, on its own IO threads. +class PyRef +{ + private: + // DATA + PyObject* d_object_p; + + public: + // CREATORS + PyRef(); + + /// Create a reference to the specified `object`, which may be 0. + explicit PyRef(PyObject* object); + + PyRef(const PyRef& other); + + ~PyRef(); + + // MANIPULATORS + PyRef& operator=(const PyRef& rhs); + + // ACCESSORS + + /// Return the referent, or 0 if this reference is empty. The reference + /// count is not adjusted; the caller borrows the returned pointer. + PyObject* get() const; +}; + +// =========================================================================== +// INLINE DEFINITIONS +// =========================================================================== + +inline PyRef::PyRef() +: d_object_p(0) +{ +} + +inline PyRef::PyRef(PyObject* object) +: d_object_p(object) +{ + GilAcquireGuard guard; + Py_XINCREF(d_object_p); +} + +inline PyRef::PyRef(const PyRef& other) +: d_object_p(other.d_object_p) +{ + GilAcquireGuard guard; + Py_XINCREF(d_object_p); +} + +inline PyRef::~PyRef() +{ + GilAcquireGuard guard; + Py_XDECREF(d_object_p); +} + +inline PyRef& +PyRef::operator=(const PyRef& rhs) +{ + if (this != &rhs) { + GilAcquireGuard guard; + Py_XINCREF(rhs.d_object_p); + Py_XDECREF(d_object_p); + d_object_p = rhs.d_object_p; + } + return *this; +} + +inline PyObject* +PyRef::get() const +{ + return d_object_p; +} + +} // namespace pybmq +} // namespace BloombergLP + +#endif diff --git a/src/cpp/pybmq_sessioneventhandler.cpp b/src/cpp/pybmq_sessioneventhandler.cpp index b5d270d..90d0668 100644 --- a/src/cpp/pybmq_sessioneventhandler.cpp +++ b/src/cpp/pybmq_sessioneventhandler.cpp @@ -39,18 +39,6 @@ SessionEventHandler::SessionEventHandler( , d_py_message_event_callback(py_message_event_callback) , d_py_ack_event_callback(py_ack_event_callback) { - GilAcquireGuard guard; - Py_INCREF(d_py_session_event_callback); - Py_INCREF(d_py_message_event_callback); - Py_INCREF(d_py_ack_event_callback); -} - -SessionEventHandler::~SessionEventHandler() -{ - GilAcquireGuard guard; - Py_DECREF(d_py_ack_event_callback); - Py_DECREF(d_py_message_event_callback); - Py_DECREF(d_py_session_event_callback); } void @@ -67,7 +55,7 @@ SessionEventHandler::onSessionEvent(const bmqa::SessionEvent& event) } bslma::ManagedPtr rv = RefUtils::toManagedPtr(PyObject_CallFunction( - d_py_session_event_callback, + d_py_session_event_callback.get(), "(N (i N i N s#))", PyBytes_FromStringAndSize( event.errorDescription().c_str(), @@ -92,16 +80,16 @@ SessionEventHandler::onMessageEvent(const bmqa::MessageEvent& event) PyObject* py_event; if (event.type() == bmqt::MessageEventType::e_PUSH) { - callback = d_py_message_event_callback; - py_event = MessageUtils::get_messages(event, d_py_session_event_callback); + callback = d_py_message_event_callback.get(); + py_event = MessageUtils::get_messages(event, d_py_session_event_callback.get()); } else if (event.type() == bmqt::MessageEventType::e_ACK) { - callback = d_py_ack_event_callback; + callback = d_py_ack_event_callback.get(); py_event = MessageUtils::get_acks(event); } else { bsl::ostringstream oss; oss << "Received an unexpected message event of type " << (int)event.type() << " (" << event.type() << ")"; - callback = d_py_session_event_callback; + callback = d_py_session_event_callback.get(); py_event = PyBytes_FromString(oss.str().c_str()); } bslma::ManagedPtr rv = diff --git a/src/cpp/pybmq_sessioneventhandler.h b/src/cpp/pybmq_sessioneventhandler.h index 3acb3b3..dd0c6f1 100644 --- a/src/cpp/pybmq_sessioneventhandler.h +++ b/src/cpp/pybmq_sessioneventhandler.h @@ -22,6 +22,8 @@ #include #include +#include + #include namespace BloombergLP { @@ -30,9 +32,9 @@ namespace pybmq { class SessionEventHandler : public bmqa::SessionEventHandler { private: - PyObject* d_py_session_event_callback; - PyObject* d_py_message_event_callback; - PyObject* d_py_ack_event_callback; + PyRef d_py_session_event_callback; + PyRef d_py_message_event_callback; + PyRef d_py_ack_event_callback; public: SessionEventHandler( @@ -40,8 +42,6 @@ class SessionEventHandler : public bmqa::SessionEventHandler PyObject* py_message_event_callback, PyObject* py_ack_event_callback); - ~SessionEventHandler(); - void onSessionEvent(const bmqa::SessionEvent& event) BSLS_KEYWORD_OVERRIDE; void onMessageEvent(const bmqa::MessageEvent& event) BSLS_KEYWORD_OVERRIDE; }; From a98d486720985e3c318b9474f4b9e1cacec41375 Mon Sep 17 00:00:00 2001 From: "Patrick M. Niedzielski" Date: Tue, 1 Sep 2026 15:20:42 -0400 Subject: [PATCH 02/26] Refactor: Collapse duplicated `cls(...)` call in `with_options` This patch collapses two branches in `with_options`, which differed only in whether `DEFAULT_TIMEOUT` or `session_options.timeouts` was passed to a function call, and duplicated the other eleven. Signed-off-by: Patrick M. Niedzielski --- src/blazingmq/_session.py | 45 +++++++++++++++------------------------ 1 file changed, 17 insertions(+), 28 deletions(-) diff --git a/src/blazingmq/_session.py b/src/blazingmq/_session.py index e0c09e3..e1433d5 100644 --- a/src/blazingmq/_session.py +++ b/src/blazingmq/_session.py @@ -503,34 +503,23 @@ def with_options( if message_compression_algorithm is None: message_compression_algorithm = CompressionAlgorithmType.NONE - if session_options.timeouts is None: - return cls( - on_session_event, - on_message, - broker, - message_compression_algorithm, - DEFAULT_TIMEOUT, - session_options.host_health_monitor, - session_options.num_processing_threads, - session_options.blob_buffer_size, - session_options.channel_high_watermark, - session_options.event_queue_watermarks, - session_options.stats_dump_interval, - ) - else: - return cls( - on_session_event, - on_message, - broker, - message_compression_algorithm, - session_options.timeouts, - session_options.host_health_monitor, - session_options.num_processing_threads, - session_options.blob_buffer_size, - session_options.channel_high_watermark, - session_options.event_queue_watermarks, - session_options.stats_dump_interval, - ) + timeout: Union[Timeouts, float] = DEFAULT_TIMEOUT + if session_options.timeouts is not None: + timeout = session_options.timeouts + + return cls( + on_session_event, + on_message, + broker, + message_compression_algorithm, + timeout, + session_options.host_health_monitor, + session_options.num_processing_threads, + session_options.blob_buffer_size, + session_options.channel_high_watermark, + session_options.event_queue_watermarks, + session_options.stats_dump_interval, + ) def open_queue( self, From 5a1e748f0f51f8ea9bf6461b87835190e40b3018 Mon Sep 17 00:00:00 2001 From: "Patrick M. Niedzielski" Date: Tue, 1 Sep 2026 15:25:16 -0400 Subject: [PATCH 03/26] Refactor: Forward to `Session` by keyword in `with_options` `with_options` passed eleven arguments positionally, so inserting a parameter anywhere in `Session.__init__` would silently rebind every argument after it rather than failing. This patch changes the call to pass all optional arguments by keyword. Signed-off-by: Patrick M. Niedzielski --- src/blazingmq/_session.py | 20 ++++++++++---------- 1 file changed, 10 insertions(+), 10 deletions(-) diff --git a/src/blazingmq/_session.py b/src/blazingmq/_session.py index e1433d5..6d126f5 100644 --- a/src/blazingmq/_session.py +++ b/src/blazingmq/_session.py @@ -509,16 +509,16 @@ def with_options( return cls( on_session_event, - on_message, - broker, - message_compression_algorithm, - timeout, - session_options.host_health_monitor, - session_options.num_processing_threads, - session_options.blob_buffer_size, - session_options.channel_high_watermark, - session_options.event_queue_watermarks, - session_options.stats_dump_interval, + on_message=on_message, + broker=broker, + message_compression_algorithm=message_compression_algorithm, + timeout=timeout, + host_health_monitor=session_options.host_health_monitor, + num_processing_threads=session_options.num_processing_threads, + blob_buffer_size=session_options.blob_buffer_size, + channel_high_watermark=session_options.channel_high_watermark, + event_queue_watermarks=session_options.event_queue_watermarks, + stats_dump_interval=session_options.stats_dump_interval, ) def open_queue( From 27c5bf00dcdc4066c670af795464016f6131962f Mon Sep 17 00:00:00 2001 From: "Patrick M. Niedzielski" Date: Tue, 1 Sep 2026 15:29:18 -0400 Subject: [PATCH 04/26] Refactor: Bundle `pybmq::Session` options into `SessionConfig` MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The constructor to `pybmq::Session` has been getting a little unwieldy, growing to 21 parameters. Originally, passing parameters directly to `pybmq::Session` like this resulted in the simplest and most transparent code in the Cython layer (which is the layer that is hardest to debug) and meant we didn’t need to worry too much about the reference counting of Python objects being marshalled through Cython. But, with so many parameters now, this is more of a liability than a benefit. This patch adds a new struct `pybmq::SessionConfig`, which holds the fourteen plain option values. The Python callbacks, host health monitor and exception types stay as constructor parameters, so no Python object lives in the struct for reference counting ease. Adding this struct allows us to simplify the constructor for `pybmq::Session` down from taking 21 arguments to taking only 8. Signed-off-by: Patrick M. Niedzielski --- src/blazingmq/_ext.pyx | 34 +++++++------ src/cpp/pybmq_session.cpp | 69 +++++++++++---------------- src/cpp/pybmq_session.h | 17 ++----- src/cpp/pybmq_sessionconfig.h | 89 +++++++++++++++++++++++++++++++++++ src/declarations/pybmq.pxd | 34 +++++++------ 5 files changed, 160 insertions(+), 83 deletions(-) create mode 100644 src/cpp/pybmq_sessionconfig.h diff --git a/src/blazingmq/_ext.pyx b/src/blazingmq/_ext.pyx index e8ecb9d..e424e06 100644 --- a/src/blazingmq/_ext.pyx +++ b/src/blazingmq/_ext.pyx @@ -37,6 +37,7 @@ from bmq.bmqt cimport k_DEFAULT_MAX_UNCONFIRMED_MESSAGES from bmq.bmqt cimport k_DEFAULT_SUSPENDS_ON_BAD_HOST_HEALTH from pybmq cimport BallUtil from pybmq cimport Session as NativeSession +from pybmq cimport SessionConfig from typing import Optional @@ -221,24 +222,29 @@ cdef class Session: cdef char *c_broker_uri = broker script_name = _script_name.get_script_name() cdef char *c_script_name = script_name + + cdef SessionConfig config + config.broker_uri = c_broker_uri + config.script_name = c_script_name + config.message_compression_type = ( + COMPRESSION_ALGO_FROM_PY_MAPPING[message_compression_algorithm]) + config.num_processing_threads = c_num_processing_threads + config.blob_buffer_size = c_blob_buffer_size + config.channel_high_watermark = c_channel_high_watermark + config.event_queue_watermarks = c_event_queue_watermarks + config.stats_dump_interval = c_stats_dump_interval + config.connect_timeout = c_connect_timeout + config.disconnect_timeout = c_disconnect_timeout + config.open_queue_timeout = c_open_queue_timeout + config.configure_queue_timeout = c_configure_queue_timeout + config.close_queue_timeout = c_close_queue_timeout + config.monitor_host_health = monitor_host_health + self._session = new NativeSession( session_cb, message_cb, ack_cb, - c_broker_uri, - c_script_name, - COMPRESSION_ALGO_FROM_PY_MAPPING[message_compression_algorithm], - c_num_processing_threads, - c_blob_buffer_size, - c_channel_high_watermark, - c_event_queue_watermarks, - c_stats_dump_interval, - c_connect_timeout, - c_disconnect_timeout, - c_open_queue_timeout, - c_configure_queue_timeout, - c_close_queue_timeout, - monitor_host_health, + config, fake_host_health_monitor_sp, Error, BrokerTimeoutError, diff --git a/src/cpp/pybmq_session.cpp b/src/cpp/pybmq_session.cpp index acab799..55491a0 100644 --- a/src/cpp/pybmq_session.cpp +++ b/src/cpp/pybmq_session.cpp @@ -77,20 +77,7 @@ Session::Session( PyObject* py_session_event_callback, PyObject* py_message_event_callback, PyObject* py_ack_event_callback, - const char* broker_uri, - const char* script_name, - bmqt::CompressionAlgorithmType::Enum message_compression_type, - bsl::optional num_processing_threads, - bsl::optional blob_buffer_size, - bsl::optional channel_high_watermark, - bsl::optional > event_queue_watermarks, - const bsls::TimeInterval& stats_dump_interval, - const bsls::TimeInterval& connect_timeout, - const bsls::TimeInterval& disconnect_timeout, - const bsls::TimeInterval& open_queue_timeout, - const bsls::TimeInterval& configure_queue_timeout, - const bsls::TimeInterval& close_queue_timeout, - bool monitor_host_health, + const SessionConfig& config, bsl::shared_ptr fake_host_health_monitor_sp, PyObject* error, PyObject* broker_timeout_error, @@ -106,66 +93,66 @@ Session::Session( if (fake_host_health_monitor_sp) { host_health_monitor_sp = fake_host_health_monitor_sp; - } else if (monitor_host_health) { + } else if (config.monitor_host_health) { } - if (message_compression_type + if (config.message_compression_type < bmqt::CompressionAlgorithmType::k_LOWEST_SUPPORTED_TYPE - || message_compression_type + || config.message_compression_type > bmqt::CompressionAlgorithmType::k_HIGHEST_SUPPORTED_TYPE) { PyErr_SetString(PyExc_ValueError, "Invalid message compression type"); throw bsl::runtime_error("propagating Python error"); } - d_message_compression_type = message_compression_type; + d_message_compression_type = config.message_compression_type; { pybmq::GilReleaseGuard guard; bmqt::SessionOptions options; - options.setBrokerUri(broker_uri) - .setProcessNameOverride(script_name) + options.setBrokerUri(config.broker_uri) + .setProcessNameOverride(config.script_name) .setHostHealthMonitor(host_health_monitor_sp); - if (num_processing_threads.has_value()) { - options.setNumProcessingThreads(num_processing_threads.value()); + if (config.num_processing_threads.has_value()) { + options.setNumProcessingThreads(config.num_processing_threads.value()); } - if (blob_buffer_size.has_value()) { - options.setBlobBufferSize(blob_buffer_size.value()); + if (config.blob_buffer_size.has_value()) { + options.setBlobBufferSize(config.blob_buffer_size.value()); } - if (channel_high_watermark.has_value()) { - options.setChannelHighWatermark(channel_high_watermark.value()); + if (config.channel_high_watermark.has_value()) { + options.setChannelHighWatermark(config.channel_high_watermark.value()); } - if (event_queue_watermarks.has_value()) { + if (config.event_queue_watermarks.has_value()) { options.configureEventQueue( - event_queue_watermarks.value().first, - event_queue_watermarks.value().second); + config.event_queue_watermarks.value().first, + config.event_queue_watermarks.value().second); } - if (stats_dump_interval != bsls::TimeInterval()) { - options.setStatsDumpInterval(stats_dump_interval); + if (config.stats_dump_interval != bsls::TimeInterval()) { + options.setStatsDumpInterval(config.stats_dump_interval); } - if (connect_timeout != bsls::TimeInterval()) { - options.setConnectTimeout(connect_timeout); + if (config.connect_timeout != bsls::TimeInterval()) { + options.setConnectTimeout(config.connect_timeout); } - if (disconnect_timeout != bsls::TimeInterval()) { - options.setDisconnectTimeout(disconnect_timeout); + if (config.disconnect_timeout != bsls::TimeInterval()) { + options.setDisconnectTimeout(config.disconnect_timeout); } - if (open_queue_timeout != bsls::TimeInterval()) { - options.setOpenQueueTimeout(open_queue_timeout); + if (config.open_queue_timeout != bsls::TimeInterval()) { + options.setOpenQueueTimeout(config.open_queue_timeout); } - if (configure_queue_timeout != bsls::TimeInterval()) { - options.setConfigureQueueTimeout(configure_queue_timeout); + if (config.configure_queue_timeout != bsls::TimeInterval()) { + options.setConfigureQueueTimeout(config.configure_queue_timeout); } - if (close_queue_timeout != bsls::TimeInterval()) { - options.setCloseQueueTimeout(close_queue_timeout); + if (config.close_queue_timeout != bsls::TimeInterval()) { + options.setCloseQueueTimeout(config.close_queue_timeout); } bslma::ManagedPtr handler( diff --git a/src/cpp/pybmq_session.h b/src/cpp/pybmq_session.h index f37a407..52acc5b 100644 --- a/src/cpp/pybmq_session.h +++ b/src/cpp/pybmq_session.h @@ -19,6 +19,8 @@ #define PY_SSIZE_T_CLEAN #include +#include + #include #include #include @@ -51,20 +53,7 @@ class Session Session(PyObject* py_session_event_callback, PyObject* py_message_event_callback, PyObject* py_ack_event_callback, - const char* broker_uri, - const char* script_name, - bmqt::CompressionAlgorithmType::Enum message_compression_type, - bsl::optional num_processing_threads, - bsl::optional blob_buffer_size, - bsl::optional channel_high_watermark, - bsl::optional > event_queue_watermarks, - const bsls::TimeInterval& stats_dump_interval, - const bsls::TimeInterval& connect_timeout, - const bsls::TimeInterval& disconnect_timeout, - const bsls::TimeInterval& open_queue_timeout, - const bsls::TimeInterval& configure_queue_timeout, - const bsls::TimeInterval& close_queue_timeout, - bool monitor_host_health, + const SessionConfig& config, bsl::shared_ptr fake_host_health_monitor, PyObject* d_error, PyObject* d_broker_timeout_error, diff --git a/src/cpp/pybmq_sessionconfig.h b/src/cpp/pybmq_sessionconfig.h new file mode 100644 index 0000000..2f936c5 --- /dev/null +++ b/src/cpp/pybmq_sessionconfig.h @@ -0,0 +1,89 @@ +// Copyright 2019-2026 Bloomberg Finance L.P. +// SPDX-License-Identifier: Apache-2.0 +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +#ifndef INCLUDED_PYBMQ_SESSIONCONFIG +#define INCLUDED_PYBMQ_SESSIONCONFIG + +#include + +#include +#include +#include + +namespace BloombergLP { +namespace pybmq { + +// =================== +// struct SessionConfig +// =================== + +/// The options a `pybmq::Session` is constructed with, other than its Python +/// callbacks and host health monitor. +/// +/// A default-constructed `SessionConfig` means "no option set": empty +/// optionals, and default-constructed `bsls::TimeInterval`s, which +/// `Session` treats as unset. A new option can therefore be added as a +/// field without changing any existing call site. +/// +/// `broker_uri` and `script_name` are held, not owned, and must outlive the +/// `Session` constructor call. +struct SessionConfig +{ + // PUBLIC DATA + const char* broker_uri; + const char* script_name; + bmqt::CompressionAlgorithmType::Enum message_compression_type; + bsl::optional num_processing_threads; + bsl::optional blob_buffer_size; + bsl::optional channel_high_watermark; + bsl::optional > event_queue_watermarks; + bsls::TimeInterval stats_dump_interval; + bsls::TimeInterval connect_timeout; + bsls::TimeInterval disconnect_timeout; + bsls::TimeInterval open_queue_timeout; + bsls::TimeInterval configure_queue_timeout; + bsls::TimeInterval close_queue_timeout; + bool monitor_host_health; + + // CREATORS + SessionConfig(); +}; + +// =========================================================================== +// INLINE DEFINITIONS +// =========================================================================== + +inline SessionConfig::SessionConfig() +: broker_uri(0) +, script_name(0) +, message_compression_type(bmqt::CompressionAlgorithmType::e_NONE) +, num_processing_threads() +, blob_buffer_size() +, channel_high_watermark() +, event_queue_watermarks() +, stats_dump_interval() +, connect_timeout() +, disconnect_timeout() +, open_queue_timeout() +, configure_queue_timeout() +, close_queue_timeout() +, monitor_host_health(false) +{ +} + +} // namespace pybmq +} // namespace BloombergLP + +#endif diff --git a/src/declarations/pybmq.pxd b/src/declarations/pybmq.pxd index 9ba293e..c0cd5c5 100644 --- a/src/declarations/pybmq.pxd +++ b/src/declarations/pybmq.pxd @@ -34,25 +34,31 @@ cdef extern from 'pybmq_ballutil.h' namespace 'BloombergLP::pybmq': @staticmethod object shutDownBallSingleton() except + +cdef extern from "pybmq_sessionconfig.h" namespace "BloombergLP::pybmq" nogil: + cdef cppclass SessionConfig: + SessionConfig() except+ + + const char* broker_uri + const char* script_name + CompressionAlgorithmType message_compression_type + optional[int] num_processing_threads + optional[int] blob_buffer_size + optional[int] channel_high_watermark + optional[pair[int, int]] event_queue_watermarks + TimeInterval stats_dump_interval + TimeInterval connect_timeout + TimeInterval disconnect_timeout + TimeInterval open_queue_timeout + TimeInterval configure_queue_timeout + TimeInterval close_queue_timeout + bint monitor_host_health + cdef extern from "pybmq_session.h" namespace "BloombergLP::pybmq" nogil: cdef cppclass Session: Session(object on_session_event, object on_message_event, object on_ack_event, - const char* broker_uri, - const char* script_name, - CompressionAlgorithmType message_compression_algorithm, - optional[int] num_processing_threads, - optional[int] blob_buffer_size, - optional[int] channel_high_watermark, - optional[pair[int, int]] event_queue_watermarks, - TimeInterval stats_dump_interval, - TimeInterval connect_timeout, - TimeInterval disconnect_timeout, - TimeInterval open_queue_timeout, - TimeInterval configure_queue_timeout, - TimeInterval close_queue_timeout, - bint monitor_host_health, + const SessionConfig& config, shared_ptr[ManualHostHealthMonitor] fake_host_health_monitor_sp, object error, object broker_timeout_error, From dd888c05f18dd160f8591fcfaf7544d4f789db1a Mon Sep 17 00:00:00 2001 From: "Patrick M. Niedzielski" Date: Fri, 4 Sep 2026 15:40:02 -0400 Subject: [PATCH 05/26] CI: Bump pinned BlazingMQ tag to v0.95.20 CI failed to compile `pybmq_session.cpp`: `AuthnCredentialCbFunctor` is called with no arguments, but `bmqt::SessionOptions::AuthnCredentialCb` at the pinned `v0.95.14` is still bsl::function(bsl::ostream& error)> The `ostream& error` parameter was removed upstream in bloomberg/blazingmq@c1614448a1c9356200e22cfa59cf623ff187dc0c ("Refactor: User authentication credentials callback (#1571)"), first released in v0.95.15. Our code was written against that signature, not the one the pinned tag actually provides. `bmqt_authncredential.h` (the `AuthnCredential` value type itself) is unchanged between v0.95.14 and v0.95.20, and no other commit in that range touches `SessionOptions`'s public surface except an internal allocator fix (c0f272850), so nothing else in this branch needs to change for the bump. Bump to v0.95.20, the latest tag, rather than the minimal v0.95.15, since CMakeLists.txt is untouched across the whole range and BDE_TAG/NTF_CORE_TAG need no corresponding change. --- bin/clone-dependencies.sh | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/bin/clone-dependencies.sh b/bin/clone-dependencies.sh index 10260a8..1949ecb 100755 --- a/bin/clone-dependencies.sh +++ b/bin/clone-dependencies.sh @@ -15,7 +15,7 @@ set -u BDE_TOOLS_TAG=4.38.0.0 BDE_TAG=4.38.0.0 NTF_CORE_TAG=2.6.12 -BLAZINGMQ_TAG=v0.95.14 +BLAZINGMQ_TAG=v0.95.20 if [ ! -d "${DIR_THIRDPARTY}/bde-tools" ]; then From f89e5155340e84cbe17dde7cc83da396645a6c14 Mon Sep 17 00:00:00 2001 From: Emelia Lei Date: Mon, 8 Dec 2025 17:25:10 -0500 Subject: [PATCH 06/26] provide a way to create credential Signed-off-by: Emelia Lei --- src/blazingmq/_ext.pyi | 4 ++ src/blazingmq/_ext.pyx | 35 ++++++++++++++++ src/blazingmq/_session.py | 22 ++++++++++ src/cpp/pybmq_session.cpp | 79 ++++++++++++++++++++++++++++++++++- src/cpp/pybmq_session.h | 6 +++ src/declarations/bmq/bmqt.pxd | 10 +++++ src/declarations/pybmq.pxd | 1 + 7 files changed, 155 insertions(+), 2 deletions(-) diff --git a/src/blazingmq/_ext.pyi b/src/blazingmq/_ext.pyi index 14d2478..8fa7b41 100644 --- a/src/blazingmq/_ext.pyi +++ b/src/blazingmq/_ext.pyi @@ -37,6 +37,9 @@ class FakeHostHealthMonitor: def set_healthy(self) -> None: ... def set_unhealthy(self) -> None: ... +class FakeAuthnCredentialCb: + def __init__(self, callback: Callable[[], Optional[tuple[str, bytes]]]) -> None: ... + class Session: def __init__( self, @@ -53,6 +56,7 @@ class Session: timeouts: Timeouts = Timeouts(), monitor_host_health: bool = False, fake_host_health_monitor: Optional[FakeHostHealthMonitor] = None, + fake_authn_credential_cb: Optional[FakeAuthnCredentialCb] = None, ) -> None: ... def stop(self) -> None: ... def open_queue_sync( diff --git a/src/blazingmq/_ext.pyx b/src/blazingmq/_ext.pyx index e424e06..ab4aa0c 100644 --- a/src/blazingmq/_ext.pyx +++ b/src/blazingmq/_ext.pyx @@ -21,12 +21,15 @@ import weakref from bsl cimport optional from bsl cimport pair from bsl cimport shared_ptr +from bsl cimport vector +from bsl cimport string from bsl.bsls cimport TimeInterval from cpython.ceval cimport PyEval_InitThreads from libcpp cimport bool as cppbool from bmq.bmqa cimport ManualHostHealthMonitor from bmq.bmqt cimport AckResult +from bmq.bmqt cimport AuthnCredential from bmq.bmqt cimport CompressionAlgorithmType from bmq.bmqt cimport HostHealthState from bmq.bmqt cimport PropertyType @@ -154,6 +157,36 @@ cdef class FakeHostHealthMonitor: self._monitor.get().setState(HostHealthState.e_UNHEALTHY) +cdef class FakeAuthnCredentialCb: + cdef object _callback # Store the Python callable + + def __cinit__(self, callback): + self._callback = callback + + # This method will be called by C++ code via PyObject_CallMethod + # Returns None for no credential, or (mechanism, data) tuple + def get_credential_data(self): + try: + result = self._callback() + if result is None: + return None + + if not isinstance(result, tuple) or len(result) != 2: + raise ValueError("callback must return (str, bytes) or None") + + mechanism, data = result + if not isinstance(mechanism, str) or not isinstance(data, bytes): + raise ValueError("callback must return (str, bytes) or None") + + # Return as-is, let C++ side handle conversion + return result + + except Exception: + # Log error or handle as needed + LOGGER.exception("Error in authentication credential callback") + return None + + cdef class Session: cdef object __weakref__ cdef NativeSession* _session @@ -175,6 +208,7 @@ cdef class Session: timeouts: _timeouts.Timeouts = _timeouts.Timeouts(), monitor_host_health: bool = False, fake_host_health_monitor: FakeHostHealthMonitor = None, + fake_authn_credential_cb: FakeAuthnCredentialCb = None, _mock: Optional[object] = None, ) -> None: cdef shared_ptr[ManualHostHealthMonitor] fake_host_health_monitor_sp @@ -244,6 +278,7 @@ cdef class Session: session_cb, message_cb, ack_cb, + fake_authn_credential_cb, config, fake_host_health_monitor_sp, Error, diff --git a/src/blazingmq/_session.py b/src/blazingmq/_session.py index 6d126f5..a87809f 100644 --- a/src/blazingmq/_session.py +++ b/src/blazingmq/_session.py @@ -35,6 +35,7 @@ from ._messages import Message from ._messages import MessageHandle from ._monitors import BasicHealthMonitor +from ._ext import FakeAuthnCredentialCb from ._timeouts import Timeouts from ._typing import PropertyTypeDict from ._typing import PropertyValueDict @@ -265,6 +266,11 @@ class SessionOptions: healthy, `.HostUnhealthy` and `.HostHealthRestored` events with never be emitted, and the *suspends_on_bad_host_health* option of `QueueOptions` cannot be used. + authn_credential_provider: + An optional callable that returns authentication credentials as a + ``(mechanism, data)`` tuple of ``(str, bytes)``, or ``None`` if no + credentials are available. If not provided, no authentication + credentials are sent to the broker. num_processing_threads: The number of threads for the SDK to use for processing events. This defaults to 1. @@ -295,6 +301,7 @@ def __init__( message_compression_algorithm: Optional[CompressionAlgorithmType] = None, timeouts: Optional[Timeouts] = None, host_health_monitor: Union[BasicHealthMonitor, None] = (DefaultMonitor()), + authn_credential_provider: Optional[Callable] = (DefaultAuthnCredentialProvider()), num_processing_threads: Optional[int] = None, blob_buffer_size: Optional[int] = None, channel_high_watermark: Optional[int] = None, @@ -304,6 +311,7 @@ def __init__( self.message_compression_algorithm = message_compression_algorithm self.timeouts = timeouts self.host_health_monitor = host_health_monitor + self.authn_credential_provider = authn_credential_provider self.num_processing_threads = num_processing_threads self.blob_buffer_size = blob_buffer_size self.channel_high_watermark = channel_high_watermark @@ -317,6 +325,7 @@ def __eq__(self, other: object) -> bool: self.message_compression_algorithm == other.message_compression_algorithm and self.timeouts == other.timeouts and self.host_health_monitor == other.host_health_monitor + and self.authn_credential_provider == other.authn_credential_provider and self.num_processing_threads == other.num_processing_threads and self.blob_buffer_size == other.blob_buffer_size and self.channel_high_watermark == other.channel_high_watermark @@ -332,6 +341,7 @@ def __repr__(self) -> str: "message_compression_algorithm", "timeouts", "host_health_monitor", + "authn_credential_provider", "num_processing_threads", "blob_buffer_size", "channel_high_watermark", @@ -379,6 +389,10 @@ class Session: `.HostHealthRestored` events will never be emitted, and the *suspends_on_bad_host_health* option of `QueueOptions` cannot be used. + authn_credential_provider: an optional callable that returns authentication + credentials as a ``(mechanism, data)`` tuple of ``(str, bytes)``, + or ``None`` if no credentials are available. If not provided, no + authentication credentials are sent to the broker. num_processing_threads: The number of threads for the SDK to use for processing events. This defaults to 1. blob_buffer_size: The size (in bytes) of the blob buffers to use. This @@ -418,6 +432,7 @@ def __init__( ), timeout: Union[Timeouts, float] = DEFAULT_TIMEOUT, host_health_monitor: Union[BasicHealthMonitor, None] = (DefaultMonitor()), + authn_credential_provider: Optional[Callable] = None, num_processing_threads: Optional[int] = None, blob_buffer_size: Optional[int] = None, channel_high_watermark: Optional[int] = None, @@ -433,6 +448,11 @@ def __init__( monitor_host_health = host_health_monitor is not None fake_host_health_monitor = getattr(host_health_monitor, "_monitor", None) + fake_authn_credential_provider = ( + FakeAuthnCredentialCb(authn_credential_provider) + if authn_credential_provider is not None + else None + ) self._has_no_on_message = on_message is None @@ -459,6 +479,7 @@ def __init__( timeouts=_validate_timeouts(timeout), monitor_host_health=monitor_host_health, fake_host_health_monitor=fake_host_health_monitor, + fake_authn_credential_cb=fake_authn_credential_provider, ) self._ext.set_owned_by_session() @@ -514,6 +535,7 @@ def with_options( message_compression_algorithm=message_compression_algorithm, timeout=timeout, host_health_monitor=session_options.host_health_monitor, + authn_credential_provider=session_options.authn_credential_provider, num_processing_threads=session_options.num_processing_threads, blob_buffer_size=session_options.blob_buffer_size, channel_high_watermark=session_options.channel_high_watermark, diff --git a/src/cpp/pybmq_session.cpp b/src/cpp/pybmq_session.cpp index 55491a0..09879ea 100644 --- a/src/cpp/pybmq_session.cpp +++ b/src/cpp/pybmq_session.cpp @@ -15,6 +15,7 @@ #include +#include #include #include #include @@ -77,6 +78,7 @@ Session::Session( PyObject* py_session_event_callback, PyObject* py_message_event_callback, PyObject* py_ack_event_callback, + PyObject* fake_authn_credential_cb, const SessionConfig& config, bsl::shared_ptr fake_host_health_monitor_sp, PyObject* error, @@ -106,6 +108,74 @@ Session::Session( } d_message_compression_type = config.message_compression_type; + + AuthnCredentialCb cpp_callback; + bool has_auth_callback = false; + + if (fake_authn_credential_cb != nullptr && fake_authn_credential_cb != Py_None) { + // Increment reference count since we're storing the Python object + Py_INCREF(fake_authn_credential_cb); + has_auth_callback = true; + + // Create a C++ lambda that wraps the Python callback + cpp_callback = + [fake_authn_credential_cb]( + bsl::ostream& error) -> bsl::optional { + pybmq::GilAcquireGuard guard; + + // Call get_credential_data() method on the Python object + bslma::ManagedPtr result = + RefUtils::toManagedPtr(PyObject_CallMethod( + fake_authn_credential_cb, + "get_credential_data", + nullptr)); + + if (!result) { + // Python exception occurred + PyErr_Print(); + error << "Error calling get_credential_data()"; + return bsl::optional(); + } + + if (result.get() == Py_None) { + return bsl::optional(); + } + + // Extract tuple (mechanism, data) + if (!PyTuple_Check(result.get()) || PyTuple_Size(result.get()) != 2) { + error << "get_credential_data() must return (str, bytes) or None"; + return bsl::optional(); + } + + PyObject* mechanism_obj = PyTuple_GetItem(result.get(), 0); + PyObject* data_obj = PyTuple_GetItem(result.get(), 1); + + if (!PyUnicode_Check(mechanism_obj) || !PyBytes_Check(data_obj)) { + error << "get_credential_data() must return (str, bytes) or None"; + return bsl::optional(); + } + + // Convert Python str to C++ string + const char* mechanism_cstr = PyUnicode_AsUTF8(mechanism_obj); + bsl::string mechanism(mechanism_cstr); + + // Convert Python bytes to vector + char* data_ptr; + Py_ssize_t data_len; + PyBytes_AsStringAndSize(data_obj, &data_ptr, &data_len); + bsl::vector data(data_ptr, data_ptr + data_len); + + // Construct and return AuthnCredential + bmqt::AuthnCredential credential; + credential.setMechanism(mechanism).setData(data); + + // Move credential into optional (AuthnCredential is move-only) + bsl::optional opt_credential; + opt_credential.emplace(bslmf::MovableRefUtil::move(credential)); + return opt_credential; + }; + } + { pybmq::GilReleaseGuard guard; bmqt::SessionOptions options; @@ -131,6 +201,11 @@ Session::Session( config.event_queue_watermarks.value().second); } + if (has_auth_callback) { + // TODO: This will only compile with setAuthnCredentialCb in SessionOptions + options.setAuthnCredentialCb(cpp_callback); + } + if (config.stats_dump_interval != bsls::TimeInterval()) { options.setStatsDumpInterval(config.stats_dump_interval); } @@ -516,8 +591,8 @@ Session::post( oss << "Failed to post message to " << queue_uri << " queue: " << post_rc; throw GenericError(oss.str()); } - // We have a successful post and the SDK now owns the `on_ack` callback object - // so release our reference without a DECREF. + // We have a successful post and the SDK now owns the `on_ack` callback + // object so release our reference without a DECREF. managed_on_ack.release(); } catch (const GenericError& exc) { PyErr_SetString(d_error, exc.what()); diff --git a/src/cpp/pybmq_session.h b/src/cpp/pybmq_session.h index 52acc5b..55b6344 100644 --- a/src/cpp/pybmq_session.h +++ b/src/cpp/pybmq_session.h @@ -23,6 +23,7 @@ #include #include +#include #include #include @@ -49,10 +50,15 @@ class Session Session(const Session&); Session& operator=(const Session&); + // TODO: Remove this once it's added in SessionOptions + typedef bsl::function(bsl::ostream& error)> + AuthnCredentialCb; + public: Session(PyObject* py_session_event_callback, PyObject* py_message_event_callback, PyObject* py_ack_event_callback, + PyObject* fake_authn_credential_cb, const SessionConfig& config, bsl::shared_ptr fake_host_health_monitor, PyObject* d_error, diff --git a/src/declarations/bmq/bmqt.pxd b/src/declarations/bmq/bmqt.pxd index 07e27e9..01b5bf7 100644 --- a/src/declarations/bmq/bmqt.pxd +++ b/src/declarations/bmq/bmqt.pxd @@ -14,6 +14,8 @@ # limitations under the License. from libcpp cimport bool +from bsl cimport string +from bsl cimport vector cdef extern from "bmqt_sessioneventtype.h" namespace "BloombergLP::bmqt::SessionEventType" nogil: @@ -73,3 +75,11 @@ cdef extern from "bmqt_queueoptions.h" namespace "BloombergLP::bmqt::QueueOption int k_DEFAULT_MAX_UNCONFIRMED_BYTES int k_DEFAULT_CONSUMER_PRIORITY bool k_DEFAULT_SUSPENDS_ON_BAD_HOST_HEALTH + +cdef extern from "bmqt_authncredential.h" namespace "BloombergLP::bmqt" nogil: + cdef cppclass AuthnCredential: + AuthnCredential() except + + AuthnCredential& setMechanism(const string&) except + + AuthnCredential& setData(const vector[char]&) except + + const string& mechanism() const + const vector[char]& data() const diff --git a/src/declarations/pybmq.pxd b/src/declarations/pybmq.pxd index c0cd5c5..bdd4b75 100644 --- a/src/declarations/pybmq.pxd +++ b/src/declarations/pybmq.pxd @@ -58,6 +58,7 @@ cdef extern from "pybmq_session.h" namespace "BloombergLP::pybmq" nogil: Session(object on_session_event, object on_message_event, object on_ack_event, + object fake_authn_credential_cb, const SessionConfig& config, shared_ptr[ManualHostHealthMonitor] fake_host_health_monitor_sp, object error, From da1268570774875c9cb5224571674ea3017479f1 Mon Sep 17 00:00:00 2001 From: "Patrick M. Niedzielski" Date: Mon, 22 Jun 2026 15:49:20 -0400 Subject: [PATCH 07/26] picking this up again Signed-off-by: Patrick M. Niedzielski --- src/cpp/pybmq_session.cpp | 2 +- src/cpp/pybmq_session.h | 4 ---- 2 files changed, 1 insertion(+), 5 deletions(-) diff --git a/src/cpp/pybmq_session.cpp b/src/cpp/pybmq_session.cpp index 09879ea..b4d32d8 100644 --- a/src/cpp/pybmq_session.cpp +++ b/src/cpp/pybmq_session.cpp @@ -118,6 +118,7 @@ Session::Session( has_auth_callback = true; // Create a C++ lambda that wraps the Python callback + // TODO this can't be a lambda cpp_callback = [fake_authn_credential_cb]( bsl::ostream& error) -> bsl::optional { @@ -202,7 +203,6 @@ Session::Session( } if (has_auth_callback) { - // TODO: This will only compile with setAuthnCredentialCb in SessionOptions options.setAuthnCredentialCb(cpp_callback); } diff --git a/src/cpp/pybmq_session.h b/src/cpp/pybmq_session.h index 55b6344..5802013 100644 --- a/src/cpp/pybmq_session.h +++ b/src/cpp/pybmq_session.h @@ -50,10 +50,6 @@ class Session Session(const Session&); Session& operator=(const Session&); - // TODO: Remove this once it's added in SessionOptions - typedef bsl::function(bsl::ostream& error)> - AuthnCredentialCb; - public: Session(PyObject* py_session_event_callback, PyObject* py_message_event_callback, From 1b6f65225f3bda965a2c505614d45cc75cd029c7 Mon Sep 17 00:00:00 2001 From: "Patrick M. Niedzielski" Date: Tue, 23 Jun 2026 17:51:26 -0400 Subject: [PATCH 08/26] Provide `DefaultAuthnCredentialCb` Right now, this defaults to `None` (i.e, no authentication). Signed-off-by: Patrick M. Niedzielski --- src/blazingmq/_session.py | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/src/blazingmq/_session.py b/src/blazingmq/_session.py index a87809f..85d6629 100644 --- a/src/blazingmq/_session.py +++ b/src/blazingmq/_session.py @@ -53,6 +53,10 @@ def DefaultMonitor() -> Union[BasicHealthMonitor, None]: return None +def DefaultAuthnCredentialCb() -> Optional[Callable]: + return None + + DEFAULT_TIMEOUT = DefaultTimeoutType() KNOWN_MONITORS = ("blazingmq.BasicHealthMonitor",) @@ -432,7 +436,7 @@ def __init__( ), timeout: Union[Timeouts, float] = DEFAULT_TIMEOUT, host_health_monitor: Union[BasicHealthMonitor, None] = (DefaultMonitor()), - authn_credential_provider: Optional[Callable] = None, + authn_credential_provider: Optional[Callable] = (DefaultAuthnCredentialCb()), num_processing_threads: Optional[int] = None, blob_buffer_size: Optional[int] = None, channel_high_watermark: Optional[int] = None, From af63f476eb715347c0bc85612aa9c8838a29ed62 Mon Sep 17 00:00:00 2001 From: "Patrick M. Niedzielski" Date: Fri, 26 Jun 2026 12:13:38 -0400 Subject: [PATCH 09/26] Fix: Compile error from AuthnCredential API Signed-off-by: Patrick M. Niedzielski --- src/cpp/pybmq_session.cpp | 12 +++++------- 1 file changed, 5 insertions(+), 7 deletions(-) diff --git a/src/cpp/pybmq_session.cpp b/src/cpp/pybmq_session.cpp index b4d32d8..32c8a38 100644 --- a/src/cpp/pybmq_session.cpp +++ b/src/cpp/pybmq_session.cpp @@ -166,13 +166,11 @@ Session::Session( PyBytes_AsStringAndSize(data_obj, &data_ptr, &data_len); bsl::vector data(data_ptr, data_ptr + data_len); - // Construct and return AuthnCredential - bmqt::AuthnCredential credential; - credential.setMechanism(mechanism).setData(data); - - // Move credential into optional (AuthnCredential is move-only) - bsl::optional opt_credential; - opt_credential.emplace(bslmf::MovableRefUtil::move(credential)); + // Construct and move credential into optional + // (AuthnCredential is move-only) + bmqt::AuthnCredential credential(mechanism, data); + bsl::optional opt_credential( + bslmf::MovableRefUtil::move(credential)); return opt_credential; }; } From 77430f625a8400755ad7c4131cf4ee2a91922340 Mon Sep 17 00:00:00 2001 From: "Patrick M. Niedzielski" Date: Fri, 26 Jun 2026 12:20:05 -0400 Subject: [PATCH 10/26] Fix: Fully qualify `AuthnCredentialCb` Signed-off-by: Patrick M. Niedzielski --- src/cpp/pybmq_session.cpp | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/cpp/pybmq_session.cpp b/src/cpp/pybmq_session.cpp index 32c8a38..8801cf2 100644 --- a/src/cpp/pybmq_session.cpp +++ b/src/cpp/pybmq_session.cpp @@ -109,7 +109,7 @@ Session::Session( d_message_compression_type = config.message_compression_type; - AuthnCredentialCb cpp_callback; + bmqt::SessionOptions::AuthnCredentialCb cpp_callback; bool has_auth_callback = false; if (fake_authn_credential_cb != nullptr && fake_authn_credential_cb != Py_None) { From 46a260956b7c6743e53cd4e5f1f012988417e75f Mon Sep 17 00:00:00 2001 From: "Patrick M. Niedzielski" Date: Fri, 26 Jun 2026 15:41:21 -0400 Subject: [PATCH 11/26] Fix: Add `fake_authn_credential_cb` value to failing tests Signed-off-by: Patrick M. Niedzielski --- tests/unit/test_session.py | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/tests/unit/test_session.py b/tests/unit/test_session.py index 78fba1d..28ec586 100644 --- a/tests/unit/test_session.py +++ b/tests/unit/test_session.py @@ -78,6 +78,7 @@ def dummy2(): ), monitor_host_health=False, fake_host_health_monitor=None, + fake_authn_credential_cb=None, ) @@ -128,6 +129,7 @@ def dummy2(): timeouts=timeouts, monitor_host_health=False, fake_host_health_monitor=None, + fake_authn_credential_cb=None, ) @@ -172,6 +174,7 @@ def dummy2(): timeouts=timeouts, monitor_host_health=False, fake_host_health_monitor=None, + fake_authn_credential_cb=None, ) @@ -207,6 +210,7 @@ def dummy2(): timeouts=Timeouts(), monitor_host_health=False, fake_host_health_monitor=None, + fake_authn_credential_cb=None, ) @@ -259,6 +263,7 @@ def dummy2(): timeouts=timeouts, monitor_host_health=False, fake_host_health_monitor=None, + fake_authn_credential_cb=None, ) @@ -304,6 +309,7 @@ def dummy2(): ), monitor_host_health=True, fake_host_health_monitor=monitor._monitor, + fake_authn_credential_cb=None, ) @@ -335,6 +341,7 @@ def dummy2(): timeouts=Timeouts(), monitor_host_health=False, fake_host_health_monitor=None, + fake_authn_credential_cb=None, ) From 66670717584fecc6ead906f091b92ce3341af036 Mon Sep 17 00:00:00 2001 From: "Patrick M. Niedzielski" Date: Fri, 26 Jun 2026 15:45:09 -0400 Subject: [PATCH 12/26] Fix: Test `authn_credential_provider` in `SessionOptions` Signed-off-by: Patrick M. Niedzielski --- tests/unit/test_session_options.py | 14 ++++++++++++++ 1 file changed, 14 insertions(+) diff --git a/tests/unit/test_session_options.py b/tests/unit/test_session_options.py index 6adc505..0f04fbf 100644 --- a/tests/unit/test_session_options.py +++ b/tests/unit/test_session_options.py @@ -58,6 +58,7 @@ def test_session_options_default_to_none(): assert options.message_compression_algorithm is None assert options.timeouts is None assert options.host_health_monitor is None + assert options.authn_credential_provider is None assert options.num_processing_threads is None assert options.blob_buffer_size is None assert options.channel_high_watermark is None @@ -92,6 +93,7 @@ def test_session_options_equality(): blazingmq.SessionOptions(channel_high_watermark=8000000), blazingmq.SessionOptions(event_queue_watermarks=(6000000, 7000000)), blazingmq.SessionOptions(stats_dump_interval=30.0), + blazingmq.SessionOptions(authn_credential_provider=lambda: None), ], ) def test_queue_options_other_inequality(right): @@ -100,3 +102,15 @@ def test_queue_options_other_inequality(right): # THEN assert not left == right + + +def test_session_options_repr_with_authn_credential_provider(): + # GIVEN + def my_provider(): + return ("mechanism", b"data") + + # WHEN + options = blazingmq.SessionOptions(authn_credential_provider=my_provider) + + # THEN + assert "authn_credential_provider=" in repr(options) From 7d5eefb68179ab562722174f457cf6c66df48faf Mon Sep 17 00:00:00 2001 From: "Patrick M. Niedzielski" Date: Fri, 26 Jun 2026 15:54:03 -0400 Subject: [PATCH 13/26] Test: Add tests for `ExtSession` construction Signed-off-by: Patrick M. Niedzielski --- tests/unit/test_session.py | 143 +++++++++++++++++++++++++++++++++++++ 1 file changed, 143 insertions(+) diff --git a/tests/unit/test_session.py b/tests/unit/test_session.py index 28ec586..7e0332d 100644 --- a/tests/unit/test_session.py +++ b/tests/unit/test_session.py @@ -82,6 +82,56 @@ def dummy2(): ) +@mock.patch("blazingmq._session.ExtSession") +def test_session_constructed_with_authn_credential_provider(ext_cls): + # GIVEN + ext_cls.mock_add_spec([]) + + def dummy1(): + pass + + def dummy2(): + pass + + def my_provider(): + return ("mechanism", b"data") + + # WHEN + Session( + dummy1, + on_message=dummy2, + broker="some_uri", + timeout=60.0, + host_health_monitor=None, + authn_credential_provider=my_provider, + ) + + # THEN + ext_cls.assert_called_once_with( + dummy1, + on_message=dummy2, + broker=b"some_uri", + message_compression_algorithm=CompressionAlgorithmType.NONE, + num_processing_threads=None, + blob_buffer_size=None, + channel_high_watermark=None, + event_queue_watermarks=None, + stats_dump_interval=None, + timeouts=Timeouts( + connect_timeout=None, + disconnect_timeout=None, + open_queue_timeout=60.0, + configure_queue_timeout=60.0, + close_queue_timeout=60.0, + ), + monitor_host_health=False, + fake_host_health_monitor=None, + fake_authn_credential_cb=mock.ANY, + ) + call_kwargs = ext_cls.call_args[1] + assert call_kwargs["fake_authn_credential_cb"] is not None + + @mock.patch("blazingmq._session.ExtSession") def test_session_constructed_with_timeouts(ext_cls): # GIVEN @@ -267,6 +317,99 @@ def dummy2(): ) +@mock.patch("blazingmq._session.ExtSession") +def test_session_default_with_options_authn_credential_provider(ext_cls): + # GIVEN + ext_cls.mock_add_spec([]) + + def dummy1(): + pass + + def dummy2(): + pass + + def my_provider(): + return ("mechanism", b"data") + + session_options = SessionOptions(authn_credential_provider=my_provider) + + # WHEN + Session.with_options( + dummy1, on_message=dummy2, broker="some_uri", session_options=session_options + ) + + # THEN + ext_cls.assert_called_once_with( + dummy1, + on_message=dummy2, + broker=b"some_uri", + message_compression_algorithm=CompressionAlgorithmType.NONE, + num_processing_threads=None, + blob_buffer_size=None, + channel_high_watermark=None, + event_queue_watermarks=None, + stats_dump_interval=None, + timeouts=Timeouts(), + monitor_host_health=False, + fake_host_health_monitor=None, + fake_authn_credential_cb=mock.ANY, + ) + call_kwargs = ext_cls.call_args[1] + assert call_kwargs["fake_authn_credential_cb"] is not None + + +@mock.patch("blazingmq._session.ExtSession") +def test_session_with_options_authn_credential_provider(ext_cls): + # GIVEN + ext_cls.mock_add_spec([]) + + def dummy1(): + pass + + def dummy2(): + pass + + def my_provider(): + return ("mechanism", b"data") + + timeouts = Timeouts( + connect_timeout=60.0, + disconnect_timeout=70.0, + open_queue_timeout=80.0, + configure_queue_timeout=90.0, + close_queue_timeout=100.0, + ) + + session_options = SessionOptions( + timeouts=timeouts, + authn_credential_provider=my_provider, + ) + + # WHEN + Session.with_options( + dummy1, on_message=dummy2, broker="some_uri", session_options=session_options + ) + + # THEN + ext_cls.assert_called_once_with( + dummy1, + on_message=dummy2, + broker=b"some_uri", + message_compression_algorithm=CompressionAlgorithmType.NONE, + num_processing_threads=None, + blob_buffer_size=None, + channel_high_watermark=None, + event_queue_watermarks=None, + stats_dump_interval=None, + timeouts=timeouts, + monitor_host_health=False, + fake_host_health_monitor=None, + fake_authn_credential_cb=mock.ANY, + ) + call_kwargs = ext_cls.call_args[1] + assert call_kwargs["fake_authn_credential_cb"] is not None + + @mock.patch("blazingmq._session.ExtSession") def test_session_basic_monitor(ext_cls): # GIVEN From e1d84cc77f82db77de9043ebb8940727d77d2a95 Mon Sep 17 00:00:00 2001 From: "Patrick M. Niedzielski" Date: Fri, 26 Jun 2026 15:55:11 -0400 Subject: [PATCH 14/26] Fix: Format `DefaultAuthnCredentialProvider` Signed-off-by: Patrick M. Niedzielski --- src/blazingmq/_session.py | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/src/blazingmq/_session.py b/src/blazingmq/_session.py index 85d6629..3ed5f5f 100644 --- a/src/blazingmq/_session.py +++ b/src/blazingmq/_session.py @@ -305,7 +305,9 @@ def __init__( message_compression_algorithm: Optional[CompressionAlgorithmType] = None, timeouts: Optional[Timeouts] = None, host_health_monitor: Union[BasicHealthMonitor, None] = (DefaultMonitor()), - authn_credential_provider: Optional[Callable] = (DefaultAuthnCredentialProvider()), + authn_credential_provider: Optional[Callable] = ( + DefaultAuthnCredentialProvider() + ), num_processing_threads: Optional[int] = None, blob_buffer_size: Optional[int] = None, channel_high_watermark: Optional[int] = None, @@ -436,7 +438,9 @@ def __init__( ), timeout: Union[Timeouts, float] = DEFAULT_TIMEOUT, host_health_monitor: Union[BasicHealthMonitor, None] = (DefaultMonitor()), - authn_credential_provider: Optional[Callable] = (DefaultAuthnCredentialCb()), + authn_credential_provider: Optional[Callable] = ( + DefaultAuthnCredentialCb() + ), num_processing_threads: Optional[int] = None, blob_buffer_size: Optional[int] = None, channel_high_watermark: Optional[int] = None, From e15b7f47d923fa2785b893bb6b741268a3848a50 Mon Sep 17 00:00:00 2001 From: "Patrick M. Niedzielski" Date: Fri, 26 Jun 2026 16:09:53 -0400 Subject: [PATCH 15/26] Fix: Update `AuthnCredential` in bmqt.pxd Signed-off-by: Patrick M. Niedzielski --- src/declarations/bmq/bmqt.pxd | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/src/declarations/bmq/bmqt.pxd b/src/declarations/bmq/bmqt.pxd index 01b5bf7..5f05d12 100644 --- a/src/declarations/bmq/bmqt.pxd +++ b/src/declarations/bmq/bmqt.pxd @@ -1,4 +1,4 @@ -# Copyright 2019-2023 Bloomberg Finance L.P. +# Copyright 2019-2026 Bloomberg Finance L.P. # SPDX-License-Identifier: Apache-2.0 # # Licensed under the Apache License, Version 2.0 (the "License"); @@ -13,9 +13,9 @@ # See the License for the specific language governing permissions and # limitations under the License. -from libcpp cimport bool from bsl cimport string from bsl cimport vector +from libcpp cimport bool cdef extern from "bmqt_sessioneventtype.h" namespace "BloombergLP::bmqt::SessionEventType" nogil: @@ -79,7 +79,7 @@ cdef extern from "bmqt_queueoptions.h" namespace "BloombergLP::bmqt::QueueOption cdef extern from "bmqt_authncredential.h" namespace "BloombergLP::bmqt" nogil: cdef cppclass AuthnCredential: AuthnCredential() except + - AuthnCredential& setMechanism(const string&) except + - AuthnCredential& setData(const vector[char]&) except + + AuthnCredential(const AuthnCredential&) except + + AuthnCredential(const string& mechanism, const vector[char]& data) except + const string& mechanism() const const vector[char]& data() const From c31f4457c5b8af0976ffc7bcd0114da807252d9e Mon Sep 17 00:00:00 2001 From: "Patrick M. Niedzielski" Date: Tue, 30 Jun 2026 14:37:26 -0400 Subject: [PATCH 16/26] Rename `FakeAuthnCredentialCb` Signed-off-by: Patrick M. Niedzielski --- src/blazingmq/_ext.pyi | 4 +- src/blazingmq/_ext.pyx | 6 +- src/blazingmq/_session.py | 8 +- src/cpp/pybmq_session.cpp | 10 +- src/cpp/pybmq_session.h | 2 +- src/declarations/pybmq.pxd | 2 +- .../unit/test_authn_credential_cb_adapter.py | 114 ++++++++++++++++++ tests/unit/test_session.py | 26 ++-- 8 files changed, 143 insertions(+), 29 deletions(-) create mode 100644 tests/unit/test_authn_credential_cb_adapter.py diff --git a/src/blazingmq/_ext.pyi b/src/blazingmq/_ext.pyi index 8fa7b41..7714270 100644 --- a/src/blazingmq/_ext.pyi +++ b/src/blazingmq/_ext.pyi @@ -37,7 +37,7 @@ class FakeHostHealthMonitor: def set_healthy(self) -> None: ... def set_unhealthy(self) -> None: ... -class FakeAuthnCredentialCb: +class AuthnCredentialCbAdapter: def __init__(self, callback: Callable[[], Optional[tuple[str, bytes]]]) -> None: ... class Session: @@ -56,7 +56,7 @@ class Session: timeouts: Timeouts = Timeouts(), monitor_host_health: bool = False, fake_host_health_monitor: Optional[FakeHostHealthMonitor] = None, - fake_authn_credential_cb: Optional[FakeAuthnCredentialCb] = None, + authn_credential_cb: Optional[AuthnCredentialCbAdapter] = None, ) -> None: ... def stop(self) -> None: ... def open_queue_sync( diff --git a/src/blazingmq/_ext.pyx b/src/blazingmq/_ext.pyx index ab4aa0c..74283fe 100644 --- a/src/blazingmq/_ext.pyx +++ b/src/blazingmq/_ext.pyx @@ -157,7 +157,7 @@ cdef class FakeHostHealthMonitor: self._monitor.get().setState(HostHealthState.e_UNHEALTHY) -cdef class FakeAuthnCredentialCb: +cdef class AuthnCredentialCbAdapter: cdef object _callback # Store the Python callable def __cinit__(self, callback): @@ -208,7 +208,7 @@ cdef class Session: timeouts: _timeouts.Timeouts = _timeouts.Timeouts(), monitor_host_health: bool = False, fake_host_health_monitor: FakeHostHealthMonitor = None, - fake_authn_credential_cb: FakeAuthnCredentialCb = None, + authn_credential_cb: AuthnCredentialCbAdapter = None, _mock: Optional[object] = None, ) -> None: cdef shared_ptr[ManualHostHealthMonitor] fake_host_health_monitor_sp @@ -278,7 +278,7 @@ cdef class Session: session_cb, message_cb, ack_cb, - fake_authn_credential_cb, + authn_credential_cb, config, fake_host_health_monitor_sp, Error, diff --git a/src/blazingmq/_session.py b/src/blazingmq/_session.py index 3ed5f5f..abc5e0a 100644 --- a/src/blazingmq/_session.py +++ b/src/blazingmq/_session.py @@ -25,6 +25,7 @@ from . import _six as six from ._enums import CompressionAlgorithmType from ._enums import PropertyType +from ._ext import AuthnCredentialCbAdapter from ._ext import DEFAULT_CONSUMER_PRIORITY from ._ext import DEFAULT_MAX_UNCONFIRMED_BYTES from ._ext import DEFAULT_MAX_UNCONFIRMED_MESSAGES @@ -35,7 +36,6 @@ from ._messages import Message from ._messages import MessageHandle from ._monitors import BasicHealthMonitor -from ._ext import FakeAuthnCredentialCb from ._timeouts import Timeouts from ._typing import PropertyTypeDict from ._typing import PropertyValueDict @@ -456,8 +456,8 @@ def __init__( monitor_host_health = host_health_monitor is not None fake_host_health_monitor = getattr(host_health_monitor, "_monitor", None) - fake_authn_credential_provider = ( - FakeAuthnCredentialCb(authn_credential_provider) + authn_credential_cb = ( + AuthnCredentialCbAdapter(authn_credential_provider) if authn_credential_provider is not None else None ) @@ -487,7 +487,7 @@ def __init__( timeouts=_validate_timeouts(timeout), monitor_host_health=monitor_host_health, fake_host_health_monitor=fake_host_health_monitor, - fake_authn_credential_cb=fake_authn_credential_provider, + authn_credential_cb=authn_credential_cb, ) self._ext.set_owned_by_session() diff --git a/src/cpp/pybmq_session.cpp b/src/cpp/pybmq_session.cpp index 8801cf2..26af48d 100644 --- a/src/cpp/pybmq_session.cpp +++ b/src/cpp/pybmq_session.cpp @@ -78,7 +78,7 @@ Session::Session( PyObject* py_session_event_callback, PyObject* py_message_event_callback, PyObject* py_ack_event_callback, - PyObject* fake_authn_credential_cb, + PyObject* authn_credential_cb, const SessionConfig& config, bsl::shared_ptr fake_host_health_monitor_sp, PyObject* error, @@ -112,22 +112,22 @@ Session::Session( bmqt::SessionOptions::AuthnCredentialCb cpp_callback; bool has_auth_callback = false; - if (fake_authn_credential_cb != nullptr && fake_authn_credential_cb != Py_None) { + if (authn_credential_cb != nullptr && authn_credential_cb != Py_None) { // Increment reference count since we're storing the Python object - Py_INCREF(fake_authn_credential_cb); + Py_INCREF(authn_credential_cb); has_auth_callback = true; // Create a C++ lambda that wraps the Python callback // TODO this can't be a lambda cpp_callback = - [fake_authn_credential_cb]( + [authn_credential_cb]( bsl::ostream& error) -> bsl::optional { pybmq::GilAcquireGuard guard; // Call get_credential_data() method on the Python object bslma::ManagedPtr result = RefUtils::toManagedPtr(PyObject_CallMethod( - fake_authn_credential_cb, + authn_credential_cb, "get_credential_data", nullptr)); diff --git a/src/cpp/pybmq_session.h b/src/cpp/pybmq_session.h index 5802013..44fe897 100644 --- a/src/cpp/pybmq_session.h +++ b/src/cpp/pybmq_session.h @@ -54,7 +54,7 @@ class Session Session(PyObject* py_session_event_callback, PyObject* py_message_event_callback, PyObject* py_ack_event_callback, - PyObject* fake_authn_credential_cb, + PyObject* authn_credential_cb, const SessionConfig& config, bsl::shared_ptr fake_host_health_monitor, PyObject* d_error, diff --git a/src/declarations/pybmq.pxd b/src/declarations/pybmq.pxd index bdd4b75..e0722d5 100644 --- a/src/declarations/pybmq.pxd +++ b/src/declarations/pybmq.pxd @@ -58,7 +58,7 @@ cdef extern from "pybmq_session.h" namespace "BloombergLP::pybmq" nogil: Session(object on_session_event, object on_message_event, object on_ack_event, - object fake_authn_credential_cb, + object authn_credential_cb, const SessionConfig& config, shared_ptr[ManualHostHealthMonitor] fake_host_health_monitor_sp, object error, diff --git a/tests/unit/test_authn_credential_cb_adapter.py b/tests/unit/test_authn_credential_cb_adapter.py new file mode 100644 index 0000000..4e639e2 --- /dev/null +++ b/tests/unit/test_authn_credential_cb_adapter.py @@ -0,0 +1,114 @@ +# Copyright 2026 Bloomberg Finance L.P. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from blazingmq._ext import AuthnCredentialCbAdapter + + +def test_valid_return(): + # GIVEN + def provider(): + return ("mechanism", b"data") + + adapter = AuthnCredentialCbAdapter(provider) + + # WHEN + result = adapter.get_credential_data() + + # THEN + assert result == ("mechanism", b"data") + + +def test_none_return(): + # GIVEN + def provider(): + return None + + adapter = AuthnCredentialCbAdapter(provider) + + # WHEN + result = adapter.get_credential_data() + + # THEN + assert result is None + + +def test_not_a_tuple(): + # GIVEN + def provider(): + return "not a tuple" + + adapter = AuthnCredentialCbAdapter(provider) + + # WHEN + result = adapter.get_credential_data() + + # THEN + assert result is None + + +def test_tuple_wrong_length(): + # GIVEN + def provider(): + return ("mechanism", b"data", "extra") + + adapter = AuthnCredentialCbAdapter(provider) + + # WHEN + result = adapter.get_credential_data() + + # THEN + assert result is None + + +def test_mechanism_not_str(): + # GIVEN + def provider(): + return (123, b"data") + + adapter = AuthnCredentialCbAdapter(provider) + + # WHEN + result = adapter.get_credential_data() + + # THEN + assert result is None + + +def test_data_not_bytes(): + # GIVEN + def provider(): + return ("mechanism", "not bytes") + + adapter = AuthnCredentialCbAdapter(provider) + + # WHEN + result = adapter.get_credential_data() + + # THEN + assert result is None + + +def test_callback_raises(): + # GIVEN + def provider(): + raise RuntimeError("broken") + + adapter = AuthnCredentialCbAdapter(provider) + + # WHEN + result = adapter.get_credential_data() + + # THEN + assert result is None diff --git a/tests/unit/test_session.py b/tests/unit/test_session.py index 7e0332d..2319527 100644 --- a/tests/unit/test_session.py +++ b/tests/unit/test_session.py @@ -78,7 +78,7 @@ def dummy2(): ), monitor_host_health=False, fake_host_health_monitor=None, - fake_authn_credential_cb=None, + authn_credential_cb=None, ) @@ -126,10 +126,10 @@ def my_provider(): ), monitor_host_health=False, fake_host_health_monitor=None, - fake_authn_credential_cb=mock.ANY, + authn_credential_cb=mock.ANY, ) call_kwargs = ext_cls.call_args[1] - assert call_kwargs["fake_authn_credential_cb"] is not None + assert call_kwargs["authn_credential_cb"] is not None @mock.patch("blazingmq._session.ExtSession") @@ -179,7 +179,7 @@ def dummy2(): timeouts=timeouts, monitor_host_health=False, fake_host_health_monitor=None, - fake_authn_credential_cb=None, + authn_credential_cb=None, ) @@ -224,7 +224,7 @@ def dummy2(): timeouts=timeouts, monitor_host_health=False, fake_host_health_monitor=None, - fake_authn_credential_cb=None, + authn_credential_cb=None, ) @@ -260,7 +260,7 @@ def dummy2(): timeouts=Timeouts(), monitor_host_health=False, fake_host_health_monitor=None, - fake_authn_credential_cb=None, + authn_credential_cb=None, ) @@ -313,7 +313,7 @@ def dummy2(): timeouts=timeouts, monitor_host_health=False, fake_host_health_monitor=None, - fake_authn_credential_cb=None, + authn_credential_cb=None, ) @@ -352,10 +352,10 @@ def my_provider(): timeouts=Timeouts(), monitor_host_health=False, fake_host_health_monitor=None, - fake_authn_credential_cb=mock.ANY, + authn_credential_cb=mock.ANY, ) call_kwargs = ext_cls.call_args[1] - assert call_kwargs["fake_authn_credential_cb"] is not None + assert call_kwargs["authn_credential_cb"] is not None @mock.patch("blazingmq._session.ExtSession") @@ -404,10 +404,10 @@ def my_provider(): timeouts=timeouts, monitor_host_health=False, fake_host_health_monitor=None, - fake_authn_credential_cb=mock.ANY, + authn_credential_cb=mock.ANY, ) call_kwargs = ext_cls.call_args[1] - assert call_kwargs["fake_authn_credential_cb"] is not None + assert call_kwargs["authn_credential_cb"] is not None @mock.patch("blazingmq._session.ExtSession") @@ -452,7 +452,7 @@ def dummy2(): ), monitor_host_health=True, fake_host_health_monitor=monitor._monitor, - fake_authn_credential_cb=None, + authn_credential_cb=None, ) @@ -484,7 +484,7 @@ def dummy2(): timeouts=Timeouts(), monitor_host_health=False, fake_host_health_monitor=None, - fake_authn_credential_cb=None, + authn_credential_cb=None, ) From 0b431fe10acc173ab8acf47304bd8c72bf62c8c3 Mon Sep 17 00:00:00 2001 From: "Patrick M. Niedzielski" Date: Tue, 30 Jun 2026 14:37:42 -0400 Subject: [PATCH 17/26] Fix `isort` order Signed-off-by: Patrick M. Niedzielski --- src/blazingmq/_ext.pyx | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/blazingmq/_ext.pyx b/src/blazingmq/_ext.pyx index 74283fe..4ca31c1 100644 --- a/src/blazingmq/_ext.pyx +++ b/src/blazingmq/_ext.pyx @@ -21,8 +21,8 @@ import weakref from bsl cimport optional from bsl cimport pair from bsl cimport shared_ptr -from bsl cimport vector from bsl cimport string +from bsl cimport vector from bsl.bsls cimport TimeInterval from cpython.ceval cimport PyEval_InitThreads from libcpp cimport bool as cppbool From d5750e173cbbdbb379ebbd3e7047ddf1552cf4f1 Mon Sep 17 00:00:00 2001 From: "Patrick M. Niedzielski" Date: Tue, 30 Jun 2026 15:19:39 -0400 Subject: [PATCH 18/26] Add `AuthnCredentialProvider` type alias Signed-off-by: Patrick M. Niedzielski --- src/blazingmq/__init__.py | 2 ++ src/blazingmq/_session.py | 9 +++++---- src/blazingmq/_typing.py | 8 ++++++++ 3 files changed, 15 insertions(+), 4 deletions(-) diff --git a/src/blazingmq/__init__.py b/src/blazingmq/__init__.py index 910d1ad..28d81ef 100644 --- a/src/blazingmq/__init__.py +++ b/src/blazingmq/__init__.py @@ -27,6 +27,7 @@ from ._session import Session from ._session import SessionOptions from ._timeouts import Timeouts +from ._typing import AuthnCredentialProvider from ._typing import PropertyTypeDict from ._typing import PropertyValueDict from .exceptions import Error @@ -34,6 +35,7 @@ __all__ = [ "Ack", "AckStatus", + "AuthnCredentialProvider", "BasicHealthMonitor", "CompressionAlgorithmType", "Error", diff --git a/src/blazingmq/_session.py b/src/blazingmq/_session.py index abc5e0a..831ea2b 100644 --- a/src/blazingmq/_session.py +++ b/src/blazingmq/_session.py @@ -37,6 +37,7 @@ from ._messages import MessageHandle from ._monitors import BasicHealthMonitor from ._timeouts import Timeouts +from ._typing import AuthnCredentialProvider from ._typing import PropertyTypeDict from ._typing import PropertyValueDict from ._typing import PropertyValueType @@ -53,7 +54,7 @@ def DefaultMonitor() -> Union[BasicHealthMonitor, None]: return None -def DefaultAuthnCredentialCb() -> Optional[Callable]: +def DefaultAuthnCredentialProvider() -> Optional[AuthnCredentialProvider]: return None @@ -305,7 +306,7 @@ def __init__( message_compression_algorithm: Optional[CompressionAlgorithmType] = None, timeouts: Optional[Timeouts] = None, host_health_monitor: Union[BasicHealthMonitor, None] = (DefaultMonitor()), - authn_credential_provider: Optional[Callable] = ( + authn_credential_provider: Optional[AuthnCredentialProvider] = ( DefaultAuthnCredentialProvider() ), num_processing_threads: Optional[int] = None, @@ -438,8 +439,8 @@ def __init__( ), timeout: Union[Timeouts, float] = DEFAULT_TIMEOUT, host_health_monitor: Union[BasicHealthMonitor, None] = (DefaultMonitor()), - authn_credential_provider: Optional[Callable] = ( - DefaultAuthnCredentialCb() + authn_credential_provider: Optional[AuthnCredentialProvider] = ( + DefaultAuthnCredentialProvider() ), num_processing_threads: Optional[int] = None, blob_buffer_size: Optional[int] = None, diff --git a/src/blazingmq/_typing.py b/src/blazingmq/_typing.py index 39cf633..7c09d27 100644 --- a/src/blazingmq/_typing.py +++ b/src/blazingmq/_typing.py @@ -13,7 +13,9 @@ # See the License for the specific language governing permissions and # limitations under the License. +from typing import Callable from typing import Mapping +from typing import Optional from typing import Union from ._enums import PropertyType @@ -23,3 +25,9 @@ PropertyValueDict = Mapping[str, PropertyValueType] PropertyTypeDict = Mapping[str, PropertyType] + +AuthnCredentialProvider = Callable[[], Optional[tuple[str, bytes]]] +"""A callable that returns authentication credentials as a tuple of +``(mechanism, data)`` tuple of ``(str, bytes)``, or ``None`` if an +error occurs while obtaining credentials. +""" From fd15953d8a857c8c49804c6981d15f0731d2790a Mon Sep 17 00:00:00 2001 From: "Patrick M. Niedzielski" Date: Tue, 30 Jun 2026 18:29:04 -0400 Subject: [PATCH 19/26] clang-format C++ code Signed-off-by: Patrick M. Niedzielski --- src/cpp/pybmq_session.cpp | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/cpp/pybmq_session.cpp b/src/cpp/pybmq_session.cpp index 26af48d..dd2f9f9 100644 --- a/src/cpp/pybmq_session.cpp +++ b/src/cpp/pybmq_session.cpp @@ -170,7 +170,7 @@ Session::Session( // (AuthnCredential is move-only) bmqt::AuthnCredential credential(mechanism, data); bsl::optional opt_credential( - bslmf::MovableRefUtil::move(credential)); + bslmf::MovableRefUtil::move(credential)); return opt_credential; }; } From 8736ec9d1c20376cdb1698e93204e8c3e6114c56 Mon Sep 17 00:00:00 2001 From: "Patrick M. Niedzielski" Date: Wed, 1 Jul 2026 11:02:49 -0400 Subject: [PATCH 20/26] Add documentation for `AuthnCredentialProvider` Signed-off-by: Patrick M. Niedzielski --- docs/api_reference.rst | 2 ++ src/blazingmq/_session.py | 11 ++++++----- 2 files changed, 8 insertions(+), 5 deletions(-) diff --git a/docs/api_reference.rst b/docs/api_reference.rst index b380ab8..438ff85 100644 --- a/docs/api_reference.rst +++ b/docs/api_reference.rst @@ -187,6 +187,8 @@ Testing Utilities Helper Types ============ +.. autoclass:: blazingmq.AuthnCredentialProvider + .. autoclass:: blazingmq.PropertyTypeDict .. autoclass:: blazingmq.PropertyValueDict diff --git a/src/blazingmq/_session.py b/src/blazingmq/_session.py index 831ea2b..955af38 100644 --- a/src/blazingmq/_session.py +++ b/src/blazingmq/_session.py @@ -271,7 +271,7 @@ class SessionOptions: healthy, `.HostUnhealthy` and `.HostHealthRestored` events with never be emitted, and the *suspends_on_bad_host_health* option of `QueueOptions` cannot be used. - authn_credential_provider: + authn_credential_provider (Optional[`~blazingmq.AuthnCredentialProvider`]): An optional callable that returns authentication credentials as a ``(mechanism, data)`` tuple of ``(str, bytes)``, or ``None`` if no credentials are available. If not provided, no authentication @@ -396,10 +396,11 @@ class Session: `.HostHealthRestored` events will never be emitted, and the *suspends_on_bad_host_health* option of `QueueOptions` cannot be used. - authn_credential_provider: an optional callable that returns authentication - credentials as a ``(mechanism, data)`` tuple of ``(str, bytes)``, - or ``None`` if no credentials are available. If not provided, no - authentication credentials are sent to the broker. + authn_credential_provider (Optional[`~blazingmq.AuthnCredentialProvider`]): + an optional callable that returns authentication credentials as a + ``(mechanism, data)`` tuple of ``(str, bytes)``, or ``None`` if no + credentials are available. If not provided, no authentication + credentials are sent to the broker. num_processing_threads: The number of threads for the SDK to use for processing events. This defaults to 1. blob_buffer_size: The size (in bytes) of the blob buffers to use. This From 2c4bf216065bc725e573cf16ff2aa53cd3b517ca Mon Sep 17 00:00:00 2001 From: "Patrick M. Niedzielski" Date: Mon, 31 Aug 2026 17:19:19 -0400 Subject: [PATCH 21/26] Fix: Remove `ostream& error` from authn callback Signed-off-by: Patrick M. Niedzielski --- src/cpp/pybmq_session.cpp | 18 +++++++++++------- 1 file changed, 11 insertions(+), 7 deletions(-) diff --git a/src/cpp/pybmq_session.cpp b/src/cpp/pybmq_session.cpp index dd2f9f9..a58672e 100644 --- a/src/cpp/pybmq_session.cpp +++ b/src/cpp/pybmq_session.cpp @@ -22,6 +22,7 @@ #include #include +#include #include #include #include @@ -119,9 +120,9 @@ Session::Session( // Create a C++ lambda that wraps the Python callback // TODO this can't be a lambda - cpp_callback = - [authn_credential_cb]( - bsl::ostream& error) -> bsl::optional { + cpp_callback = [authn_credential_cb]() -> bsl::optional { + BALL_LOG_SET_CATEGORY("pybmq_session"); + pybmq::GilAcquireGuard guard; // Call get_credential_data() method on the Python object @@ -132,9 +133,10 @@ Session::Session( nullptr)); if (!result) { - // Python exception occurred + // Python exception occurred. Clear it before logging, so we + // don't re-enter Python via the BALL observer with it set. PyErr_Print(); - error << "Error calling get_credential_data()"; + BALL_LOG_ERROR << "Error calling get_credential_data()"; return bsl::optional(); } @@ -144,7 +146,8 @@ Session::Session( // Extract tuple (mechanism, data) if (!PyTuple_Check(result.get()) || PyTuple_Size(result.get()) != 2) { - error << "get_credential_data() must return (str, bytes) or None"; + BALL_LOG_ERROR + << "get_credential_data() must return (str, bytes) or None"; return bsl::optional(); } @@ -152,7 +155,8 @@ Session::Session( PyObject* data_obj = PyTuple_GetItem(result.get(), 1); if (!PyUnicode_Check(mechanism_obj) || !PyBytes_Check(data_obj)) { - error << "get_credential_data() must return (str, bytes) or None"; + BALL_LOG_ERROR + << "get_credential_data() must return (str, bytes) or None"; return bsl::optional(); } From a64a7c205d0eafae063931c76491f0322a668afb Mon Sep 17 00:00:00 2001 From: "Patrick M. Niedzielski" Date: Tue, 1 Sep 2026 10:55:13 -0400 Subject: [PATCH 22/26] Remove lambda for C++03 compat Signed-off-by: Patrick M. Niedzielski --- src/cpp/pybmq_session.cpp | 155 ++++++++++++++++++++++---------------- src/cpp/pybmq_session.h | 1 + 2 files changed, 90 insertions(+), 66 deletions(-) diff --git a/src/cpp/pybmq_session.cpp b/src/cpp/pybmq_session.cpp index a58672e..78e6f95 100644 --- a/src/cpp/pybmq_session.cpp +++ b/src/cpp/pybmq_session.cpp @@ -73,6 +73,81 @@ class BrokerTimeoutError : public bsl::runtime_error } }; +// Invoke `get_credential_data()` on a Python object and convert the result to +// a `bmqt::AuthnCredential`. The object is held, not owned: the `Session` +// holds the reference and drops it only after destroying the `bmqa::Session` +// that owns every copy of this functor. +class AuthnCredentialCbFunctor +{ + private: + // DATA + PyObject* d_callback_p; + + public: + // CREATORS + explicit AuthnCredentialCbFunctor(PyObject* callback); + + // ACCESSORS + bsl::optional operator()() const; +}; + +AuthnCredentialCbFunctor::AuthnCredentialCbFunctor(PyObject* callback) +: d_callback_p(callback) +{ +} + +bsl::optional +AuthnCredentialCbFunctor::operator()() const +{ + BALL_LOG_SET_CATEGORY("pybmq_session"); + + pybmq::GilAcquireGuard guard; + + // Call get_credential_data() method on the Python object + bslma::ManagedPtr result = RefUtils::toManagedPtr( + PyObject_CallMethod(d_callback_p, "get_credential_data", NULL)); + + if (!result) { + // Python exception occurred. Clear it before logging, so we + // don't re-enter Python via the BALL observer with it set. + PyErr_Print(); + BALL_LOG_ERROR << "Error calling get_credential_data()"; + return bsl::optional(); + } + + if (result.get() == Py_None) { + return bsl::optional(); + } + + // Extract tuple (mechanism, data) + if (!PyTuple_Check(result.get()) || PyTuple_Size(result.get()) != 2) { + BALL_LOG_ERROR << "get_credential_data() must return (str, bytes) or None"; + return bsl::optional(); + } + + PyObject* mechanism_obj = PyTuple_GetItem(result.get(), 0); + PyObject* data_obj = PyTuple_GetItem(result.get(), 1); + + if (!PyUnicode_Check(mechanism_obj) || !PyBytes_Check(data_obj)) { + BALL_LOG_ERROR << "get_credential_data() must return (str, bytes) or None"; + return bsl::optional(); + } + + // Convert Python str to C++ string + const char* mechanism_cstr = PyUnicode_AsUTF8(mechanism_obj); + bsl::string mechanism(mechanism_cstr); + + // Convert Python bytes to vector + char* data_ptr; + Py_ssize_t data_len; + PyBytes_AsStringAndSize(data_obj, &data_ptr, &data_len); + bsl::vector data(data_ptr, data_ptr + data_len); + + bmqt::AuthnCredential credential(mechanism, data); + return bsl::optional( + bslmf::MovableRefUtil::move(credential)); +} + } // namespace Session::Session( @@ -90,6 +165,7 @@ Session::Session( , d_message_compression_type(bmqt::CompressionAlgorithmType::e_NONE) , d_error(error) , d_broker_timeout_error(broker_timeout_error) +, d_authn_credential_cb(NULL) , d_session_mp() { bsl::shared_ptr host_health_monitor_sp; @@ -113,70 +189,10 @@ Session::Session( bmqt::SessionOptions::AuthnCredentialCb cpp_callback; bool has_auth_callback = false; - if (authn_credential_cb != nullptr && authn_credential_cb != Py_None) { - // Increment reference count since we're storing the Python object - Py_INCREF(authn_credential_cb); + if (authn_credential_cb != NULL && authn_credential_cb != Py_None) { + d_authn_credential_cb = authn_credential_cb; + cpp_callback = AuthnCredentialCbFunctor(d_authn_credential_cb); has_auth_callback = true; - - // Create a C++ lambda that wraps the Python callback - // TODO this can't be a lambda - cpp_callback = [authn_credential_cb]() -> bsl::optional { - BALL_LOG_SET_CATEGORY("pybmq_session"); - - pybmq::GilAcquireGuard guard; - - // Call get_credential_data() method on the Python object - bslma::ManagedPtr result = - RefUtils::toManagedPtr(PyObject_CallMethod( - authn_credential_cb, - "get_credential_data", - nullptr)); - - if (!result) { - // Python exception occurred. Clear it before logging, so we - // don't re-enter Python via the BALL observer with it set. - PyErr_Print(); - BALL_LOG_ERROR << "Error calling get_credential_data()"; - return bsl::optional(); - } - - if (result.get() == Py_None) { - return bsl::optional(); - } - - // Extract tuple (mechanism, data) - if (!PyTuple_Check(result.get()) || PyTuple_Size(result.get()) != 2) { - BALL_LOG_ERROR - << "get_credential_data() must return (str, bytes) or None"; - return bsl::optional(); - } - - PyObject* mechanism_obj = PyTuple_GetItem(result.get(), 0); - PyObject* data_obj = PyTuple_GetItem(result.get(), 1); - - if (!PyUnicode_Check(mechanism_obj) || !PyBytes_Check(data_obj)) { - BALL_LOG_ERROR - << "get_credential_data() must return (str, bytes) or None"; - return bsl::optional(); - } - - // Convert Python str to C++ string - const char* mechanism_cstr = PyUnicode_AsUTF8(mechanism_obj); - bsl::string mechanism(mechanism_cstr); - - // Convert Python bytes to vector - char* data_ptr; - Py_ssize_t data_len; - PyBytes_AsStringAndSize(data_obj, &data_ptr, &data_len); - bsl::vector data(data_ptr, data_ptr + data_len); - - // Construct and move credential into optional - // (AuthnCredential is move-only) - bmqt::AuthnCredential credential(mechanism, data); - bsl::optional opt_credential( - bslmf::MovableRefUtil::move(credential)); - return opt_credential; - }; } { @@ -247,15 +263,22 @@ Session::Session( } Py_INCREF(d_error); Py_INCREF(d_broker_timeout_error); + Py_XINCREF(d_authn_credential_cb); } Session::~Session() { + BSLS_ASSERT(!d_started); + { + // Destroy the session first: it owns the copies of + // `AuthnCredentialCbFunctor`, which borrow `d_authn_credential_cb`. + pybmq::GilReleaseGuard gil_release_guard; + d_session_mp.reset(); + } + + Py_XDECREF(d_authn_credential_cb); Py_DECREF(d_broker_timeout_error); Py_DECREF(d_error); - BSLS_ASSERT(!d_started); - pybmq::GilReleaseGuard gil_release_guard; - d_session_mp.reset(); } PyObject* diff --git a/src/cpp/pybmq_session.h b/src/cpp/pybmq_session.h index 44fe897..3404720 100644 --- a/src/cpp/pybmq_session.h +++ b/src/cpp/pybmq_session.h @@ -44,6 +44,7 @@ class Session bmqt::CompressionAlgorithmType::Enum d_message_compression_type; PyObject* d_error; PyObject* d_broker_timeout_error; + PyObject* d_authn_credential_cb; bslma::ManagedPtr d_session_mp; // NOT IMPLEMENTED From c8f16a89b93b4371d9cd13fcbea69969003ae38e Mon Sep 17 00:00:00 2001 From: "Patrick M. Niedzielski" Date: Tue, 1 Sep 2026 14:10:25 -0400 Subject: [PATCH 23/26] Move `authn_credential_provider` argument to avoid API break Signed-off-by: Patrick M. Niedzielski --- src/blazingmq/_session.py | 40 +++++++++++++++++++-------------------- 1 file changed, 20 insertions(+), 20 deletions(-) diff --git a/src/blazingmq/_session.py b/src/blazingmq/_session.py index 955af38..b397dc5 100644 --- a/src/blazingmq/_session.py +++ b/src/blazingmq/_session.py @@ -271,11 +271,6 @@ class SessionOptions: healthy, `.HostUnhealthy` and `.HostHealthRestored` events with never be emitted, and the *suspends_on_bad_host_health* option of `QueueOptions` cannot be used. - authn_credential_provider (Optional[`~blazingmq.AuthnCredentialProvider`]): - An optional callable that returns authentication credentials as a - ``(mechanism, data)`` tuple of ``(str, bytes)``, or ``None`` if no - credentials are available. If not provided, no authentication - credentials are sent to the broker. num_processing_threads: The number of threads for the SDK to use for processing events. This defaults to 1. @@ -299,6 +294,11 @@ class SessionOptions: 0, disable the recurring dump of stats (final stats are always dumped at the end of the session). The default is 5min; the value must be a multiple of 30s, in the range ``[0s - 60min]``. + authn_credential_provider (Optional[`~blazingmq.AuthnCredentialProvider`]): + An optional callable that returns authentication credentials as a + ``(mechanism, data)`` tuple of ``(str, bytes)``, or ``None`` if no + credentials are available. If not provided, no authentication + credentials are sent to the broker. """ def __init__( @@ -306,24 +306,24 @@ def __init__( message_compression_algorithm: Optional[CompressionAlgorithmType] = None, timeouts: Optional[Timeouts] = None, host_health_monitor: Union[BasicHealthMonitor, None] = (DefaultMonitor()), - authn_credential_provider: Optional[AuthnCredentialProvider] = ( - DefaultAuthnCredentialProvider() - ), num_processing_threads: Optional[int] = None, blob_buffer_size: Optional[int] = None, channel_high_watermark: Optional[int] = None, event_queue_watermarks: Optional[tuple[int, int]] = None, stats_dump_interval: Optional[float] = None, + authn_credential_provider: Optional[AuthnCredentialProvider] = ( + DefaultAuthnCredentialProvider() + ), ) -> None: self.message_compression_algorithm = message_compression_algorithm self.timeouts = timeouts self.host_health_monitor = host_health_monitor - self.authn_credential_provider = authn_credential_provider self.num_processing_threads = num_processing_threads self.blob_buffer_size = blob_buffer_size self.channel_high_watermark = channel_high_watermark self.event_queue_watermarks = event_queue_watermarks self.stats_dump_interval = stats_dump_interval + self.authn_credential_provider = authn_credential_provider def __eq__(self, other: object) -> bool: if not isinstance(other, SessionOptions): @@ -332,12 +332,12 @@ def __eq__(self, other: object) -> bool: self.message_compression_algorithm == other.message_compression_algorithm and self.timeouts == other.timeouts and self.host_health_monitor == other.host_health_monitor - and self.authn_credential_provider == other.authn_credential_provider and self.num_processing_threads == other.num_processing_threads and self.blob_buffer_size == other.blob_buffer_size and self.channel_high_watermark == other.channel_high_watermark and self.event_queue_watermarks == other.event_queue_watermarks and self.stats_dump_interval == other.stats_dump_interval + and self.authn_credential_provider == other.authn_credential_provider ) def __ne__(self, other: object) -> bool: @@ -348,12 +348,12 @@ def __repr__(self) -> str: "message_compression_algorithm", "timeouts", "host_health_monitor", - "authn_credential_provider", "num_processing_threads", "blob_buffer_size", "channel_high_watermark", "event_queue_watermarks", "stats_dump_interval", + "authn_credential_provider", ) params = [] @@ -396,11 +396,6 @@ class Session: `.HostHealthRestored` events will never be emitted, and the *suspends_on_bad_host_health* option of `QueueOptions` cannot be used. - authn_credential_provider (Optional[`~blazingmq.AuthnCredentialProvider`]): - an optional callable that returns authentication credentials as a - ``(mechanism, data)`` tuple of ``(str, bytes)``, or ``None`` if no - credentials are available. If not provided, no authentication - credentials are sent to the broker. num_processing_threads: The number of threads for the SDK to use for processing events. This defaults to 1. blob_buffer_size: The size (in bytes) of the blob buffers to use. This @@ -421,6 +416,11 @@ class Session: stats are always dumped at the end of the session). The default is 5min; the value must be a multiple of 30s, in the range ``[0s - 60min]``. + authn_credential_provider (Optional[`~blazingmq.AuthnCredentialProvider`]): + an optional callable that returns authentication credentials as a + ``(mechanism, data)`` tuple of ``(str, bytes)``, or ``None`` if no + credentials are available. If not provided, no authentication + credentials are sent to the broker. Raises: `~blazingmq.Error`: If the session start request was not successful. @@ -440,14 +440,14 @@ def __init__( ), timeout: Union[Timeouts, float] = DEFAULT_TIMEOUT, host_health_monitor: Union[BasicHealthMonitor, None] = (DefaultMonitor()), - authn_credential_provider: Optional[AuthnCredentialProvider] = ( - DefaultAuthnCredentialProvider() - ), num_processing_threads: Optional[int] = None, blob_buffer_size: Optional[int] = None, channel_high_watermark: Optional[int] = None, event_queue_watermarks: Optional[tuple[int, int]] = None, stats_dump_interval: Optional[float] = None, + authn_credential_provider: Optional[AuthnCredentialProvider] = ( + DefaultAuthnCredentialProvider() + ), ) -> None: if host_health_monitor is not None: if not isinstance(host_health_monitor, BasicHealthMonitor): @@ -545,12 +545,12 @@ def with_options( message_compression_algorithm=message_compression_algorithm, timeout=timeout, host_health_monitor=session_options.host_health_monitor, - authn_credential_provider=session_options.authn_credential_provider, num_processing_threads=session_options.num_processing_threads, blob_buffer_size=session_options.blob_buffer_size, channel_high_watermark=session_options.channel_high_watermark, event_queue_watermarks=session_options.event_queue_watermarks, stats_dump_interval=session_options.stats_dump_interval, + authn_credential_provider=session_options.authn_credential_provider, ) def open_queue( From bace6e5feaa8587bd1df67b5d8f4d1bb7c620cd0 Mon Sep 17 00:00:00 2001 From: "Patrick M. Niedzielski" Date: Tue, 1 Sep 2026 14:25:46 -0400 Subject: [PATCH 24/26] Fix: clarify docstring for `authn_credential_provider` Signed-off-by: Patrick M. Niedzielski --- src/blazingmq/_session.py | 20 ++++++++++++++------ src/blazingmq/_typing.py | 7 ++++--- 2 files changed, 18 insertions(+), 9 deletions(-) diff --git a/src/blazingmq/_session.py b/src/blazingmq/_session.py index b397dc5..6ef216e 100644 --- a/src/blazingmq/_session.py +++ b/src/blazingmq/_session.py @@ -296,9 +296,13 @@ class SessionOptions: must be a multiple of 30s, in the range ``[0s - 60min]``. authn_credential_provider (Optional[`~blazingmq.AuthnCredentialProvider`]): An optional callable that returns authentication credentials as a - ``(mechanism, data)`` tuple of ``(str, bytes)``, or ``None`` if no - credentials are available. If not provided, no authentication - credentials are sent to the broker. + ``(mechanism, data)`` tuple of ``(str, bytes)``. It is called + each time the session authenticates with the broker, including on + reauthentication. If it returns ``None`` or raises, the + connection is closed: starting a session fails, while an + already-started session sees `.ConnectionLost` and then + reconnects, calling this callable again. If not provided, no + authentication credentials are sent to the broker. """ def __init__( @@ -418,9 +422,13 @@ class Session: ``[0s - 60min]``. authn_credential_provider (Optional[`~blazingmq.AuthnCredentialProvider`]): an optional callable that returns authentication credentials as a - ``(mechanism, data)`` tuple of ``(str, bytes)``, or ``None`` if no - credentials are available. If not provided, no authentication - credentials are sent to the broker. + ``(mechanism, data)`` tuple of ``(str, bytes)``. It is called + each time the session authenticates with the broker, including on + reauthentication. If it returns ``None`` or raises, the + connection is closed: starting a session fails, while an + already-started session sees `.ConnectionLost` and then + reconnects, calling this callable again. If not provided, no + authentication credentials are sent to the broker. Raises: `~blazingmq.Error`: If the session start request was not successful. diff --git a/src/blazingmq/_typing.py b/src/blazingmq/_typing.py index 7c09d27..1110405 100644 --- a/src/blazingmq/_typing.py +++ b/src/blazingmq/_typing.py @@ -27,7 +27,8 @@ PropertyTypeDict = Mapping[str, PropertyType] AuthnCredentialProvider = Callable[[], Optional[tuple[str, bytes]]] -"""A callable that returns authentication credentials as a tuple of -``(mechanism, data)`` tuple of ``(str, bytes)``, or ``None`` if an -error occurs while obtaining credentials. +"""A callable that returns authentication credentials as a ``(mechanism, +data)`` tuple of ``(str, bytes)``, or ``None`` if an error occurs while +obtaining them, in which case authentication fails and the connection is +closed. """ From f3187052ea06a04acd5f6a8327b779c9738fd987 Mon Sep 17 00:00:00 2001 From: "Patrick M. Niedzielski" Date: Tue, 1 Sep 2026 14:54:03 -0400 Subject: [PATCH 25/26] Style: Move UTF-8 encoding logic into Cython from C++ Signed-off-by: Patrick M. Niedzielski --- src/blazingmq/_ext.pyx | 24 ++++---- src/cpp/pybmq_session.cpp | 55 +++++++++---------- src/cpp/pybmq_session.h | 1 - src/declarations/bmq/bmqt.pxd | 12 +--- .../unit/test_authn_credential_cb_adapter.py | 44 ++++++++++++++- 5 files changed, 80 insertions(+), 56 deletions(-) diff --git a/src/blazingmq/_ext.pyx b/src/blazingmq/_ext.pyx index 4ca31c1..befa011 100644 --- a/src/blazingmq/_ext.pyx +++ b/src/blazingmq/_ext.pyx @@ -21,15 +21,12 @@ import weakref from bsl cimport optional from bsl cimport pair from bsl cimport shared_ptr -from bsl cimport string -from bsl cimport vector from bsl.bsls cimport TimeInterval from cpython.ceval cimport PyEval_InitThreads from libcpp cimport bool as cppbool from bmq.bmqa cimport ManualHostHealthMonitor from bmq.bmqt cimport AckResult -from bmq.bmqt cimport AuthnCredential from bmq.bmqt cimport CompressionAlgorithmType from bmq.bmqt cimport HostHealthState from bmq.bmqt cimport PropertyType @@ -158,31 +155,32 @@ cdef class FakeHostHealthMonitor: cdef class AuthnCredentialCbAdapter: - cdef object _callback # Store the Python callable + cdef object _callback def __cinit__(self, callback): self._callback = callback - # This method will be called by C++ code via PyObject_CallMethod - # Returns None for no credential, or (mechanism, data) tuple def get_credential_data(self): + """Call the provider and marshal its result for the C++ session. + + Called by ``pybmq::AuthnCredentialCbFunctor``. Returns the mechanism + and data as a tuple of ``bytes``, or `None` if credentials could not + be obtained, in which case authentication fails. + """ try: result = self._callback() if result is None: return None - if not isinstance(result, tuple) or len(result) != 2: - raise ValueError("callback must return (str, bytes) or None") - mechanism, data = result if not isinstance(mechanism, str) or not isinstance(data, bytes): - raise ValueError("callback must return (str, bytes) or None") + raise TypeError( + "authn_credential_provider must return (str, bytes) or None" + ) - # Return as-is, let C++ side handle conversion - return result + return mechanism.encode('utf-8'), data except Exception: - # Log error or handle as needed LOGGER.exception("Error in authentication credential callback") return None diff --git a/src/cpp/pybmq_session.cpp b/src/cpp/pybmq_session.cpp index 78e6f95..1262dae 100644 --- a/src/cpp/pybmq_session.cpp +++ b/src/cpp/pybmq_session.cpp @@ -22,17 +22,20 @@ #include #include -#include #include #include #include #include +#include +#include #include #include +#include #include #include #include +#include #include #include #include @@ -99,19 +102,16 @@ AuthnCredentialCbFunctor::AuthnCredentialCbFunctor(PyObject* callback) bsl::optional AuthnCredentialCbFunctor::operator()() const { - BALL_LOG_SET_CATEGORY("pybmq_session"); - pybmq::GilAcquireGuard guard; - // Call get_credential_data() method on the Python object + // The adapter validates the provider's result and reports any problem to + // the Python logger, so a failure here is just an empty credential. It + // hands back the mechanism and data already marshalled to `bytes`. bslma::ManagedPtr result = RefUtils::toManagedPtr( PyObject_CallMethod(d_callback_p, "get_credential_data", NULL)); if (!result) { - // Python exception occurred. Clear it before logging, so we - // don't re-enter Python via the BALL observer with it set. - PyErr_Print(); - BALL_LOG_ERROR << "Error calling get_credential_data()"; + PyErr_WriteUnraisable(d_callback_p); return bsl::optional(); } @@ -119,31 +119,26 @@ AuthnCredentialCbFunctor::operator()() const return bsl::optional(); } - // Extract tuple (mechanism, data) - if (!PyTuple_Check(result.get()) || PyTuple_Size(result.get()) != 2) { - BALL_LOG_ERROR << "get_credential_data() must return (str, bytes) or None"; - return bsl::optional(); - } - - PyObject* mechanism_obj = PyTuple_GetItem(result.get(), 0); - PyObject* data_obj = PyTuple_GetItem(result.get(), 1); - - if (!PyUnicode_Check(mechanism_obj) || !PyBytes_Check(data_obj)) { - BALL_LOG_ERROR << "get_credential_data() must return (str, bytes) or None"; + const char* mechanism_p; + Py_ssize_t mechanism_len; + const char* data_p; + Py_ssize_t data_len; + if (!PyArg_ParseTuple( + result.get(), + "y#y#", + &mechanism_p, + &mechanism_len, + &data_p, + &data_len)) + { + // The adapter broke its contract with us. + PyErr_WriteUnraisable(d_callback_p); return bsl::optional(); } - // Convert Python str to C++ string - const char* mechanism_cstr = PyUnicode_AsUTF8(mechanism_obj); - bsl::string mechanism(mechanism_cstr); - - // Convert Python bytes to vector - char* data_ptr; - Py_ssize_t data_len; - PyBytes_AsStringAndSize(data_obj, &data_ptr, &data_len); - bsl::vector data(data_ptr, data_ptr + data_len); - - bmqt::AuthnCredential credential(mechanism, data); + bmqt::AuthnCredential credential( + bsl::string_view(mechanism_p, mechanism_len), + bsl::vector(data_p, data_p + data_len)); return bsl::optional( bslmf::MovableRefUtil::move(credential)); } diff --git a/src/cpp/pybmq_session.h b/src/cpp/pybmq_session.h index 3404720..e7795ff 100644 --- a/src/cpp/pybmq_session.h +++ b/src/cpp/pybmq_session.h @@ -23,7 +23,6 @@ #include #include -#include #include #include diff --git a/src/declarations/bmq/bmqt.pxd b/src/declarations/bmq/bmqt.pxd index 5f05d12..07e27e9 100644 --- a/src/declarations/bmq/bmqt.pxd +++ b/src/declarations/bmq/bmqt.pxd @@ -1,4 +1,4 @@ -# Copyright 2019-2026 Bloomberg Finance L.P. +# Copyright 2019-2023 Bloomberg Finance L.P. # SPDX-License-Identifier: Apache-2.0 # # Licensed under the Apache License, Version 2.0 (the "License"); @@ -13,8 +13,6 @@ # See the License for the specific language governing permissions and # limitations under the License. -from bsl cimport string -from bsl cimport vector from libcpp cimport bool @@ -75,11 +73,3 @@ cdef extern from "bmqt_queueoptions.h" namespace "BloombergLP::bmqt::QueueOption int k_DEFAULT_MAX_UNCONFIRMED_BYTES int k_DEFAULT_CONSUMER_PRIORITY bool k_DEFAULT_SUSPENDS_ON_BAD_HOST_HEALTH - -cdef extern from "bmqt_authncredential.h" namespace "BloombergLP::bmqt" nogil: - cdef cppclass AuthnCredential: - AuthnCredential() except + - AuthnCredential(const AuthnCredential&) except + - AuthnCredential(const string& mechanism, const vector[char]& data) except + - const string& mechanism() const - const vector[char]& data() const diff --git a/tests/unit/test_authn_credential_cb_adapter.py b/tests/unit/test_authn_credential_cb_adapter.py index 4e639e2..54b2d11 100644 --- a/tests/unit/test_authn_credential_cb_adapter.py +++ b/tests/unit/test_authn_credential_cb_adapter.py @@ -27,7 +27,49 @@ def provider(): result = adapter.get_credential_data() # THEN - assert result == ("mechanism", b"data") + assert result == (b"mechanism", b"data") + + +def test_mechanism_is_encoded_as_utf8(): + # GIVEN + def provider(): + return ("mécanisme", b"data") + + adapter = AuthnCredentialCbAdapter(provider) + + # WHEN + result = adapter.get_credential_data() + + # THEN + assert result == ("mécanisme".encode("utf-8"), b"data") + + +def test_mechanism_not_encodable(): + # GIVEN a mechanism holding a lone surrogate, which has no UTF-8 encoding + def provider(): + return ("\ud800", b"data") + + adapter = AuthnCredentialCbAdapter(provider) + + # WHEN + result = adapter.get_credential_data() + + # THEN + assert result is None + + +def test_mechanism_with_embedded_nul_is_not_truncated(): + # GIVEN + def provider(): + return ("PLAIN\x00extra", b"data") + + adapter = AuthnCredentialCbAdapter(provider) + + # WHEN + result = adapter.get_credential_data() + + # THEN + assert result == (b"PLAIN\x00extra", b"data") def test_none_return(): From 909a8d4f174c4f9c93201718db3040619605a78d Mon Sep 17 00:00:00 2001 From: "Patrick M. Niedzielski" Date: Tue, 1 Sep 2026 15:14:13 -0400 Subject: [PATCH 26/26] Fix: Simplify authn callback guard and bump copyright years Remove the redundant `has_auth_callback` flag in favour of testing the `bsl::function` directly, and bump copyright years on every file the branch modifies. --- src/blazingmq/__init__.py | 2 +- src/blazingmq/_ext.pyi | 2 +- src/blazingmq/_ext.pyx | 2 +- src/blazingmq/_session.py | 2 +- src/blazingmq/_typing.py | 2 +- src/cpp/pybmq_session.cpp | 6 ++---- src/cpp/pybmq_session.h | 2 +- src/declarations/pybmq.pxd | 2 +- tests/unit/test_session.py | 2 +- tests/unit/test_session_options.py | 2 +- 10 files changed, 11 insertions(+), 13 deletions(-) diff --git a/src/blazingmq/__init__.py b/src/blazingmq/__init__.py index 28d81ef..e630190 100644 --- a/src/blazingmq/__init__.py +++ b/src/blazingmq/__init__.py @@ -1,4 +1,4 @@ -# Copyright 2019-2023 Bloomberg Finance L.P. +# Copyright 2019-2026 Bloomberg Finance L.P. # SPDX-License-Identifier: Apache-2.0 # # Licensed under the Apache License, Version 2.0 (the "License"); diff --git a/src/blazingmq/_ext.pyi b/src/blazingmq/_ext.pyi index 7714270..1637d33 100644 --- a/src/blazingmq/_ext.pyi +++ b/src/blazingmq/_ext.pyi @@ -1,4 +1,4 @@ -# Copyright 2019-2023 Bloomberg Finance L.P. +# Copyright 2019-2026 Bloomberg Finance L.P. # SPDX-License-Identifier: Apache-2.0 # # Licensed under the Apache License, Version 2.0 (the "License"); diff --git a/src/blazingmq/_ext.pyx b/src/blazingmq/_ext.pyx index befa011..30a3b76 100644 --- a/src/blazingmq/_ext.pyx +++ b/src/blazingmq/_ext.pyx @@ -1,4 +1,4 @@ -# Copyright 2019-2023 Bloomberg Finance L.P. +# Copyright 2019-2026 Bloomberg Finance L.P. # SPDX-License-Identifier: Apache-2.0 # # Licensed under the Apache License, Version 2.0 (the "License"); diff --git a/src/blazingmq/_session.py b/src/blazingmq/_session.py index 6ef216e..ecba670 100644 --- a/src/blazingmq/_session.py +++ b/src/blazingmq/_session.py @@ -1,4 +1,4 @@ -# Copyright 2019-2023 Bloomberg Finance L.P. +# Copyright 2019-2026 Bloomberg Finance L.P. # SPDX-License-Identifier: Apache-2.0 # # Licensed under the Apache License, Version 2.0 (the "License"); diff --git a/src/blazingmq/_typing.py b/src/blazingmq/_typing.py index 1110405..c6fe00a 100644 --- a/src/blazingmq/_typing.py +++ b/src/blazingmq/_typing.py @@ -1,4 +1,4 @@ -# Copyright 2019-2023 Bloomberg Finance L.P. +# Copyright 2019-2026 Bloomberg Finance L.P. # SPDX-License-Identifier: Apache-2.0 # # Licensed under the Apache License, Version 2.0 (the "License"); diff --git a/src/cpp/pybmq_session.cpp b/src/cpp/pybmq_session.cpp index 1262dae..d88c3f5 100644 --- a/src/cpp/pybmq_session.cpp +++ b/src/cpp/pybmq_session.cpp @@ -1,4 +1,4 @@ -// Copyright 2019-2023 Bloomberg Finance L.P. +// Copyright 2019-2026 Bloomberg Finance L.P. // SPDX-License-Identifier: Apache-2.0 // // Licensed under the Apache License, Version 2.0 (the "License"); @@ -182,12 +182,10 @@ Session::Session( d_message_compression_type = config.message_compression_type; bmqt::SessionOptions::AuthnCredentialCb cpp_callback; - bool has_auth_callback = false; if (authn_credential_cb != NULL && authn_credential_cb != Py_None) { d_authn_credential_cb = authn_credential_cb; cpp_callback = AuthnCredentialCbFunctor(d_authn_credential_cb); - has_auth_callback = true; } { @@ -215,7 +213,7 @@ Session::Session( config.event_queue_watermarks.value().second); } - if (has_auth_callback) { + if (cpp_callback) { options.setAuthnCredentialCb(cpp_callback); } diff --git a/src/cpp/pybmq_session.h b/src/cpp/pybmq_session.h index e7795ff..51fc1bd 100644 --- a/src/cpp/pybmq_session.h +++ b/src/cpp/pybmq_session.h @@ -1,4 +1,4 @@ -// Copyright 2019-2023 Bloomberg Finance L.P. +// Copyright 2019-2026 Bloomberg Finance L.P. // SPDX-License-Identifier: Apache-2.0 // // Licensed under the Apache License, Version 2.0 (the "License"); diff --git a/src/declarations/pybmq.pxd b/src/declarations/pybmq.pxd index e0722d5..699c542 100644 --- a/src/declarations/pybmq.pxd +++ b/src/declarations/pybmq.pxd @@ -1,4 +1,4 @@ -# Copyright 2019-2023 Bloomberg Finance L.P. +# Copyright 2019-2026 Bloomberg Finance L.P. # SPDX-License-Identifier: Apache-2.0 # # Licensed under the Apache License, Version 2.0 (the "License"); diff --git a/tests/unit/test_session.py b/tests/unit/test_session.py index 2319527..3858369 100644 --- a/tests/unit/test_session.py +++ b/tests/unit/test_session.py @@ -1,4 +1,4 @@ -# Copyright 2019-2023 Bloomberg Finance L.P. +# Copyright 2019-2026 Bloomberg Finance L.P. # SPDX-License-Identifier: Apache-2.0 # # Licensed under the Apache License, Version 2.0 (the "License"); diff --git a/tests/unit/test_session_options.py b/tests/unit/test_session_options.py index 0f04fbf..ec575b7 100644 --- a/tests/unit/test_session_options.py +++ b/tests/unit/test_session_options.py @@ -1,4 +1,4 @@ -# Copyright 2019-2023 Bloomberg Finance L.P. +# Copyright 2019-2026 Bloomberg Finance L.P. # SPDX-License-Identifier: Apache-2.0 # # Licensed under the Apache License, Version 2.0 (the "License");