From 916f30f3c58476dca5d62e503e330ed862d0c079 Mon Sep 17 00:00:00 2001 From: Emelia Lei Date: Mon, 8 Dec 2025 17:25:10 -0500 Subject: [PATCH 01/22] provide a way to create credential Signed-off-by: Emelia Lei --- src/blazingmq/__init__.py | 2 + src/blazingmq/_authncb.py | 32 ++++++++++++++ src/blazingmq/_ext.pyi | 4 ++ src/blazingmq/_ext.pyx | 37 ++++++++++++++++ src/blazingmq/_session.py | 4 ++ src/cpp/pybmq_session.cpp | 81 +++++++++++++++++++++++++++++++++-- src/cpp/pybmq_session.h | 6 +++ src/declarations/bmq/bmqt.pxd | 10 +++++ src/declarations/pybmq.pxd | 1 + 9 files changed, 174 insertions(+), 3 deletions(-) create mode 100644 src/blazingmq/_authncb.py diff --git a/src/blazingmq/__init__.py b/src/blazingmq/__init__.py index 910d1ad..9f18654 100644 --- a/src/blazingmq/__init__.py +++ b/src/blazingmq/__init__.py @@ -16,6 +16,7 @@ from . import exceptions from . import session_events from ._about import __version__ +from ._authncb import BasicAuthnCredentialCb from ._enums import AckStatus from ._enums import CompressionAlgorithmType from ._enums import PropertyType @@ -34,6 +35,7 @@ __all__ = [ "Ack", "AckStatus", + "BasicAuthnCredentialCb", "BasicHealthMonitor", "CompressionAlgorithmType", "Error", diff --git a/src/blazingmq/_authncb.py b/src/blazingmq/_authncb.py new file mode 100644 index 0000000..a5c2390 --- /dev/null +++ b/src/blazingmq/_authncb.py @@ -0,0 +1,32 @@ +# Copyright 2019-2023 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 __future__ import annotations +from typing import Callable, Optional, Tuple +from ._ext import FakeAuthnCredentialCb + +CredentialTuple = Tuple[str, bytes] + + +class BasicAuthnCredentialCb: + """Wrap a Python callable returning (mechanism:str, data:bytes) or None.""" + + def __init__(self, callback: Callable[[], Optional[CredentialTuple]]): + if not callable(callback): + raise TypeError("callback must be callable") + self._authncb = FakeAuthnCredentialCb(callback) + + def __repr__(self) -> str: + return "BasicAuthnCredentialCb(...)" 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 e8ecb9d..36a324d 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 @@ -153,6 +156,38 @@ cdef class FakeHostHealthMonitor: self._monitor.get().setState(HostHealthState.e_UNHEALTHY) +cdef class FakeAuthnCredentialCb: + cdef object _callback # Store the Python callable + + def __cinit__(self, callback): + if not callable(callback): + raise TypeError("callback must be callable") + 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 as e: + # Log error or handle as needed + LOGGER.exception("Error in authentication credential callback") + return None + + cdef class Session: cdef object __weakref__ cdef NativeSession* _session @@ -174,6 +209,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 @@ -225,6 +261,7 @@ cdef class Session: session_cb, message_cb, ack_cb, + fake_authn_credential_cb, c_broker_uri, c_script_name, COMPRESSION_ALGO_FROM_PY_MAPPING[message_compression_algorithm], diff --git a/src/blazingmq/_session.py b/src/blazingmq/_session.py index e0c09e3..1c07107 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 ._authncb import BasicAuthnCredentialCb from ._timeouts import Timeouts from ._typing import PropertyTypeDict from ._typing import PropertyValueDict @@ -418,6 +419,7 @@ def __init__( ), timeout: Union[Timeouts, float] = DEFAULT_TIMEOUT, host_health_monitor: Union[BasicHealthMonitor, None] = (DefaultMonitor()), + authn_credential_cb: Optional[BasicAuthnCredentialCb] = None, num_processing_threads: Optional[int] = None, blob_buffer_size: Optional[int] = None, channel_high_watermark: Optional[int] = None, @@ -433,6 +435,7 @@ def __init__( monitor_host_health = host_health_monitor is not None fake_host_health_monitor = getattr(host_health_monitor, "_monitor", None) + fake_authn_credential_cb = getattr(authn_credential_cb, "_authncb", None) self._has_no_on_message = on_message is None @@ -459,6 +462,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_cb, ) self._ext.set_owned_by_session() diff --git a/src/cpp/pybmq_session.cpp b/src/cpp/pybmq_session.cpp index acab799..d5b0d59 100644 --- a/src/cpp/pybmq_session.cpp +++ b/src/cpp/pybmq_session.cpp @@ -15,6 +15,7 @@ #include +#include #include #include #include @@ -77,13 +78,14 @@ Session::Session( PyObject* py_session_event_callback, PyObject* py_message_event_callback, PyObject* py_ack_event_callback, + PyObject* fake_authn_credential_cb, 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, + bsl::optional> event_queue_watermarks, const bsls::TimeInterval& stats_dump_interval, const bsls::TimeInterval& connect_timeout, const bsls::TimeInterval& disconnect_timeout, @@ -119,6 +121,74 @@ Session::Session( } d_message_compression_type = 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; @@ -144,6 +214,11 @@ Session::Session( event_queue_watermarks.value().second); } + if (has_auth_callback) { + // TODO: This will only compile with setAuthnCredentialCb in SessionOptions + options.setAuthnCredentialCb(cpp_callback); + } + if (stats_dump_interval != bsls::TimeInterval()) { options.setStatsDumpInterval(stats_dump_interval); } @@ -529,8 +604,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 f37a407..0a5f223 100644 --- a/src/cpp/pybmq_session.h +++ b/src/cpp/pybmq_session.h @@ -21,6 +21,7 @@ #include #include +#include #include #include @@ -47,10 +48,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 char* broker_uri, const char* script_name, bmqt::CompressionAlgorithmType::Enum message_compression_type, 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 9ba293e..ba64645 100644 --- a/src/declarations/pybmq.pxd +++ b/src/declarations/pybmq.pxd @@ -39,6 +39,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 char* broker_uri, const char* script_name, CompressionAlgorithmType message_compression_algorithm, From 36a76ec17cb8867a5966866aa60ef6373056da93 Mon Sep 17 00:00:00 2001 From: "Patrick M. Niedzielski" Date: Mon, 22 Jun 2026 15:49:20 -0400 Subject: [PATCH 02/22] 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 d5b0d59..4cc537a 100644 --- a/src/cpp/pybmq_session.cpp +++ b/src/cpp/pybmq_session.cpp @@ -131,6 +131,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 { @@ -215,7 +216,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 0a5f223..38f4c40 100644 --- a/src/cpp/pybmq_session.h +++ b/src/cpp/pybmq_session.h @@ -48,10 +48,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 24efeed00b7c19e500fc4c4aea6a0d777dd75365 Mon Sep 17 00:00:00 2001 From: "Patrick M. Niedzielski" Date: Tue, 23 Jun 2026 17:32:04 -0400 Subject: [PATCH 03/22] Remove unused variable binding `LOGGER.exception` already captures and prints the currently in-flight exception. This patch removes the unused variable binding on the exception. 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 36a324d..4c34cf7 100644 --- a/src/blazingmq/_ext.pyx +++ b/src/blazingmq/_ext.pyx @@ -182,7 +182,7 @@ cdef class FakeAuthnCredentialCb: # Return as-is, let C++ side handle conversion return result - except Exception as e: + except Exception: # Log error or handle as needed LOGGER.exception("Error in authentication credential callback") return None From 1fe5bf956deb42fc40d419467ee9bdfe86764aad Mon Sep 17 00:00:00 2001 From: "Patrick M. Niedzielski" Date: Tue, 23 Jun 2026 17:38:11 -0400 Subject: [PATCH 04/22] Remove duplicate error checking We check this error in different layers, along with providing type annotations for use with something like mypy. The closest parallel example we have is `FakeHostHealthMonitor`, which only does runtime type checking at the highest layer, in pure Python. This patch removes the duplicate error checking from the Cython layer. Signed-off-by: Patrick M. Niedzielski --- src/blazingmq/_ext.pyx | 2 -- 1 file changed, 2 deletions(-) diff --git a/src/blazingmq/_ext.pyx b/src/blazingmq/_ext.pyx index 4c34cf7..abbd268 100644 --- a/src/blazingmq/_ext.pyx +++ b/src/blazingmq/_ext.pyx @@ -160,8 +160,6 @@ cdef class FakeAuthnCredentialCb: cdef object _callback # Store the Python callable def __cinit__(self, callback): - if not callable(callback): - raise TypeError("callback must be callable") self._callback = callback # This method will be called by C++ code via PyObject_CallMethod From c6148f9bf603a3f6758dd8351ee54d7aa72bfec0 Mon Sep 17 00:00:00 2001 From: "Patrick M. Niedzielski" Date: Tue, 23 Jun 2026 17:49:43 -0400 Subject: [PATCH 05/22] Remove `BasicAuthnCredentialCb` MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit This type doesn’t do much of anything on its own, and only makes it harder for a user to provide their own authentication credential provider, by forcing them to wrap whatever function they have in this type. This patch removes the class and lets users just pass any `Callable` of the right signature when constructing a `Session`. Signed-off-by: Patrick M. Niedzielski --- src/blazingmq/__init__.py | 2 -- src/blazingmq/_authncb.py | 32 -------------------------------- src/blazingmq/_session.py | 10 +++++++--- 3 files changed, 7 insertions(+), 37 deletions(-) delete mode 100644 src/blazingmq/_authncb.py diff --git a/src/blazingmq/__init__.py b/src/blazingmq/__init__.py index 9f18654..910d1ad 100644 --- a/src/blazingmq/__init__.py +++ b/src/blazingmq/__init__.py @@ -16,7 +16,6 @@ from . import exceptions from . import session_events from ._about import __version__ -from ._authncb import BasicAuthnCredentialCb from ._enums import AckStatus from ._enums import CompressionAlgorithmType from ._enums import PropertyType @@ -35,7 +34,6 @@ __all__ = [ "Ack", "AckStatus", - "BasicAuthnCredentialCb", "BasicHealthMonitor", "CompressionAlgorithmType", "Error", diff --git a/src/blazingmq/_authncb.py b/src/blazingmq/_authncb.py deleted file mode 100644 index a5c2390..0000000 --- a/src/blazingmq/_authncb.py +++ /dev/null @@ -1,32 +0,0 @@ -# Copyright 2019-2023 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 __future__ import annotations -from typing import Callable, Optional, Tuple -from ._ext import FakeAuthnCredentialCb - -CredentialTuple = Tuple[str, bytes] - - -class BasicAuthnCredentialCb: - """Wrap a Python callable returning (mechanism:str, data:bytes) or None.""" - - def __init__(self, callback: Callable[[], Optional[CredentialTuple]]): - if not callable(callback): - raise TypeError("callback must be callable") - self._authncb = FakeAuthnCredentialCb(callback) - - def __repr__(self) -> str: - return "BasicAuthnCredentialCb(...)" diff --git a/src/blazingmq/_session.py b/src/blazingmq/_session.py index 1c07107..1708c38 100644 --- a/src/blazingmq/_session.py +++ b/src/blazingmq/_session.py @@ -35,7 +35,7 @@ from ._messages import Message from ._messages import MessageHandle from ._monitors import BasicHealthMonitor -from ._authncb import BasicAuthnCredentialCb +from ._ext import FakeAuthnCredentialCb from ._timeouts import Timeouts from ._typing import PropertyTypeDict from ._typing import PropertyValueDict @@ -419,7 +419,7 @@ def __init__( ), timeout: Union[Timeouts, float] = DEFAULT_TIMEOUT, host_health_monitor: Union[BasicHealthMonitor, None] = (DefaultMonitor()), - authn_credential_cb: Optional[BasicAuthnCredentialCb] = None, + authn_credential_cb: Optional[Callable] = None, num_processing_threads: Optional[int] = None, blob_buffer_size: Optional[int] = None, channel_high_watermark: Optional[int] = None, @@ -435,7 +435,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_cb = getattr(authn_credential_cb, "_authncb", None) + fake_authn_credential_cb = ( + FakeAuthnCredentialCb(authn_credential_cb) + if authn_credential_cb is not None + else None + ) self._has_no_on_message = on_message is None From 7694d06ed96984369fbb25429588869ff37f87db Mon Sep 17 00:00:00 2001 From: "Patrick M. Niedzielski" Date: Tue, 23 Jun 2026 17:51:26 -0400 Subject: [PATCH 06/22] 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 1708c38..4eb6a9d 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",) @@ -419,7 +423,7 @@ def __init__( ), timeout: Union[Timeouts, float] = DEFAULT_TIMEOUT, host_health_monitor: Union[BasicHealthMonitor, None] = (DefaultMonitor()), - authn_credential_cb: Optional[Callable] = None, + authn_credential_cb: Optional[Callable] = (DefaultAuthnCredentialCb()), num_processing_threads: Optional[int] = None, blob_buffer_size: Optional[int] = None, channel_high_watermark: Optional[int] = None, From aa0b762ef097d00867bf7aa3511bd83ba076812a Mon Sep 17 00:00:00 2001 From: "Patrick M. Niedzielski" Date: Tue, 23 Jun 2026 17:54:20 -0400 Subject: [PATCH 07/22] Add `authn_credential_cb` to `SessionOptions` Signed-off-by: Patrick M. Niedzielski --- src/blazingmq/_session.py | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/src/blazingmq/_session.py b/src/blazingmq/_session.py index 4eb6a9d..9596192 100644 --- a/src/blazingmq/_session.py +++ b/src/blazingmq/_session.py @@ -300,6 +300,7 @@ def __init__( message_compression_algorithm: Optional[CompressionAlgorithmType] = None, timeouts: Optional[Timeouts] = None, host_health_monitor: Union[BasicHealthMonitor, None] = (DefaultMonitor()), + authn_credential_cb: Optional[Callable] = (DefaultAuthnCredentialCb()), num_processing_threads: Optional[int] = None, blob_buffer_size: Optional[int] = None, channel_high_watermark: Optional[int] = None, @@ -309,6 +310,7 @@ def __init__( self.message_compression_algorithm = message_compression_algorithm self.timeouts = timeouts self.host_health_monitor = host_health_monitor + self.authn_credential_cb = authn_credential_cb self.num_processing_threads = num_processing_threads self.blob_buffer_size = blob_buffer_size self.channel_high_watermark = channel_high_watermark @@ -322,6 +324,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_cb == other.authn_credential_cb 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 @@ -337,6 +340,7 @@ def __repr__(self) -> str: "message_compression_algorithm", "timeouts", "host_health_monitor", + "authn_credential_cb", "num_processing_threads", "blob_buffer_size", "channel_high_watermark", @@ -523,6 +527,7 @@ def with_options( message_compression_algorithm, DEFAULT_TIMEOUT, session_options.host_health_monitor, + session_options.authn_credential_cb, session_options.num_processing_threads, session_options.blob_buffer_size, session_options.channel_high_watermark, @@ -537,6 +542,7 @@ def with_options( message_compression_algorithm, session_options.timeouts, session_options.host_health_monitor, + session_options.authn_credential_cb, session_options.num_processing_threads, session_options.blob_buffer_size, session_options.channel_high_watermark, From 30b9653dd131747d59046c885ba3edc238cb0391 Mon Sep 17 00:00:00 2001 From: "Patrick M. Niedzielski" Date: Tue, 23 Jun 2026 17:55:29 -0400 Subject: [PATCH 08/22] Document `authn_credential_cb` Signed-off-by: Patrick M. Niedzielski --- src/blazingmq/_session.py | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/src/blazingmq/_session.py b/src/blazingmq/_session.py index 9596192..ba97dd8 100644 --- a/src/blazingmq/_session.py +++ b/src/blazingmq/_session.py @@ -270,6 +270,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_cb: + 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. @@ -388,6 +393,10 @@ class Session: `.HostHealthRestored` events will never be emitted, and the *suspends_on_bad_host_health* option of `QueueOptions` cannot be used. + authn_credential_cb: 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 30b86b185295ea4468539206869a1f67e2b0d5ee Mon Sep 17 00:00:00 2001 From: "Patrick M. Niedzielski" Date: Tue, 23 Jun 2026 17:57:54 -0400 Subject: [PATCH 09/22] Rename `authn_credential_cb` to something more Pythonic Signed-off-by: Patrick M. Niedzielski --- src/blazingmq/_session.py | 28 ++++++++++++++-------------- 1 file changed, 14 insertions(+), 14 deletions(-) diff --git a/src/blazingmq/_session.py b/src/blazingmq/_session.py index ba97dd8..d18bdb4 100644 --- a/src/blazingmq/_session.py +++ b/src/blazingmq/_session.py @@ -53,7 +53,7 @@ def DefaultMonitor() -> Union[BasicHealthMonitor, None]: return None -def DefaultAuthnCredentialCb() -> Optional[Callable]: +def DefaultAuthnCredentialProvider() -> Optional[Callable]: return None @@ -270,7 +270,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_cb: + 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 @@ -305,7 +305,7 @@ def __init__( message_compression_algorithm: Optional[CompressionAlgorithmType] = None, timeouts: Optional[Timeouts] = None, host_health_monitor: Union[BasicHealthMonitor, None] = (DefaultMonitor()), - authn_credential_cb: Optional[Callable] = (DefaultAuthnCredentialCb()), + authn_credential_provider: Optional[Callable] = (DefaultAuthnCredentialProvider()), num_processing_threads: Optional[int] = None, blob_buffer_size: Optional[int] = None, channel_high_watermark: Optional[int] = None, @@ -315,7 +315,7 @@ def __init__( self.message_compression_algorithm = message_compression_algorithm self.timeouts = timeouts self.host_health_monitor = host_health_monitor - self.authn_credential_cb = authn_credential_cb + 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 @@ -329,7 +329,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_cb == other.authn_credential_cb + 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 @@ -345,7 +345,7 @@ def __repr__(self) -> str: "message_compression_algorithm", "timeouts", "host_health_monitor", - "authn_credential_cb", + "authn_credential_provider", "num_processing_threads", "blob_buffer_size", "channel_high_watermark", @@ -393,7 +393,7 @@ class Session: `.HostHealthRestored` events will never be emitted, and the *suspends_on_bad_host_health* option of `QueueOptions` cannot be used. - authn_credential_cb: an optional callable that returns authentication + 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. @@ -436,7 +436,7 @@ def __init__( ), timeout: Union[Timeouts, float] = DEFAULT_TIMEOUT, host_health_monitor: Union[BasicHealthMonitor, None] = (DefaultMonitor()), - authn_credential_cb: Optional[Callable] = (DefaultAuthnCredentialCb()), + authn_credential_provider: Optional[Callable] = (DefaultAuthnCredentialProvider()), num_processing_threads: Optional[int] = None, blob_buffer_size: Optional[int] = None, channel_high_watermark: Optional[int] = None, @@ -452,9 +452,9 @@ def __init__( monitor_host_health = host_health_monitor is not None fake_host_health_monitor = getattr(host_health_monitor, "_monitor", None) - fake_authn_credential_cb = ( - FakeAuthnCredentialCb(authn_credential_cb) - if authn_credential_cb is not None + fake_authn_credential_provider = ( + FakeAuthnCredentialCb(authn_credential_provider) + if authn_credential_provider is not None else None ) @@ -483,7 +483,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_cb, + fake_authn_credential_cb=fake_authn_credential_provider, ) self._ext.set_owned_by_session() @@ -536,7 +536,7 @@ def with_options( message_compression_algorithm, DEFAULT_TIMEOUT, session_options.host_health_monitor, - session_options.authn_credential_cb, + session_options.authn_credential_provider, session_options.num_processing_threads, session_options.blob_buffer_size, session_options.channel_high_watermark, @@ -551,7 +551,7 @@ def with_options( message_compression_algorithm, session_options.timeouts, session_options.host_health_monitor, - session_options.authn_credential_cb, + session_options.authn_credential_provider, session_options.num_processing_threads, session_options.blob_buffer_size, session_options.channel_high_watermark, From 96a473158f7902ea99b111c897c2251261e98930 Mon Sep 17 00:00:00 2001 From: "Patrick M. Niedzielski" Date: Fri, 26 Jun 2026 12:13:38 -0400 Subject: [PATCH 10/22] 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 4cc537a..b6d7b71 100644 --- a/src/cpp/pybmq_session.cpp +++ b/src/cpp/pybmq_session.cpp @@ -179,13 +179,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 e95583e4420d5def08fbf78e820e1f6e6d91ba9e Mon Sep 17 00:00:00 2001 From: "Patrick M. Niedzielski" Date: Fri, 26 Jun 2026 12:20:05 -0400 Subject: [PATCH 11/22] 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 b6d7b71..21dfaa9 100644 --- a/src/cpp/pybmq_session.cpp +++ b/src/cpp/pybmq_session.cpp @@ -122,7 +122,7 @@ Session::Session( d_message_compression_type = 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 4df19ed46ce4c51abbe2d1213540589f2be3ccfa Mon Sep 17 00:00:00 2001 From: "Patrick M. Niedzielski" Date: Fri, 26 Jun 2026 15:41:21 -0400 Subject: [PATCH 12/22] 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 d1ec4f4d155abd7c2da31d30f5c95bd0bbcc6cad Mon Sep 17 00:00:00 2001 From: "Patrick M. Niedzielski" Date: Fri, 26 Jun 2026 15:45:09 -0400 Subject: [PATCH 13/22] 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 266645fd772567e7e34a7c7c709a4da5b7bc0d3d Mon Sep 17 00:00:00 2001 From: "Patrick M. Niedzielski" Date: Fri, 26 Jun 2026 15:54:03 -0400 Subject: [PATCH 14/22] 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 02041f74154bed495abc1752099a6f17cc77acf4 Mon Sep 17 00:00:00 2001 From: "Patrick M. Niedzielski" Date: Fri, 26 Jun 2026 15:55:11 -0400 Subject: [PATCH 15/22] 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 d18bdb4..3d6b2ae 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] = (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, From 88ab05748a8868df817c04036c46f45cd2756358 Mon Sep 17 00:00:00 2001 From: "Patrick M. Niedzielski" Date: Fri, 26 Jun 2026 16:09:53 -0400 Subject: [PATCH 16/22] 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 36d3f805523eb68f1ade8dc7adacfde81c9ddc54 Mon Sep 17 00:00:00 2001 From: "Patrick M. Niedzielski" Date: Tue, 30 Jun 2026 14:37:26 -0400 Subject: [PATCH 17/22] Rename `FakeAuthnCredentialCb` Signed-off-by: Patrick M. Niedzielski --- src/blazingmq/_ext.pyi | 4 +- src/blazingmq/_ext.pyx | 4 +- src/blazingmq/_session.py | 4 +- .../unit/test_authn_credential_cb_adapter.py | 114 ++++++++++++++++++ 4 files changed, 120 insertions(+), 6 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..c0c3b4a 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, + fake_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 abbd268..c5fde7f 100644 --- a/src/blazingmq/_ext.pyx +++ b/src/blazingmq/_ext.pyx @@ -156,7 +156,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): @@ -207,7 +207,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, + fake_authn_credential_cb: AuthnCredentialCbAdapter = None, _mock: Optional[object] = None, ) -> None: cdef shared_ptr[ManualHostHealthMonitor] fake_host_health_monitor_sp diff --git a/src/blazingmq/_session.py b/src/blazingmq/_session.py index 3d6b2ae..b09f06a 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 @@ -457,7 +457,7 @@ 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) + AuthnCredentialCbAdapter(authn_credential_provider) if authn_credential_provider is not None else None ) 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 From 0a066f41ac6bf841a5d69055b2ad84b0a9b4a52f Mon Sep 17 00:00:00 2001 From: "Patrick M. Niedzielski" Date: Tue, 30 Jun 2026 14:37:42 -0400 Subject: [PATCH 18/22] 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 c5fde7f..5b94369 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 ac486a954bc06cb449a384c1c9d8fcc89fadd5db Mon Sep 17 00:00:00 2001 From: "Patrick M. Niedzielski" Date: Tue, 30 Jun 2026 15:02:54 -0400 Subject: [PATCH 19/22] =?UTF-8?q?Fully=20rename=20`fake=5Fauthn=5F?= =?UTF-8?q?=E2=80=A6`?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Patrick M. Niedzielski --- src/blazingmq/_ext.pyi | 2 +- src/blazingmq/_ext.pyx | 4 ++-- src/blazingmq/_session.py | 4 ++-- src/cpp/pybmq_session.cpp | 10 +++++----- src/cpp/pybmq_session.h | 2 +- src/declarations/pybmq.pxd | 2 +- tests/unit/test_session.py | 26 +++++++++++++------------- 7 files changed, 25 insertions(+), 25 deletions(-) diff --git a/src/blazingmq/_ext.pyi b/src/blazingmq/_ext.pyi index c0c3b4a..7714270 100644 --- a/src/blazingmq/_ext.pyi +++ b/src/blazingmq/_ext.pyi @@ -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[AuthnCredentialCbAdapter] = 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 5b94369..e21d89a 100644 --- a/src/blazingmq/_ext.pyx +++ b/src/blazingmq/_ext.pyx @@ -207,7 +207,7 @@ cdef class Session: timeouts: _timeouts.Timeouts = _timeouts.Timeouts(), monitor_host_health: bool = False, fake_host_health_monitor: FakeHostHealthMonitor = None, - fake_authn_credential_cb: AuthnCredentialCbAdapter = None, + authn_credential_cb: AuthnCredentialCbAdapter = None, _mock: Optional[object] = None, ) -> None: cdef shared_ptr[ManualHostHealthMonitor] fake_host_health_monitor_sp @@ -259,7 +259,7 @@ cdef class Session: session_cb, message_cb, ack_cb, - fake_authn_credential_cb, + authn_credential_cb, c_broker_uri, c_script_name, COMPRESSION_ALGO_FROM_PY_MAPPING[message_compression_algorithm], diff --git a/src/blazingmq/_session.py b/src/blazingmq/_session.py index b09f06a..be1eba4 100644 --- a/src/blazingmq/_session.py +++ b/src/blazingmq/_session.py @@ -456,7 +456,7 @@ 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 = ( + 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 21dfaa9..97627c5 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 char* broker_uri, const char* script_name, bmqt::CompressionAlgorithmType::Enum message_compression_type, @@ -125,22 +125,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 38f4c40..2c30525 100644 --- a/src/cpp/pybmq_session.h +++ b/src/cpp/pybmq_session.h @@ -52,7 +52,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 char* broker_uri, const char* script_name, bmqt::CompressionAlgorithmType::Enum message_compression_type, diff --git a/src/declarations/pybmq.pxd b/src/declarations/pybmq.pxd index ba64645..e25db5d 100644 --- a/src/declarations/pybmq.pxd +++ b/src/declarations/pybmq.pxd @@ -39,7 +39,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 char* broker_uri, const char* script_name, CompressionAlgorithmType message_compression_algorithm, 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 9f50ec58a5cf19f23b1df41b3f204c08ff2e1893 Mon Sep 17 00:00:00 2001 From: "Patrick M. Niedzielski" Date: Tue, 30 Jun 2026 15:19:39 -0400 Subject: [PATCH 20/22] Add `AuthnCredentialProvider` type alias Signed-off-by: Patrick M. Niedzielski --- src/blazingmq/__init__.py | 2 ++ src/blazingmq/_session.py | 7 ++++--- src/blazingmq/_typing.py | 8 ++++++++ 3 files changed, 14 insertions(+), 3 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 be1eba4..417986d 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 DefaultAuthnCredentialProvider() -> 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,7 +439,7 @@ def __init__( ), timeout: Union[Timeouts, float] = DEFAULT_TIMEOUT, host_health_monitor: Union[BasicHealthMonitor, None] = (DefaultMonitor()), - authn_credential_provider: Optional[Callable] = ( + authn_credential_provider: Optional[AuthnCredentialProvider] = ( DefaultAuthnCredentialProvider() ), num_processing_threads: 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 cfd0a2c3bb38cb08c86cf46342310c12eb3f4fed Mon Sep 17 00:00:00 2001 From: "Patrick M. Niedzielski" Date: Tue, 30 Jun 2026 18:29:04 -0400 Subject: [PATCH 21/22] 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 97627c5..9f08c83 100644 --- a/src/cpp/pybmq_session.cpp +++ b/src/cpp/pybmq_session.cpp @@ -183,7 +183,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 dc5e839f0f553758849bdd3844581d84f9592f0d Mon Sep 17 00:00:00 2001 From: "Patrick M. Niedzielski" Date: Wed, 1 Jul 2026 11:02:49 -0400 Subject: [PATCH 22/22] Add documentation for `AuthnCredentialProvider` Signed-off-by: Patrick M. Niedzielski --- docs/api_reference.rst | 2 ++ docs/conf.py | 3 +++ 2 files changed, 5 insertions(+) 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/docs/conf.py b/docs/conf.py index 69e98e9..6fd0fb8 100644 --- a/docs/conf.py +++ b/docs/conf.py @@ -185,6 +185,9 @@ autodoc_typehints = 'description' autodoc_typehints_description_target = 'documented' +autodoc_type_aliases = { + "AuthnCredentialProvider": "blazingmq.AuthnCredentialProvider", +} def document_as_keyword_only(app, what, name, obj, options, signature, return_hint):