Skip to content

Commit 85fd80c

Browse files
pniedzielskipniedzielski
committed
Refactor: Bundle pybmq::Session options into SessionConfig
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 <pniedzielski@bloomberg.net>
1 parent 9c6ed2d commit 85fd80c

5 files changed

Lines changed: 160 additions & 83 deletions

File tree

‎src/blazingmq/_ext.pyx‎

Lines changed: 20 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -37,6 +37,7 @@ from bmq.bmqt cimport k_DEFAULT_MAX_UNCONFIRMED_MESSAGES
3737
from bmq.bmqt cimport k_DEFAULT_SUSPENDS_ON_BAD_HOST_HEALTH
3838
from pybmq cimport BallUtil
3939
from pybmq cimport Session as NativeSession
40+
from pybmq cimport SessionConfig
4041

4142
from typing import Optional
4243

@@ -226,24 +227,29 @@ cdef class Session:
226227
script_name = _script_name.get_script_name()
227228
cdef char *c_script_name = script_name
228229
cdef char *c_user_agent_prefix = user_agent_prefix
230+
231+
cdef SessionConfig config
232+
config.broker_uri = c_broker_uri
233+
config.script_name = c_script_name
234+
config.message_compression_type = (
235+
COMPRESSION_ALGO_FROM_PY_MAPPING[message_compression_algorithm])
236+
config.num_processing_threads = c_num_processing_threads
237+
config.blob_buffer_size = c_blob_buffer_size
238+
config.channel_high_watermark = c_channel_high_watermark
239+
config.event_queue_watermarks = c_event_queue_watermarks
240+
config.stats_dump_interval = c_stats_dump_interval
241+
config.connect_timeout = c_connect_timeout
242+
config.disconnect_timeout = c_disconnect_timeout
243+
config.open_queue_timeout = c_open_queue_timeout
244+
config.configure_queue_timeout = c_configure_queue_timeout
245+
config.close_queue_timeout = c_close_queue_timeout
246+
config.monitor_host_health = monitor_host_health
247+
229248
self._session = new NativeSession(
230249
session_cb,
231250
message_cb,
232251
ack_cb,
233-
c_broker_uri,
234-
c_script_name,
235-
COMPRESSION_ALGO_FROM_PY_MAPPING[message_compression_algorithm],
236-
c_num_processing_threads,
237-
c_blob_buffer_size,
238-
c_channel_high_watermark,
239-
c_event_queue_watermarks,
240-
c_stats_dump_interval,
241-
c_connect_timeout,
242-
c_disconnect_timeout,
243-
c_open_queue_timeout,
244-
c_configure_queue_timeout,
245-
c_close_queue_timeout,
246-
monitor_host_health,
252+
config,
247253
fake_host_health_monitor_sp,
248254
Error,
249255
BrokerTimeoutError,

‎src/cpp/pybmq_session.cpp‎

Lines changed: 28 additions & 41 deletions
Original file line numberDiff line numberDiff line change
@@ -77,20 +77,7 @@ Session::Session(
7777
PyObject* py_session_event_callback,
7878
PyObject* py_message_event_callback,
7979
PyObject* py_ack_event_callback,
80-
const char* broker_uri,
81-
const char* script_name,
82-
bmqt::CompressionAlgorithmType::Enum message_compression_type,
83-
bsl::optional<int> num_processing_threads,
84-
bsl::optional<int> blob_buffer_size,
85-
bsl::optional<int> channel_high_watermark,
86-
bsl::optional<bsl::pair<int, int> > event_queue_watermarks,
87-
const bsls::TimeInterval& stats_dump_interval,
88-
const bsls::TimeInterval& connect_timeout,
89-
const bsls::TimeInterval& disconnect_timeout,
90-
const bsls::TimeInterval& open_queue_timeout,
91-
const bsls::TimeInterval& configure_queue_timeout,
92-
const bsls::TimeInterval& close_queue_timeout,
93-
bool monitor_host_health,
80+
const SessionConfig& config,
9481
bsl::shared_ptr<bmqa::ManualHostHealthMonitor> fake_host_health_monitor_sp,
9582
PyObject* error,
9683
PyObject* broker_timeout_error,
@@ -107,66 +94,66 @@ Session::Session(
10794

10895
if (fake_host_health_monitor_sp) {
10996
host_health_monitor_sp = fake_host_health_monitor_sp;
110-
} else if (monitor_host_health) {
97+
} else if (config.monitor_host_health) {
11198
}
11299

113-
if (message_compression_type
100+
if (config.message_compression_type
114101
< bmqt::CompressionAlgorithmType::k_LOWEST_SUPPORTED_TYPE
115-
|| message_compression_type
102+
|| config.message_compression_type
116103
> bmqt::CompressionAlgorithmType::k_HIGHEST_SUPPORTED_TYPE)
117104
{
118105
PyErr_SetString(PyExc_ValueError, "Invalid message compression type");
119106
throw bsl::runtime_error("propagating Python error");
120107
}
121108

122-
d_message_compression_type = message_compression_type;
109+
d_message_compression_type = config.message_compression_type;
123110
{
124111
pybmq::GilReleaseGuard guard;
125112
bmqt::SessionOptions options;
126-
options.setBrokerUri(broker_uri)
127-
.setProcessNameOverride(script_name)
113+
options.setBrokerUri(config.broker_uri)
114+
.setProcessNameOverride(config.script_name)
128115
.setHostHealthMonitor(host_health_monitor_sp);
129116

130-
if (num_processing_threads.has_value()) {
131-
options.setNumProcessingThreads(num_processing_threads.value());
117+
if (config.num_processing_threads.has_value()) {
118+
options.setNumProcessingThreads(config.num_processing_threads.value());
132119
}
133120

134-
if (blob_buffer_size.has_value()) {
135-
options.setBlobBufferSize(blob_buffer_size.value());
121+
if (config.blob_buffer_size.has_value()) {
122+
options.setBlobBufferSize(config.blob_buffer_size.value());
136123
}
137124

138-
if (channel_high_watermark.has_value()) {
139-
options.setChannelHighWatermark(channel_high_watermark.value());
125+
if (config.channel_high_watermark.has_value()) {
126+
options.setChannelHighWatermark(config.channel_high_watermark.value());
140127
}
141128

142-
if (event_queue_watermarks.has_value()) {
129+
if (config.event_queue_watermarks.has_value()) {
143130
options.configureEventQueue(
144-
event_queue_watermarks.value().first,
145-
event_queue_watermarks.value().second);
131+
config.event_queue_watermarks.value().first,
132+
config.event_queue_watermarks.value().second);
146133
}
147134

148-
if (stats_dump_interval != bsls::TimeInterval()) {
149-
options.setStatsDumpInterval(stats_dump_interval);
135+
if (config.stats_dump_interval != bsls::TimeInterval()) {
136+
options.setStatsDumpInterval(config.stats_dump_interval);
150137
}
151138

152-
if (connect_timeout != bsls::TimeInterval()) {
153-
options.setConnectTimeout(connect_timeout);
139+
if (config.connect_timeout != bsls::TimeInterval()) {
140+
options.setConnectTimeout(config.connect_timeout);
154141
}
155142

156-
if (disconnect_timeout != bsls::TimeInterval()) {
157-
options.setDisconnectTimeout(disconnect_timeout);
143+
if (config.disconnect_timeout != bsls::TimeInterval()) {
144+
options.setDisconnectTimeout(config.disconnect_timeout);
158145
}
159146

160-
if (open_queue_timeout != bsls::TimeInterval()) {
161-
options.setOpenQueueTimeout(open_queue_timeout);
147+
if (config.open_queue_timeout != bsls::TimeInterval()) {
148+
options.setOpenQueueTimeout(config.open_queue_timeout);
162149
}
163150

164-
if (configure_queue_timeout != bsls::TimeInterval()) {
165-
options.setConfigureQueueTimeout(configure_queue_timeout);
151+
if (config.configure_queue_timeout != bsls::TimeInterval()) {
152+
options.setConfigureQueueTimeout(config.configure_queue_timeout);
166153
}
167154

168-
if (close_queue_timeout != bsls::TimeInterval()) {
169-
options.setCloseQueueTimeout(close_queue_timeout);
155+
if (config.close_queue_timeout != bsls::TimeInterval()) {
156+
options.setCloseQueueTimeout(config.close_queue_timeout);
170157
}
171158

172159
options.setUserAgentPrefix(user_agent_prefix);

‎src/cpp/pybmq_session.h‎

Lines changed: 3 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,8 @@
1919
#define PY_SSIZE_T_CLEAN
2020
#include <Python.h>
2121

22+
#include <pybmq_sessionconfig.h>
23+
2224
#include <bmqa_abstractsession.h>
2325
#include <bmqa_manualhosthealthmonitor.h>
2426
#include <bmqt_compressionalgorithmtype.h>
@@ -51,20 +53,7 @@ class Session
5153
Session(PyObject* py_session_event_callback,
5254
PyObject* py_message_event_callback,
5355
PyObject* py_ack_event_callback,
54-
const char* broker_uri,
55-
const char* script_name,
56-
bmqt::CompressionAlgorithmType::Enum message_compression_type,
57-
bsl::optional<int> num_processing_threads,
58-
bsl::optional<int> blob_buffer_size,
59-
bsl::optional<int> channel_high_watermark,
60-
bsl::optional<bsl::pair<int, int> > event_queue_watermarks,
61-
const bsls::TimeInterval& stats_dump_interval,
62-
const bsls::TimeInterval& connect_timeout,
63-
const bsls::TimeInterval& disconnect_timeout,
64-
const bsls::TimeInterval& open_queue_timeout,
65-
const bsls::TimeInterval& configure_queue_timeout,
66-
const bsls::TimeInterval& close_queue_timeout,
67-
bool monitor_host_health,
56+
const SessionConfig& config,
6857
bsl::shared_ptr<bmqa::ManualHostHealthMonitor> fake_host_health_monitor,
6958
PyObject* d_error,
7059
PyObject* d_broker_timeout_error,

‎src/cpp/pybmq_sessionconfig.h‎

Lines changed: 89 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,89 @@
1+
// Copyright 2019-2026 Bloomberg Finance L.P.
2+
// SPDX-License-Identifier: Apache-2.0
3+
//
4+
// Licensed under the Apache License, Version 2.0 (the "License");
5+
// you may not use this file except in compliance with the License.
6+
// You may obtain a copy of the License at
7+
//
8+
// http://www.apache.org/licenses/LICENSE-2.0
9+
//
10+
// Unless required by applicable law or agreed to in writing, software
11+
// distributed under the License is distributed on an "AS IS" BASIS,
12+
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
// See the License for the specific language governing permissions and
14+
// limitations under the License.
15+
16+
#ifndef INCLUDED_PYBMQ_SESSIONCONFIG
17+
#define INCLUDED_PYBMQ_SESSIONCONFIG
18+
19+
#include <bmqt_compressionalgorithmtype.h>
20+
21+
#include <bsl_optional.h>
22+
#include <bsl_utility.h>
23+
#include <bsls_timeinterval.h>
24+
25+
namespace BloombergLP {
26+
namespace pybmq {
27+
28+
// ===================
29+
// struct SessionConfig
30+
// ===================
31+
32+
/// The options a `pybmq::Session` is constructed with, other than its Python
33+
/// callbacks and host health monitor.
34+
///
35+
/// A default-constructed `SessionConfig` means "no option set": empty
36+
/// optionals, and default-constructed `bsls::TimeInterval`s, which
37+
/// `Session` treats as unset. A new option can therefore be added as a
38+
/// field without changing any existing call site.
39+
///
40+
/// `broker_uri` and `script_name` are held, not owned, and must outlive the
41+
/// `Session` constructor call.
42+
struct SessionConfig
43+
{
44+
// PUBLIC DATA
45+
const char* broker_uri;
46+
const char* script_name;
47+
bmqt::CompressionAlgorithmType::Enum message_compression_type;
48+
bsl::optional<int> num_processing_threads;
49+
bsl::optional<int> blob_buffer_size;
50+
bsl::optional<int> channel_high_watermark;
51+
bsl::optional<bsl::pair<int, int> > event_queue_watermarks;
52+
bsls::TimeInterval stats_dump_interval;
53+
bsls::TimeInterval connect_timeout;
54+
bsls::TimeInterval disconnect_timeout;
55+
bsls::TimeInterval open_queue_timeout;
56+
bsls::TimeInterval configure_queue_timeout;
57+
bsls::TimeInterval close_queue_timeout;
58+
bool monitor_host_health;
59+
60+
// CREATORS
61+
SessionConfig();
62+
};
63+
64+
// ===========================================================================
65+
// INLINE DEFINITIONS
66+
// ===========================================================================
67+
68+
inline SessionConfig::SessionConfig()
69+
: broker_uri(0)
70+
, script_name(0)
71+
, message_compression_type(bmqt::CompressionAlgorithmType::e_NONE)
72+
, num_processing_threads()
73+
, blob_buffer_size()
74+
, channel_high_watermark()
75+
, event_queue_watermarks()
76+
, stats_dump_interval()
77+
, connect_timeout()
78+
, disconnect_timeout()
79+
, open_queue_timeout()
80+
, configure_queue_timeout()
81+
, close_queue_timeout()
82+
, monitor_host_health(false)
83+
{
84+
}
85+
86+
} // namespace pybmq
87+
} // namespace BloombergLP
88+
89+
#endif

‎src/declarations/pybmq.pxd‎

Lines changed: 20 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -34,25 +34,31 @@ cdef extern from 'pybmq_ballutil.h' namespace 'BloombergLP::pybmq':
3434
@staticmethod
3535
object shutDownBallSingleton() except +
3636

37+
cdef extern from "pybmq_sessionconfig.h" namespace "BloombergLP::pybmq" nogil:
38+
cdef cppclass SessionConfig:
39+
SessionConfig() except+
40+
41+
const char* broker_uri
42+
const char* script_name
43+
CompressionAlgorithmType message_compression_type
44+
optional[int] num_processing_threads
45+
optional[int] blob_buffer_size
46+
optional[int] channel_high_watermark
47+
optional[pair[int, int]] event_queue_watermarks
48+
TimeInterval stats_dump_interval
49+
TimeInterval connect_timeout
50+
TimeInterval disconnect_timeout
51+
TimeInterval open_queue_timeout
52+
TimeInterval configure_queue_timeout
53+
TimeInterval close_queue_timeout
54+
bint monitor_host_health
55+
3756
cdef extern from "pybmq_session.h" namespace "BloombergLP::pybmq" nogil:
3857
cdef cppclass Session:
3958
Session(object on_session_event,
4059
object on_message_event,
4160
object on_ack_event,
42-
const char* broker_uri,
43-
const char* script_name,
44-
CompressionAlgorithmType message_compression_algorithm,
45-
optional[int] num_processing_threads,
46-
optional[int] blob_buffer_size,
47-
optional[int] channel_high_watermark,
48-
optional[pair[int, int]] event_queue_watermarks,
49-
TimeInterval stats_dump_interval,
50-
TimeInterval connect_timeout,
51-
TimeInterval disconnect_timeout,
52-
TimeInterval open_queue_timeout,
53-
TimeInterval configure_queue_timeout,
54-
TimeInterval close_queue_timeout,
55-
bint monitor_host_health,
61+
const SessionConfig& config,
5662
shared_ptr[ManualHostHealthMonitor] fake_host_health_monitor_sp,
5763
object error,
5864
object broker_timeout_error,

0 commit comments

Comments
 (0)