diff --git a/news/71.feature.rst b/news/71.feature.rst new file mode 100644 index 0000000..2e7cd76 --- /dev/null +++ b/news/71.feature.rst @@ -0,0 +1 @@ +Added user agent prefix string to session negotiation diff --git a/src/blazingmq/_ext.pyi b/src/blazingmq/_ext.pyi index 14d2478..b455d49 100644 --- a/src/blazingmq/_ext.pyi +++ b/src/blazingmq/_ext.pyi @@ -53,6 +53,7 @@ class Session: timeouts: Timeouts = Timeouts(), monitor_host_health: bool = False, fake_host_health_monitor: Optional[FakeHostHealthMonitor] = None, + user_agent_prefix: bytes = b"", ) -> None: ... def stop(self) -> None: ... def open_queue_sync( diff --git a/src/blazingmq/_ext.pyx b/src/blazingmq/_ext.pyx index e8ecb9d..7fde14a 100644 --- a/src/blazingmq/_ext.pyx +++ b/src/blazingmq/_ext.pyx @@ -175,6 +175,7 @@ cdef class Session: monitor_host_health: bool = False, fake_host_health_monitor: FakeHostHealthMonitor = None, _mock: Optional[object] = None, + user_agent_prefix: bytes = b"", ) -> None: cdef shared_ptr[ManualHostHealthMonitor] fake_host_health_monitor_sp cdef optional[int] c_num_processing_threads @@ -221,6 +222,7 @@ cdef class Session: cdef char *c_broker_uri = broker script_name = _script_name.get_script_name() cdef char *c_script_name = script_name + cdef char *c_user_agent_prefix = user_agent_prefix self._session = new NativeSession( session_cb, message_cb, @@ -242,7 +244,8 @@ cdef class Session: fake_host_health_monitor_sp, Error, BrokerTimeoutError, - _mock) + _mock, + c_user_agent_prefix) self._session.start(c_connect_timeout) atexit.register(ensure_stop_session_impl, weakref.ref(self)) diff --git a/src/blazingmq/_session.py b/src/blazingmq/_session.py index e0c09e3..ae2d8cd 100644 --- a/src/blazingmq/_session.py +++ b/src/blazingmq/_session.py @@ -15,6 +15,7 @@ from __future__ import annotations +import platform from typing import Any from typing import Callable from typing import Dict @@ -23,6 +24,7 @@ from typing import Union from . import _six as six +from ._about import __version__ from ._enums import CompressionAlgorithmType from ._enums import PropertyType from ._ext import DEFAULT_CONSUMER_PRIORITY @@ -72,6 +74,41 @@ def _validate_timeouts(timeouts: Timeouts) -> Timeouts: ) +def _make_user_agent_prefix(user_agent_prefix: Optional[bytes]) -> bytes: + """Validate and construct a user agent string for use by the Cython layer. + + If the user agent prefix string is more than 96 bytes or contains + non-printable characters, raise a `ValueError`. Otherwise, append the SDK + identifier and return the full user agent string. + """ + if user_agent_prefix is None: + user_agent_prefix = b"" + if len(user_agent_prefix) > 96: + raise ValueError( + f"user_agent_prefix ({user_agent_prefix!r}) must be at most 96 " + f"bytes (is {len(user_agent_prefix)} bytes)" + ) + if ( + not user_agent_prefix.isascii() + or not user_agent_prefix.decode("ascii").isprintable() + ): + raise ValueError( + f"user_agent_prefix ({user_agent_prefix!r}) must only contain " + f"printable characters" + ) + return _construct_user_agent_prefix(user_agent_prefix) + + +def _construct_user_agent_prefix(user_agent_prefix: bytes) -> bytes: + """Construct the user agent prefix for use by the Cython layer.""" + python_version = platform.python_version().encode("ascii", errors="strict") + blazingmq_version = __version__.encode("ascii", errors="strict") + sdk_identifier = b"blazingmq(python" + python_version + b"):" + blazingmq_version + if user_agent_prefix: + return user_agent_prefix + b" " + sdk_identifier + return sdk_identifier + + def _convert_timeout(timeout: Optional[float]) -> Optional[float]: """Convert the timeout for use by the Cython layer. @@ -288,6 +325,12 @@ 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]``. + user_agent_prefix: + Bytestring to include in the user agent for broker telemetry. This + string must only contain printable characters and must be at most + 96 bytes long. This is provided for libraries that are wrapping + this SDK. Applications directly using the SDK are encouraged *NOT* + to set this value. """ def __init__( @@ -300,6 +343,7 @@ def __init__( channel_high_watermark: Optional[int] = None, event_queue_watermarks: Optional[tuple[int, int]] = None, stats_dump_interval: Optional[float] = None, + user_agent_prefix: Optional[bytes] = None, ) -> None: self.message_compression_algorithm = message_compression_algorithm self.timeouts = timeouts @@ -309,6 +353,7 @@ def __init__( self.channel_high_watermark = channel_high_watermark self.event_queue_watermarks = event_queue_watermarks self.stats_dump_interval = stats_dump_interval + self.user_agent_prefix = user_agent_prefix def __eq__(self, other: object) -> bool: if not isinstance(other, SessionOptions): @@ -322,6 +367,7 @@ def __eq__(self, other: object) -> bool: 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.user_agent_prefix == other.user_agent_prefix ) def __ne__(self, other: object) -> bool: @@ -337,6 +383,7 @@ def __repr__(self) -> str: "channel_high_watermark", "event_queue_watermarks", "stats_dump_interval", + "user_agent_prefix", ) params = [] @@ -399,6 +446,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]``. + user_agent_prefix: Bytestring to include in the user agent for broker + telemetry. This string must only contain printable characters and + must be at most 96 bytes long. This is provided for libraries + that are wrapping this SDK. Applications directly using the SDK + are encouraged *NOT* to set this value. Raises: `~blazingmq.Error`: If the session start request was not successful. @@ -423,6 +475,7 @@ def __init__( channel_high_watermark: Optional[int] = None, event_queue_watermarks: Optional[tuple[int, int]] = None, stats_dump_interval: Optional[float] = None, + user_agent_prefix: Optional[bytes] = None, ) -> None: if host_health_monitor is not None: if not isinstance(host_health_monitor, BasicHealthMonitor): @@ -459,6 +512,7 @@ def __init__( timeouts=_validate_timeouts(timeout), monitor_host_health=monitor_host_health, fake_host_health_monitor=fake_host_health_monitor, + user_agent_prefix=_make_user_agent_prefix(user_agent_prefix), ) self._ext.set_owned_by_session() @@ -516,6 +570,7 @@ def with_options( session_options.channel_high_watermark, session_options.event_queue_watermarks, session_options.stats_dump_interval, + session_options.user_agent_prefix, ) else: return cls( @@ -530,6 +585,7 @@ def with_options( session_options.channel_high_watermark, session_options.event_queue_watermarks, session_options.stats_dump_interval, + session_options.user_agent_prefix, ) def open_queue( diff --git a/src/cpp/pybmq_mocksession.cpp b/src/cpp/pybmq_mocksession.cpp index 6e1050a..55f51bd 100644 --- a/src/cpp/pybmq_mocksession.cpp +++ b/src/cpp/pybmq_mocksession.cpp @@ -295,7 +295,8 @@ MockSession::MockSession( "channel_high_watermark", "event_queue_low_watermark", "event_queue_high_watermark", - "stats_dump_interval"}; + "stats_dump_interval", + "user_agent_prefix"}; double timeout_connect_secs = time_interval_to_seconds(options.connectTimeout()); double timeout_disconnect_secs = @@ -309,7 +310,7 @@ MockSession::MockSession( bslma::ManagedPtr py_options = RefUtils::toManagedPtr(_Py_DictBuilder( option_names, - "(s# N f f f f f i i i i i f)", + "(s# N f f f f f i i i i i f s#)", options.brokerUri().c_str(), options.brokerUri().length(), PyBytes_FromStringAndSize( @@ -325,7 +326,9 @@ MockSession::MockSession( options.channelHighWatermark(), options.eventQueueLowWatermark(), options.eventQueueHighWatermark(), - stats_dump_interval_secs)); + stats_dump_interval_secs, + options.userAgentPrefix().c_str(), + options.userAgentPrefix().length())); if (!py_options) throw bsl::runtime_error("propagating Python error"); PyObject_SetAttrString(d_mock, "options", py_options.get()); } diff --git a/src/cpp/pybmq_session.cpp b/src/cpp/pybmq_session.cpp index acab799..3ec024c 100644 --- a/src/cpp/pybmq_session.cpp +++ b/src/cpp/pybmq_session.cpp @@ -94,7 +94,8 @@ Session::Session( bsl::shared_ptr fake_host_health_monitor_sp, PyObject* error, PyObject* broker_timeout_error, - PyObject* mock) + PyObject* mock, + const char* user_agent_prefix) : d_started_lock() , d_started(false) , d_message_compression_type(bmqt::CompressionAlgorithmType::e_NONE) @@ -168,6 +169,8 @@ Session::Session( options.setCloseQueueTimeout(close_queue_timeout); } + options.setUserAgentPrefix(user_agent_prefix); + bslma::ManagedPtr handler( new pybmq::SessionEventHandler( py_session_event_callback, diff --git a/src/cpp/pybmq_session.h b/src/cpp/pybmq_session.h index f37a407..d35b4c4 100644 --- a/src/cpp/pybmq_session.h +++ b/src/cpp/pybmq_session.h @@ -68,7 +68,8 @@ class Session bsl::shared_ptr fake_host_health_monitor, PyObject* d_error, PyObject* d_broker_timeout_error, - PyObject* mock); + PyObject* mock, + const char* user_agent_prefix); ~Session(); diff --git a/src/declarations/pybmq.pxd b/src/declarations/pybmq.pxd index 9ba293e..bf9a208 100644 --- a/src/declarations/pybmq.pxd +++ b/src/declarations/pybmq.pxd @@ -56,7 +56,8 @@ cdef extern from "pybmq_session.h" namespace "BloombergLP::pybmq" nogil: shared_ptr[ManualHostHealthMonitor] fake_host_health_monitor_sp, object error, object broker_timeout_error, - object mock) except+ + object mock, + const char* user_agent_prefix) except+ object start(TimeInterval) except+ object stop(bint) except+ diff --git a/tests/unit/test_ext_session.py b/tests/unit/test_ext_session.py index 9ad0825..e5679e3 100644 --- a/tests/unit/test_ext_session.py +++ b/tests/unit/test_ext_session.py @@ -238,6 +238,7 @@ def test_session_session_options_propagated(): event_queue_low_watermark = 1000000 event_queue_high_watermark = 10000000 stats_dump_interval = 90.0 + user_agent_prefix = b"mylib:1.0" # WHEN Session( @@ -247,6 +248,7 @@ def test_session_session_options_propagated(): channel_high_watermark=channel_high_watermark, event_queue_watermarks=(event_queue_low_watermark, event_queue_high_watermark), stats_dump_interval=stats_dump_interval, + user_agent_prefix=user_agent_prefix, _mock=mock, ) @@ -257,6 +259,7 @@ def test_session_session_options_propagated(): assert mock.options["event_queue_low_watermark"] == event_queue_low_watermark assert mock.options["event_queue_high_watermark"] == event_queue_high_watermark assert mock.options["stats_dump_interval"] == stats_dump_interval + assert mock.options["user_agent_prefix"] == user_agent_prefix.decode("ascii") def test_ensure_stop_session_callback_calls_sdk_stop(): diff --git a/tests/unit/test_session.py b/tests/unit/test_session.py index 78fba1d..eec352b 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, + user_agent_prefix=mock.ANY, # varies by version; see dedicated tests ) @@ -128,6 +129,7 @@ def dummy2(): timeouts=timeouts, monitor_host_health=False, fake_host_health_monitor=None, + user_agent_prefix=mock.ANY, # varies by version; see dedicated tests ) @@ -172,6 +174,7 @@ def dummy2(): timeouts=timeouts, monitor_host_health=False, fake_host_health_monitor=None, + user_agent_prefix=mock.ANY, # varies by version; see dedicated tests ) @@ -207,6 +210,7 @@ def dummy2(): timeouts=Timeouts(), monitor_host_health=False, fake_host_health_monitor=None, + user_agent_prefix=mock.ANY, # varies by version; see dedicated tests ) @@ -259,6 +263,35 @@ def dummy2(): timeouts=timeouts, monitor_host_health=False, fake_host_health_monitor=None, + user_agent_prefix=mock.ANY, # varies by version; see dedicated tests + ) + + +@mock.patch("blazingmq._session.ExtSession") +def test_session_with_options_user_agent_prefix(ext_cls): + # GIVEN + ext_cls.mock_add_spec([]) + + def dummy1(): + pass + + def dummy2(): + pass + + session_options = SessionOptions( + user_agent_prefix=b"mylib:1.0", + ) + + # WHEN + Session.with_options( + dummy1, on_message=dummy2, broker="some_uri", session_options=session_options + ) + + # THEN + _, kwargs = ext_cls.call_args + assert re.fullmatch( + rb"mylib:1\.0 blazingmq\(python[^)]+\):\S+", + kwargs["user_agent_prefix"], ) @@ -304,6 +337,7 @@ def dummy2(): ), monitor_host_health=True, fake_host_health_monitor=monitor._monitor, + user_agent_prefix=mock.ANY, # varies by version; see dedicated tests ) @@ -335,6 +369,7 @@ def dummy2(): timeouts=Timeouts(), monitor_host_health=False, fake_host_health_monitor=None, + user_agent_prefix=mock.ANY, # varies by version; see dedicated tests ) @@ -772,6 +807,231 @@ def test_basic_monitor_repr(): assert msg == "BasicHealthMonitor()" +@mock.patch("blazingmq._session.ExtSession") +def test_session_constructed_with_user_agent_prefix(ext_cls): + # GIVEN + ext_cls.mock_add_spec([]) + + def dummy(): + pass + + # WHEN + Session( + dummy, + on_message=dummy, + broker="some_uri", + user_agent_prefix=b"mylib:1.0", + ) + + # THEN + _, kwargs = ext_cls.call_args + assert re.fullmatch( + rb"mylib:1\.0 blazingmq\(python[^)]+\):\S+", + kwargs["user_agent_prefix"], + ) + + +@mock.patch("blazingmq._session.ExtSession") +def test_session_constructed_without_user_agent_prefix(ext_cls): + # GIVEN + ext_cls.mock_add_spec([]) + + def dummy(): + pass + + # WHEN + Session( + dummy, + on_message=dummy, + broker="some_uri", + ) + + # THEN + _, kwargs = ext_cls.call_args + assert re.fullmatch( + rb"blazingmq\(python[^)]+\):\S+", + kwargs["user_agent_prefix"], + ) + + +@mock.patch("blazingmq._session.ExtSession") +def test_session_constructed_with_empty_user_agent_prefix(ext_cls): + # GIVEN + ext_cls.mock_add_spec([]) + + def dummy(): + pass + + # WHEN + Session( + dummy, + on_message=dummy, + broker="some_uri", + user_agent_prefix=b"", + ) + + # THEN + _, kwargs = ext_cls.call_args + assert re.fullmatch( + rb"blazingmq\(python[^)]+\):\S+", + kwargs["user_agent_prefix"], + ) + + +@mock.patch("blazingmq._session.ExtSession") +def test_session_user_agent_prefix_satisfies_bmq_length_precondition(ext_cls): + # GIVEN + ext_cls.mock_add_spec([]) + + def dummy(): + pass + + # WHEN + Session( + dummy, + on_message=dummy, + broker="some_uri", + user_agent_prefix=b"x" * 96, + ) + + # THEN + _, kwargs = ext_cls.call_args + assert len(kwargs["user_agent_prefix"]) < 128 + + +@mock.patch("blazingmq._session.ExtSession") +def test_session_user_agent_prefix_satisfies_bmq_printable_precondition(ext_cls): + # GIVEN + ext_cls.mock_add_spec([]) + + def dummy(): + pass + + # WHEN + Session( + dummy, + on_message=dummy, + broker="some_uri", + user_agent_prefix=b"mylib:1.0", + ) + + # THEN + _, kwargs = ext_cls.call_args + assert kwargs["user_agent_prefix"].decode("ascii").isprintable() + + +def test_session_bad_user_agent_prefix_too_long(): + # GIVEN + def dummy(): + pass + + user_agent_prefix = b"x" * 97 + expected_pat = re.escape( + f"user_agent_prefix ({user_agent_prefix!r}) must be at most 96 " + f"bytes (is {len(user_agent_prefix)} bytes)" + ) + + # WHEN + with pytest.raises(Exception) as exc: + Session( + dummy, + on_message=dummy, + broker="some_uri", + user_agent_prefix=user_agent_prefix, + ) + + # THEN + assert exc.type is ValueError + assert exc.match(expected_pat) + + +def test_session_bad_user_agent_prefix_non_printable(): + # GIVEN + def dummy(): + pass + + user_agent_prefix = b"\x07" + + # WHEN + with pytest.raises(Exception) as exc: + Session( + dummy, + on_message=dummy, + broker="some_uri", + user_agent_prefix=user_agent_prefix, + ) + + # THEN + assert exc.type is ValueError + assert exc.match("must only contain printable characters") + + +def test_session_bad_user_agent_prefix_non_ascii(): + # GIVEN + def dummy(): + pass + + user_agent_prefix = b"\x80" + + # WHEN + with pytest.raises(Exception) as exc: + Session( + dummy, + on_message=dummy, + broker="some_uri", + user_agent_prefix=user_agent_prefix, + ) + + # THEN + assert exc.type is ValueError + assert exc.match("must only contain printable characters") + + +def test_session_bad_user_agent_prefix_del(): + # GIVEN + def dummy(): + pass + + user_agent_prefix = b"\x7f" + + # WHEN + with pytest.raises(Exception) as exc: + Session( + dummy, + on_message=dummy, + broker="some_uri", + user_agent_prefix=user_agent_prefix, + ) + + # THEN + assert exc.type is ValueError + assert exc.match("must only contain printable characters") + + +@mock.patch("blazingmq._session.ExtSession") +def test_session_user_agent_prefix_with_spaces(ext_cls): + # GIVEN + ext_cls.mock_add_spec([]) + + def dummy(): + pass + + # WHEN + Session( + dummy, + on_message=dummy, + broker="some_uri", + user_agent_prefix=b"my lib:1.0", + ) + + # THEN + _, kwargs = ext_cls.call_args + assert re.fullmatch( + rb"my lib:1\.0 blazingmq\(python[^)]+\):\S+", + kwargs["user_agent_prefix"], + ) + + def test_host_health_repr(): # GIVEN # WHEN diff --git a/tests/unit/test_session_options.py b/tests/unit/test_session_options.py index 6adc505..ce14459 100644 --- a/tests/unit/test_session_options.py +++ b/tests/unit/test_session_options.py @@ -29,6 +29,7 @@ def test_session_options_repr(): channel_high_watermark=8000000, event_queue_watermarks=(6000000, 7000000), stats_dump_interval=30.0, + user_agent_prefix=b"mylib:1.0", ) # THEN assert ( @@ -40,7 +41,8 @@ def test_session_options_repr(): " blob_buffer_size=5000," " channel_high_watermark=8000000," " event_queue_watermarks=(6000000, 7000000)," - " stats_dump_interval=30.0)" == repr(one) + " stats_dump_interval=30.0," + " user_agent_prefix=b'mylib:1.0')" == repr(one) ) @@ -63,6 +65,7 @@ def test_session_options_default_to_none(): assert options.channel_high_watermark is None assert options.event_queue_watermarks is None assert options.stats_dump_interval is None + assert options.user_agent_prefix is None def test_session_options_equality(): @@ -92,6 +95,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(user_agent_prefix=b"mylib:1.0"), ], ) def test_queue_options_other_inequality(right):