This is an automated email from the ASF dual-hosted git repository.
mmodzelewski pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new 3c1a8ccf6 feat(python): add WebSocketConfig transport configuration
(#4000)
3c1a8ccf6 is described below
commit 3c1a8ccf65eb67a57f4e92ffaeb92dd2df506765
Author: saie-ch <[email protected]>
AuthorDate: Tue Sep 8 00:56:38 2026 +0530
feat(python): add WebSocketConfig transport configuration (#4000)
---
.../actions/python-maturin/pre-merge/action.yml | 5 +-
.github/workflows/coverage-baseline.yml | 5 +-
.../websocket_config/websocket_client_config.rs | 75 ++-
core/sdk/src/prelude.rs | 4 +-
examples/python/getting-started/consumer.py | 17 +-
examples/python/getting-started/producer.py | 17 +-
foreign/python/README.md | 11 +-
foreign/python/apache_iggy.pyi | 209 +++++++-
foreign/python/docker-compose.test.yml | 2 +
foreign/python/src/client.rs | 21 +-
foreign/python/src/config.rs | 529 ++++++++++++++++++++-
foreign/python/src/lib.rs | 4 +
foreign/python/tests/test_websocket_config.py | 522 ++++++++++++++++++++
foreign/python/tests/utils.py | 11 +
14 files changed, 1388 insertions(+), 44 deletions(-)
diff --git a/.github/actions/python-maturin/pre-merge/action.yml
b/.github/actions/python-maturin/pre-merge/action.yml
index 53bf430dc..d8ab37ad6 100644
--- a/.github/actions/python-maturin/pre-merge/action.yml
+++ b/.github/actions/python-maturin/pre-merge/action.yml
@@ -175,8 +175,8 @@ runs:
# TCP and HTTP ports come from server-start's own outputs rather than
# duplicated literals, so a future default change here can't silently
- # desync from what the server actually started with. QUIC stays a
- # literal below because server-start exposes no output for it.
+ # desync from what the server actually started with. QUIC and WebSocket
+ # stay literals below because server-start exposes no output for them.
tcp_port="${{ steps.iggy.outputs.tcp_address }}"
http_port="${{ steps.iggy.outputs.http_address }}"
tcp_port="${tcp_port##*:}"
@@ -191,6 +191,7 @@ runs:
IGGY_SERVER_TCP_PORT="$tcp_port" \
IGGY_SERVER_QUIC_PORT=8080 \
IGGY_SERVER_HTTP_PORT="$http_port" \
+ IGGY_SERVER_WS_PORT=8092 \
IGGY_SERVER_DOCKER_IMAGE=iggy-server:local \
uv run --no-sync pytest tests/ -v \
--junitxml=../../reports/python-junit.xml \
diff --git a/.github/workflows/coverage-baseline.yml
b/.github/workflows/coverage-baseline.yml
index 76d12d1b0..b4bd7c315 100644
--- a/.github/workflows/coverage-baseline.yml
+++ b/.github/workflows/coverage-baseline.yml
@@ -350,8 +350,8 @@ jobs:
# TCP and HTTP ports come from server-start's own outputs rather than
# duplicated literals, so a future default change here can't silently
- # desync from what the server actually started with. QUIC stays a
- # literal below because server-start exposes no output for it.
+ # desync from what the server actually started with. QUIC and
WebSocket
+ # stay literals below because server-start exposes no output for
them.
tcp_port="${{ steps.iggy.outputs.tcp_address }}"
http_port="${{ steps.iggy.outputs.http_address }}"
tcp_port="${tcp_port##*:}"
@@ -361,6 +361,7 @@ jobs:
IGGY_SERVER_TCP_PORT="$tcp_port" \
IGGY_SERVER_QUIC_PORT=8080 \
IGGY_SERVER_HTTP_PORT="$http_port" \
+ IGGY_SERVER_WS_PORT=8092 \
IGGY_SERVER_DOCKER_IMAGE=iggy-server:local \
uv run --no-sync pytest tests/ -v \
--junitxml=../../reports/python-junit.xml \
diff --git
a/core/common/src/types/configuration/websocket_config/websocket_client_config.rs
b/core/common/src/types/configuration/websocket_config/websocket_client_config.rs
index 4295ab527..6fdd79179 100644
---
a/core/common/src/types/configuration/websocket_config/websocket_client_config.rs
+++
b/core/common/src/types/configuration/websocket_config/websocket_client_config.rs
@@ -55,9 +55,13 @@ pub struct WebSocketConfig {
pub write_buffer_size: Option<usize>,
/// Maximum write buffer size.
pub max_write_buffer_size: Option<usize>,
- /// Maximum message size.
+ /// Maximum message size, or `None` for no limit. Defaults to tungstenite's
+ /// own limit rather than `None`, so `None` always means the limit was
lifted
+ /// deliberately.
pub max_message_size: Option<usize>,
- /// Maximum frame size.
+ /// Maximum frame size, or `None` for no limit. Defaults to tungstenite's
own
+ /// limit rather than `None`, so `None` always means the limit was lifted
+ /// deliberately.
pub max_frame_size: Option<usize>,
/// Accept unmasked frames (client should typically keep as false).
pub accept_unmasked_frames: bool,
@@ -111,13 +115,13 @@ impl WebSocketConfig {
config = config.max_write_buffer_size(max_write_buf_size);
}
- if let Some(max_msg_size) = self.max_message_size {
- config = config.max_message_size(Some(max_msg_size));
- }
-
- if let Some(max_frame_size) = self.max_frame_size {
- config = config.max_frame_size(Some(max_frame_size));
- }
+ // Set unconditionally, unlike the buffer sizes above: `None` here
means
+ // "no limit", not "unset". `Default` seeds both from tungstenite's own
+ // values, so a `None` can only come from a caller that asked for the
+ // limit to be lifted, and skipping the setter would silently leave
+ // tungstenite's default in force instead.
+ config = config.max_message_size(self.max_message_size);
+ config = config.max_frame_size(self.max_frame_size);
config = config.accept_unmasked_frames(self.accept_unmasked_frames);
@@ -189,3 +193,56 @@ impl Display for WebSocketConfig {
)
}
}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn default_should_carry_tungstenites_own_size_limits() {
+ let config = WebSocketConfig::default();
+ let tungstenite = TungsteniteConfig::default();
+
+ assert_eq!(config.max_message_size, tungstenite.max_message_size);
+ assert_eq!(config.max_frame_size, tungstenite.max_frame_size);
+ assert!(config.max_message_size.is_some());
+ assert!(config.max_frame_size.is_some());
+ }
+
+ #[test]
+ fn none_size_limits_should_reach_tungstenite_as_no_limit() {
+ let config = WebSocketConfig {
+ max_message_size: None,
+ max_frame_size: None,
+ ..Default::default()
+ };
+
+ let tungstenite = config.to_tungstenite_config();
+
+ assert_eq!(tungstenite.max_message_size, None);
+ assert_eq!(tungstenite.max_frame_size, None);
+ }
+
+ #[test]
+ fn default_size_limits_should_reach_tungstenite_unchanged() {
+ let tungstenite = WebSocketConfig::default().to_tungstenite_config();
+ let expected = TungsteniteConfig::default();
+
+ assert_eq!(tungstenite.max_message_size, expected.max_message_size);
+ assert_eq!(tungstenite.max_frame_size, expected.max_frame_size);
+ }
+
+ #[test]
+ fn explicit_size_limits_should_reach_tungstenite_unchanged() {
+ let config = WebSocketConfig {
+ max_message_size: Some(1024),
+ max_frame_size: Some(512),
+ ..Default::default()
+ };
+
+ let tungstenite = config.to_tungstenite_config();
+
+ assert_eq!(tungstenite.max_message_size, Some(1024));
+ assert_eq!(tungstenite.max_frame_size, Some(512));
+ }
+}
diff --git a/core/sdk/src/prelude.rs b/core/sdk/src/prelude.rs
index 79c6774b2..f239eec67 100644
--- a/core/sdk/src/prelude.rs
+++ b/core/sdk/src/prelude.rs
@@ -67,8 +67,8 @@ pub use iggy_common::{
TcpClientConfigBuilder, TcpClientReconnectionConfig, Topic,
TopicCreateOptions, TopicDetails,
TopicPermissions, TopicUpdateOptions, TransportEndpoints,
TransportProtocol, UserId, UserInfo,
UserInfoDetails, UserStatus, UserUpdateOptions, Validatable,
WebSocketClientConfig,
- WebSocketClientConfigBuilder, WebSocketClientReconnectionConfig, defaults,
locking,
- topic_option_keys,
+ WebSocketClientConfigBuilder, WebSocketClientReconnectionConfig,
WebSocketConfig, defaults,
+ locking, topic_option_keys,
};
pub use iggy_common::{
Client, ClusterClient, ConsumerGroupClient, ConsumerOffsetClient,
MessageClient,
diff --git a/examples/python/getting-started/consumer.py
b/examples/python/getting-started/consumer.py
index 3a48f2a2e..16405d9d4 100755
--- a/examples/python/getting-started/consumer.py
+++ b/examples/python/getting-started/consumer.py
@@ -31,6 +31,7 @@ from apache_iggy import (
ReceiveMessage,
TcpConfig,
TcpReconnectionConfig,
+ WebSocketConfig,
)
from loguru import logger
@@ -103,7 +104,9 @@ def parse_args() -> ArgNamespace:
return ArgNamespace(**vars(args))
-def build_config(args: ArgNamespace) -> TcpConfig | QuicConfig | HttpConfig:
+def build_config(
+ args: ArgNamespace,
+) -> TcpConfig | QuicConfig | HttpConfig | WebSocketConfig:
"""Build the client configuration, TCP with auto-login and reconnection."""
# IggyClient(...) also accepts a QuicConfig for the QUIC transport. To use
@@ -125,6 +128,18 @@ def build_config(args: ArgNamespace) -> TcpConfig |
QuicConfig | HttpConfig:
# after connecting:
# return HttpConfig(api_url="http://127.0.0.1:3000";)
+ # IggyClient(...) also accepts a WebSocketConfig for the WebSocket
transport.
+ # To use it, uncomment the return below and import
WebSocketReconnectionConfig,
+ # which is left out above because only the commented block names it:
+ #
+ # return WebSocketConfig(
+ # server_address="127.0.0.1:8092",
+ # auto_login=AutoLogin.username_password(args.username, args.password),
+ # reconnection=WebSocketReconnectionConfig(
+ # enabled=True, interval=timedelta(seconds=1)
+ # ),
+ # )
+
return TcpConfig(
server_address=args.tcp_server_address,
auto_login=AutoLogin.username_password(args.username, args.password),
diff --git a/examples/python/getting-started/producer.py
b/examples/python/getting-started/producer.py
index e8a8fea35..2cde76e57 100755
--- a/examples/python/getting-started/producer.py
+++ b/examples/python/getting-started/producer.py
@@ -30,6 +30,7 @@ from apache_iggy import (
TcpConfig,
TcpReconnectionConfig,
TopicDetails,
+ WebSocketConfig,
)
from apache_iggy import SendMessage as Message
from loguru import logger
@@ -102,7 +103,9 @@ def parse_args() -> ArgNamespace:
return ArgNamespace(**vars(args))
-def build_config(args: ArgNamespace) -> TcpConfig | QuicConfig | HttpConfig:
+def build_config(
+ args: ArgNamespace,
+) -> TcpConfig | QuicConfig | HttpConfig | WebSocketConfig:
"""Build the client configuration, TCP with auto-login and reconnection."""
# IggyClient(...) also accepts a QuicConfig for the QUIC transport. To use
@@ -124,6 +127,18 @@ def build_config(args: ArgNamespace) -> TcpConfig |
QuicConfig | HttpConfig:
# after connecting:
# return HttpConfig(api_url="http://127.0.0.1:3000";)
+ # IggyClient(...) also accepts a WebSocketConfig for the WebSocket
transport.
+ # To use it, uncomment the return below and import
WebSocketReconnectionConfig,
+ # which is left out above because only the commented block names it:
+ #
+ # return WebSocketConfig(
+ # server_address="127.0.0.1:8092",
+ # auto_login=AutoLogin.username_password(args.username, args.password),
+ # reconnection=WebSocketReconnectionConfig(
+ # enabled=True, interval=timedelta(seconds=1)
+ # ),
+ # )
+
return TcpConfig(
server_address=args.tcp_server_address,
auto_login=AutoLogin.username_password(args.username, args.password),
diff --git a/foreign/python/README.md b/foreign/python/README.md
index 293bbdfbc..b4a33a116 100644
--- a/foreign/python/README.md
+++ b/foreign/python/README.md
@@ -134,8 +134,8 @@ running prek / committing / pushing. This list is not
exhaustive and other hook
## Client Configuration
-`IggyClient` takes a server address, a `TcpConfig`, a `QuicConfig`, or an
-`HttpConfig`:
+`IggyClient` takes a server address, a `TcpConfig`, a `QuicConfig`, an
+`HttpConfig`, or a `WebSocketConfig`:
```python
import asyncio
@@ -169,9 +169,10 @@ async def main():
asyncio.run(main())
```
-`IggyClient(...)` also accepts a `QuicConfig` for the QUIC transport and an
-`HttpConfig` for the HTTP transport;
-`examples/python/getting-started/producer.py` shows either swap in context.
+`IggyClient(...)` also accepts a `QuicConfig` for the QUIC transport, an
+`HttpConfig` for the HTTP transport, and a `WebSocketConfig` for the WebSocket
+transport. `examples/python/getting-started/producer.py` shows each swap in
+context.
`HttpConfig` differs from TCP in two ways. There is no reconnection policy and
no
`AutoLogin`: `connect()` does not dial over HTTP, but it does start the
diff --git a/foreign/python/apache_iggy.pyi b/foreign/python/apache_iggy.pyi
index ffed2857d..901af64aa 100644
--- a/foreign/python/apache_iggy.pyi
+++ b/foreign/python/apache_iggy.pyi
@@ -68,6 +68,9 @@ __all__ = [
"UserInfo",
"UserInfoDetails",
"UserStatus",
+ "WebSocketConfig",
+ "WebSocketFramingConfig",
+ "WebSocketReconnectionConfig",
]
class AutoCommit:
@@ -975,21 +978,27 @@ class IggyClient:
It provides asynchronous functionality through the contained runtime.
"""
def __new__(
- cls, conn: TcpConfig | QuicConfig | HttpConfig | builtins.str | None =
None
+ cls,
+ conn: TcpConfig
+ | QuicConfig
+ | HttpConfig
+ | WebSocketConfig
+ | builtins.str
+ | None = None,
) -> IggyClient:
r"""
Constructs a new IggyClient from a TCP server address, a `TcpConfig`, a
- `QuicConfig`, or an `HttpConfig`. This initializes a new runtime for
- asynchronous operations.
+ `QuicConfig`, an `HttpConfig`, or a `WebSocketConfig`. This
initializes a
+ new runtime for asynchronous operations.
Future versions might utilize asyncio for more Pythonic async.
Args:
- conn: A `host:port` address, a `TcpConfig`, a `QuicConfig`, or an
- `HttpConfig`. Defaults to `127.0.0.1:8090` over TCP with
auto-login
- disabled. A malformed address is reported differently
depending on
- the form: the string form raises `RuntimeError` here, while
- `TcpConfig`/`QuicConfig`/`HttpConfig` raise `ValueError` when
they
- are constructed, before any of them ever reaches this call.
Neither
+ conn: A `host:port` address, a `TcpConfig`, a `QuicConfig`, an
+ `HttpConfig`, or a `WebSocketConfig`. Defaults to
`127.0.0.1:8090`
+ over TCP with auto-login disabled. A malformed address is
reported
+ differently depending on the form: the string form raises
+ `RuntimeError` here, while every config type raises
`ValueError`
+ when it is constructed, before any of them reaches this call.
Neither
exception is a subclass of the other.
Raises:
@@ -2094,7 +2103,7 @@ class QuicConfig:
seconds) instead, since `configure()` skips the setter
entirely when
zero. Defaults to 10 seconds.
validate_certificate: Whether to validate the server certificate.
Defaults
- to disabled, unlike the TCP and WebSocket transports.
+ to disabled; only the TCP transport validates by default.
Raises:
ValueError: If `server_address` or `client_address` is not a valid
@@ -3002,6 +3011,186 @@ class UserInfoDetails:
The permissions of the user, or `None` when the user has none assigned.
"""
[email protected]
+class WebSocketConfig:
+ r"""
+ Configuration for the WebSocket transport, accepted by `IggyClient(...)`.
+
+ Every field is keyword-only and optional.
+ """
+ @property
+ def server_address(self) -> builtins.str: ...
+ @property
+ def auto_login(self) -> AutoLogin: ...
+ @property
+ def reconnection(self) -> WebSocketReconnectionConfig: ...
+ @property
+ def heartbeat_interval(self) -> datetime.timedelta: ...
+ @property
+ def framing(self) -> WebSocketFramingConfig: ...
+ @property
+ def tls_enabled(self) -> builtins.bool: ...
+ @property
+ def tls_domain(self) -> builtins.str: ...
+ @property
+ def tls_ca_file(self) -> builtins.str | None: ...
+ @property
+ def tls_validate_certificate(self) -> builtins.bool: ...
+ def __new__(
+ cls,
+ *,
+ server_address: builtins.str | None = None,
+ auto_login: AutoLogin | None = None,
+ reconnection: WebSocketReconnectionConfig | None = None,
+ heartbeat_interval: datetime.timedelta | None = None,
+ framing: WebSocketFramingConfig | None = None,
+ tls_enabled: builtins.bool | None = None,
+ tls_domain: builtins.str | None = None,
+ tls_ca_file: builtins.str | None = None,
+ tls_validate_certificate: builtins.bool | None = None,
+ ) -> WebSocketConfig:
+ r"""
+ Constructs a WebSocket configuration.
+
+ Args:
+ server_address: `host:port` of the Iggy server. Defaults to
`127.0.0.1:8092`.
+ auto_login: Credentials replayed on every connect. Defaults to
`AutoLogin.disabled()`.
+ reconnection: Reconnection policy. Defaults to
`WebSocketReconnectionConfig()`.
+ heartbeat_interval: Interval of heartbeats sent by the client.
Defaults to 5 seconds.
+ framing: Frame- and buffer-level options. Defaults to
`WebSocketFramingConfig()`.
+ tls_enabled: Whether to connect over TLS. Defaults to disabled.
+ tls_domain: Domain to validate the certificate against. Defaults to
+ `localhost`. Empty means it is taken from the IP
`server_address`
+ resolves to.
+ tls_ca_file: Path to the CA file for TLS. Read only when
`tls_enabled`
+ and `tls_validate_certificate` are both on; with either one
off it
+ is kept but never consulted, so pairing it with
+ `tls_validate_certificate=False` pins nothing.
+ tls_validate_certificate: Whether to validate the server
certificate.
+ Defaults to `False`; only the TCP transport validates by
default.
+ Disabling this accepts any certificate the server presents,
+ including self-signed and mismatched ones, and takes precedence
+ over `tls_ca_file`.
+
+ Raises:
+ ValueError: If `server_address` is not a valid `host:port` pair,
if a
+ duration is negative, or if `heartbeat_interval` is zero.
+ """
+ def __repr__(self) -> builtins.str: ...
+
[email protected]
+class WebSocketFramingConfig:
+ r"""
+ Frame- and buffer-level options passed through to the underlying WebSocket
+ implementation, accepted by `WebSocketConfig`'s `framing` argument.
+
+ Every field is keyword-only and optional; unset fields fall back to the
+ underlying WebSocket library's own defaults.
+ """
+ @property
+ def read_buffer_size(self) -> builtins.int | None: ...
+ @property
+ def write_buffer_size(self) -> builtins.int | None: ...
+ @property
+ def max_write_buffer_size(self) -> builtins.int | None: ...
+ @property
+ def max_message_size(self) -> builtins.int | None: ...
+ @property
+ def max_frame_size(self) -> builtins.int | None: ...
+ @property
+ def accept_unmasked_frames(self) -> builtins.bool: ...
+ def __new__(
+ cls,
+ *,
+ read_buffer_size: builtins.int | None = None,
+ write_buffer_size: builtins.int | None = None,
+ max_write_buffer_size: builtins.int | None = None,
+ max_message_size: builtins.int | None = 64 << 20,
+ max_frame_size: builtins.int | None = 16 << 20,
+ accept_unmasked_frames: builtins.bool | None = None,
+ ) -> WebSocketFramingConfig:
+ r"""
+ Constructs a WebSocket framing configuration.
+
+ Args:
+ read_buffer_size: Read buffer size in bytes. Defaults to 128 KiB.
+ write_buffer_size: Write buffer size in bytes. Defaults to 128 KiB.
+ max_write_buffer_size: Maximum write buffer size in bytes.
Defaults to
+ unbounded, which reads back as the largest value a
pointer-sized
+ unsigned integer holds rather than as `None`.
+ max_message_size: Maximum message size in bytes, or an explicit
`None`
+ to lift the limit entirely. Omitting the argument is not the
same
+ as passing `None`: it keeps the underlying default of 64 MiB.
+ Lifting the limit lets a peer queue an arbitrarily large
message
+ in memory, so prefer a finite value.
+ max_frame_size: Maximum frame size in bytes, or an explicit `None`
to
+ lift the limit entirely. Omitting the argument keeps the
+ underlying default of 16 MiB, with the same caveat as
+ `max_message_size`.
+ accept_unmasked_frames: Whether to accept unmasked frames.
Defaults to
+ `False`; clients should typically keep this off for RFC
compliance.
+
+ Raises:
+ ValueError: If a numeric field is outside the range of a
pointer-sized
+ unsigned integer, or if `max_write_buffer_size` does not come
out
+ greater than `write_buffer_size`. tungstenite enforces the same
+ invariant with an `assert!` at connect time, which would
otherwise
+ surface as an unrecoverable Rust panic instead of a
`ValueError`.
+ OverflowError: If a numeric field does not fit a signed 128-bit
integer,
+ raised by the underlying conversion before this constructor
runs.
+ """
+ def __repr__(self) -> builtins.str: ...
+
[email protected]
+class WebSocketReconnectionConfig:
+ r"""
+ How the WebSocket client reconnects after the connection to the server is
lost.
+ """
+ @property
+ def enabled(self) -> builtins.bool: ...
+ @property
+ def max_retries(self) -> builtins.int | None: ...
+ @property
+ def interval(self) -> datetime.timedelta: ...
+ @property
+ def reestablish_after(self) -> datetime.timedelta: ...
+ def __new__(
+ cls,
+ *,
+ enabled: builtins.bool | None = None,
+ max_retries: builtins.int | None = None,
+ interval: datetime.timedelta | None = None,
+ reestablish_after: datetime.timedelta | None = None,
+ ) -> WebSocketReconnectionConfig:
+ r"""
+ Constructs a reconnection policy.
+
+ Args:
+ enabled: Whether to reconnect at all. Defaults to enabled.
+ max_retries: Redials of the configured server address after the
first
+ attempt, or `None` for unlimited; `0` still makes that first
+ attempt. Unlike the TCP transport, WebSocket redials the one
+ address it was configured with rather than walking a cluster
+ roster, so this counts dials. Defaults to unlimited, which
means
+ a call awaited while the server is down never returns:
+ `connect()` waits inside the retry loop, as do
`send_messages()`
+ and `poll_messages()` once auto-login is configured. Set a
finite
+ number for request/reply style usage, so a call fails instead.
+ interval: Delay before each redial. Defaults to 1 second.
+ reestablish_after: Cooldown before redialing after a previously
+ successful connection, measured from when it was established,
so
+ a session that outlived the interval is redialed at once.
Applied
+ from the first redial onward, not to the initial connect.
+ Defaults to 5 seconds.
+
+ Raises:
+ ValueError: If a duration is negative, if `max_retries` is outside
the
+ range of an unsigned 32-bit integer, or if `interval` is zero.
+ OverflowError: If `max_retries` does not fit a signed 64-bit
integer,
+ raised by the underlying conversion before this constructor
runs.
+ """
+ def __repr__(self) -> builtins.str: ...
+
@typing.final
class UserStatus(enum.Enum):
r"""
diff --git a/foreign/python/docker-compose.test.yml
b/foreign/python/docker-compose.test.yml
index 665df9121..c16eded3c 100644
--- a/foreign/python/docker-compose.test.yml
+++ b/foreign/python/docker-compose.test.yml
@@ -33,6 +33,7 @@ services:
- "3000:3000"
- "8080:8080"
- "8090:8090"
+ - "8092:8092"
environment:
- IGGY_HTTP_ADDRESS=0.0.0.0:3000
- IGGY_TCP_ADDRESS=0.0.0.0:8090
@@ -65,6 +66,7 @@ services:
- IGGY_SERVER_TCP_PORT=8090
- IGGY_SERVER_HTTP_PORT=3000
- IGGY_SERVER_QUIC_PORT=8080
+ - IGGY_SERVER_WS_PORT=8092
- PYTHONPATH=/workspace/foreign/python
- PYTEST_ARGS=-v --tb=short
volumes:
diff --git a/foreign/python/src/client.rs b/foreign/python/src/client.rs
index 168ef2f6d..0fdef39d9 100644
--- a/foreign/python/src/client.rs
+++ b/foreign/python/src/client.rs
@@ -94,17 +94,17 @@ fn resolve_topic_params(
#[pymethods]
impl IggyClient {
/// Constructs a new IggyClient from a TCP server address, a `TcpConfig`, a
- /// `QuicConfig`, or an `HttpConfig`. This initializes a new runtime for
- /// asynchronous operations.
+ /// `QuicConfig`, an `HttpConfig`, or a `WebSocketConfig`. This
initializes a
+ /// new runtime for asynchronous operations.
/// Future versions might utilize asyncio for more Pythonic async.
///
/// Args:
- /// conn: A `host:port` address, a `TcpConfig`, a `QuicConfig`, or an
- /// `HttpConfig`. Defaults to `127.0.0.1:8090` over TCP with
auto-login
- /// disabled. A malformed address is reported differently
depending on
- /// the form: the string form raises `RuntimeError` here, while
- /// `TcpConfig`/`QuicConfig`/`HttpConfig` raise `ValueError` when
they
- /// are constructed, before any of them ever reaches this call.
Neither
+ /// conn: A `host:port` address, a `TcpConfig`, a `QuicConfig`, an
+ /// `HttpConfig`, or a `WebSocketConfig`. Defaults to
`127.0.0.1:8090`
+ /// over TCP with auto-login disabled. A malformed address is
reported
+ /// differently depending on the form: the string form raises
+ /// `RuntimeError` here, while every config type raises
`ValueError`
+ /// when it is constructed, before any of them reaches this call.
Neither
/// exception is a subclass of the other.
///
/// Raises:
@@ -115,7 +115,7 @@ impl IggyClient {
#[pyo3(signature = (conn=None))]
fn new(
#[gen_stub(override_type(
- type_repr = "TcpConfig | QuicConfig | HttpConfig | builtins.str |
None"
+ type_repr = "TcpConfig | QuicConfig | HttpConfig | WebSocketConfig
| builtins.str | None"
))]
conn: Option<PyClientConfig>,
) -> PyResult<Self> {
@@ -146,6 +146,9 @@ impl IggyClient {
Some(PyClientConfig::Http(config)) => ClientWrapper::Http(
HttpClient::create(config.client_config()).map_err(to_runtime_error)?,
),
+ Some(PyClientConfig::WebSocket(config)) =>
ClientWrapper::WebSocket(
+
WebSocketClient::create(config.client_config()).map_err(to_runtime_error)?,
+ ),
None => ClientWrapper::Tcp(
TcpClient::create(Arc::new(TcpClientConfig::default()))
.map_err(to_runtime_error)?,
diff --git a/foreign/python/src/config.rs b/foreign/python/src/config.rs
index f699a2359..e685d048c 100644
--- a/foreign/python/src/config.rs
+++ b/foreign/python/src/config.rs
@@ -22,6 +22,9 @@ use iggy::prelude::{
QuicClientReconnectionConfig as RustQuicClientReconnectionConfig,
TcpClientConfig as RustTcpClientConfig, TcpClientConfigBuilder,
TcpClientReconnectionConfig as RustTcpClientReconnectionConfig,
+ WebSocketClientConfig as RustWebSocketClientConfig,
WebSocketClientConfigBuilder,
+ WebSocketClientReconnectionConfig as RustWebSocketClientReconnectionConfig,
+ WebSocketConfig as RustWebSocketFramingConfig,
};
use pyo3::exceptions::PyValueError;
use pyo3::prelude::*;
@@ -571,7 +574,7 @@ impl QuicConfig {
/// seconds) instead, since `configure()` skips the setter
entirely when
/// zero. Defaults to 10 seconds.
/// validate_certificate: Whether to validate the server certificate.
Defaults
- /// to disabled, unlike the TCP and WebSocket transports.
+ /// to disabled; only the TCP transport validates by default.
///
/// Raises:
/// ValueError: If `server_address` or `client_address` is not a valid
@@ -964,6 +967,466 @@ impl HttpConfig {
}
}
+/// How the WebSocket client reconnects after the connection to the server is
lost.
+#[gen_stub_pyclass]
+#[pyclass(from_py_object)]
+#[derive(Clone)]
+pub struct WebSocketReconnectionConfig {
+ pub(crate) inner: RustWebSocketClientReconnectionConfig,
+}
+
+#[gen_stub_pymethods]
+#[pymethods]
+impl WebSocketReconnectionConfig {
+ /// Constructs a reconnection policy.
+ ///
+ /// Args:
+ /// enabled: Whether to reconnect at all. Defaults to enabled.
+ /// max_retries: Redials of the configured server address after the
first
+ /// attempt, or `None` for unlimited; `0` still makes that first
+ /// attempt. Unlike the TCP transport, WebSocket redials the one
+ /// address it was configured with rather than walking a cluster
+ /// roster, so this counts dials. Defaults to unlimited, which
means
+ /// a call awaited while the server is down never returns:
+ /// `connect()` waits inside the retry loop, as do
`send_messages()`
+ /// and `poll_messages()` once auto-login is configured. Set a
finite
+ /// number for request/reply style usage, so a call fails instead.
+ /// interval: Delay before each redial. Defaults to 1 second.
+ /// reestablish_after: Cooldown before redialing after a previously
+ /// successful connection, measured from when it was established,
so
+ /// a session that outlived the interval is redialed at once.
Applied
+ /// from the first redial onward, not to the initial connect.
+ /// Defaults to 5 seconds.
+ ///
+ /// Raises:
+ /// ValueError: If a duration is negative, if `max_retries` is outside
the
+ /// range of an unsigned 32-bit integer, or if `interval` is zero.
+ /// OverflowError: If `max_retries` does not fit a signed 64-bit
integer,
+ /// raised by the underlying conversion before this constructor
runs.
+ #[new]
+ #[pyo3(signature = (*, enabled=None, max_retries=None, interval=None,
reestablish_after=None))]
+ fn new(
+ #[gen_stub(override_type(type_repr = "builtins.bool | None"))]
enabled: Option<bool>,
+ #[gen_stub(override_type(type_repr = "builtins.int | None"))]
max_retries: Option<i64>,
+ #[gen_stub(override_type(type_repr = "datetime.timedelta | None",
imports=("datetime")))]
+ interval: Option<Py<PyDelta>>,
+ #[gen_stub(override_type(type_repr = "datetime.timedelta | None",
imports=("datetime")))]
+ reestablish_after: Option<Py<PyDelta>>,
+ ) -> PyResult<Self> {
+ let defaults = RustWebSocketClientReconnectionConfig::default();
+ let enabled = enabled.unwrap_or(defaults.enabled);
+ let max_retries = max_retries
+ .map(|max_retries| u32_param(max_retries, "max_retries"))
+ .transpose()?;
+ let interval = interval
+ .as_ref()
+ .map(py_delta_to_iggy_duration)
+ .transpose()?
+ .map(|interval| reject_zero(interval, "interval"))
+ .transpose()?
+ .unwrap_or(defaults.interval);
+ Ok(Self {
+ inner: RustWebSocketClientReconnectionConfig {
+ enabled,
+ max_retries,
+ interval,
+ reestablish_after: reestablish_after
+ .as_ref()
+ .map(py_delta_to_iggy_duration)
+ .transpose()?
+ .unwrap_or(defaults.reestablish_after),
+ },
+ })
+ }
+
+ #[getter]
+ fn enabled(&self) -> bool {
+ self.inner.enabled
+ }
+
+ #[gen_stub(override_return_type(type_repr = "builtins.int | None"))]
+ #[getter]
+ fn max_retries(&self) -> Option<u32> {
+ self.inner.max_retries
+ }
+
+ #[gen_stub(override_return_type(type_repr = "datetime.timedelta",
imports=("datetime")))]
+ #[getter]
+ fn interval<'a>(&self, py: Python<'a>) -> PyResult<Bound<'a, PyDelta>> {
+ iggy_duration_to_py_delta(py, self.inner.interval.get())
+ }
+
+ #[gen_stub(override_return_type(type_repr = "datetime.timedelta",
imports=("datetime")))]
+ #[getter]
+ fn reestablish_after<'a>(&self, py: Python<'a>) -> PyResult<Bound<'a,
PyDelta>> {
+ iggy_duration_to_py_delta(py, self.inner.reestablish_after)
+ }
+
+ fn __repr__(&self) -> String {
+ let max_retries = match self.inner.max_retries {
+ Some(max_retries) => max_retries.to_string(),
+ None => "None".to_owned(),
+ };
+ format!(
+ "WebSocketReconnectionConfig(enabled={},
max_retries={max_retries}, interval={}, reestablish_after={})",
+ python_bool(self.inner.enabled),
+ duration_repr(self.inner.interval.get()),
+ duration_repr(self.inner.reestablish_after),
+ )
+ }
+}
+
+/// Frame- and buffer-level options passed through to the underlying WebSocket
+/// implementation, accepted by `WebSocketConfig`'s `framing` argument.
+///
+/// Every field is keyword-only and optional; unset fields fall back to the
+/// underlying WebSocket library's own defaults.
+#[gen_stub_pyclass]
+#[pyclass(from_py_object)]
+#[derive(Clone)]
+pub struct WebSocketFramingConfig {
+ pub(crate) inner: RustWebSocketFramingConfig,
+}
+
+#[gen_stub_pymethods]
+#[pymethods]
+impl WebSocketFramingConfig {
+ /// Constructs a WebSocket framing configuration.
+ ///
+ /// Args:
+ /// read_buffer_size: Read buffer size in bytes. Defaults to 128 KiB.
+ /// write_buffer_size: Write buffer size in bytes. Defaults to 128 KiB.
+ /// max_write_buffer_size: Maximum write buffer size in bytes.
Defaults to
+ /// unbounded, which reads back as the largest value a
pointer-sized
+ /// unsigned integer holds rather than as `None`.
+ /// max_message_size: Maximum message size in bytes, or an explicit
`None`
+ /// to lift the limit entirely. Omitting the argument is not the
same
+ /// as passing `None`: it keeps the underlying default of 64 MiB.
+ /// Lifting the limit lets a peer queue an arbitrarily large
message
+ /// in memory, so prefer a finite value.
+ /// max_frame_size: Maximum frame size in bytes, or an explicit `None`
to
+ /// lift the limit entirely. Omitting the argument keeps the
+ /// underlying default of 16 MiB, with the same caveat as
+ /// `max_message_size`.
+ /// accept_unmasked_frames: Whether to accept unmasked frames.
Defaults to
+ /// `False`; clients should typically keep this off for RFC
compliance.
+ ///
+ /// Raises:
+ /// ValueError: If a numeric field is outside the range of a
pointer-sized
+ /// unsigned integer, or if `max_write_buffer_size` does not come
out
+ /// greater than `write_buffer_size`. tungstenite enforces the same
+ /// invariant with an `assert!` at connect time, which would
otherwise
+ /// surface as an unrecoverable Rust panic instead of a
`ValueError`.
+ /// OverflowError: If a numeric field does not fit a signed 128-bit
integer,
+ /// raised by the underlying conversion before this constructor
runs.
+ #[new]
+ #[pyo3(signature = (
+ *,
+ read_buffer_size=None,
+ write_buffer_size=None,
+ max_write_buffer_size=None,
+ max_message_size=64 << 20,
+ max_frame_size=16 << 20,
+ accept_unmasked_frames=None,
+ ))]
+ fn new(
+ #[gen_stub(override_type(type_repr = "builtins.int | None"))]
read_buffer_size: Option<
+ i128,
+ >,
+ #[gen_stub(override_type(type_repr = "builtins.int | None"))]
write_buffer_size: Option<
+ i128,
+ >,
+ #[gen_stub(override_type(type_repr = "builtins.int | None"))]
max_write_buffer_size: Option<
+ i128,
+ >,
+ #[gen_stub(override_type(type_repr = "builtins.int | None"))]
max_message_size: Option<
+ i128,
+ >,
+ #[gen_stub(override_type(type_repr = "builtins.int | None"))]
max_frame_size: Option<i128>,
+ #[gen_stub(override_type(type_repr = "builtins.bool | None"))]
+ accept_unmasked_frames: Option<bool>,
+ ) -> PyResult<Self> {
+ let mut inner = RustWebSocketFramingConfig::default();
+ if let Some(read_buffer_size) = read_buffer_size {
+ inner.read_buffer_size = Some(usize_param(read_buffer_size,
"read_buffer_size")?);
+ }
+ if let Some(write_buffer_size) = write_buffer_size {
+ inner.write_buffer_size = Some(usize_param(write_buffer_size,
"write_buffer_size")?);
+ }
+ if let Some(max_write_buffer_size) = max_write_buffer_size {
+ inner.max_write_buffer_size =
+ Some(usize_param(max_write_buffer_size,
"max_write_buffer_size")?);
+ }
+ // Assigned unconditionally, unlike the buffer sizes above: `None` here
+ // means "no limit", and pyo3 cannot tell an omitted argument from an
+ // explicit `None` on its own. The signature defaults carry the
+ // underlying limits instead, so omission lands on `Some(default)` and
+ // only an explicit `None` reaches this as `None`.
+ inner.max_message_size = max_message_size
+ .map(|max_message_size| usize_param(max_message_size,
"max_message_size"))
+ .transpose()?;
+ inner.max_frame_size = max_frame_size
+ .map(|max_frame_size| usize_param(max_frame_size,
"max_frame_size"))
+ .transpose()?;
+ if let Some(accept_unmasked_frames) = accept_unmasked_frames {
+ inner.accept_unmasked_frames = accept_unmasked_frames;
+ }
+ if let (Some(write_buffer_size), Some(max_write_buffer_size)) =
+ (inner.write_buffer_size, inner.max_write_buffer_size)
+ && max_write_buffer_size <= write_buffer_size
+ {
+ return Err(PyValueError::new_err(format!(
+ "'max_write_buffer_size' ({max_write_buffer_size}) must be
greater than \
+ 'write_buffer_size' ({write_buffer_size})"
+ )));
+ }
+
+ Ok(Self { inner })
+ }
+
+ #[gen_stub(override_return_type(type_repr = "builtins.int | None"))]
+ #[getter]
+ fn read_buffer_size(&self) -> Option<usize> {
+ self.inner.read_buffer_size
+ }
+
+ #[gen_stub(override_return_type(type_repr = "builtins.int | None"))]
+ #[getter]
+ fn write_buffer_size(&self) -> Option<usize> {
+ self.inner.write_buffer_size
+ }
+
+ #[gen_stub(override_return_type(type_repr = "builtins.int | None"))]
+ #[getter]
+ fn max_write_buffer_size(&self) -> Option<usize> {
+ self.inner.max_write_buffer_size
+ }
+
+ #[gen_stub(override_return_type(type_repr = "builtins.int | None"))]
+ #[getter]
+ fn max_message_size(&self) -> Option<usize> {
+ self.inner.max_message_size
+ }
+
+ #[gen_stub(override_return_type(type_repr = "builtins.int | None"))]
+ #[getter]
+ fn max_frame_size(&self) -> Option<usize> {
+ self.inner.max_frame_size
+ }
+
+ #[getter]
+ fn accept_unmasked_frames(&self) -> bool {
+ self.inner.accept_unmasked_frames
+ }
+
+ fn __repr__(&self) -> String {
+ let optional_usize = |value: Option<usize>| match value {
+ Some(value) => value.to_string(),
+ None => "None".to_owned(),
+ };
+ format!(
+ "WebSocketFramingConfig(read_buffer_size={}, write_buffer_size={},
max_write_buffer_size={}, max_message_size={}, max_frame_size={},
accept_unmasked_frames={})",
+ optional_usize(self.inner.read_buffer_size),
+ optional_usize(self.inner.write_buffer_size),
+ optional_usize(self.inner.max_write_buffer_size),
+ optional_usize(self.inner.max_message_size),
+ optional_usize(self.inner.max_frame_size),
+ python_bool(self.inner.accept_unmasked_frames),
+ )
+ }
+}
+
+/// Configuration for the WebSocket transport, accepted by `IggyClient(...)`.
+///
+/// Every field is keyword-only and optional.
+#[gen_stub_pyclass]
+#[pyclass(from_py_object)]
+#[derive(Clone)]
+pub struct WebSocketConfig {
+ inner: Arc<RustWebSocketClientConfig>,
+}
+
+impl WebSocketConfig {
+ /// The configuration in the shape `WebSocketClient::create` expects.
+ pub(crate) fn client_config(&self) -> Arc<RustWebSocketClientConfig> {
+ self.inner.clone()
+ }
+}
+
+#[gen_stub_pymethods]
+#[pymethods]
+impl WebSocketConfig {
+ /// Constructs a WebSocket configuration.
+ ///
+ /// Args:
+ /// server_address: `host:port` of the Iggy server. Defaults to
`127.0.0.1:8092`.
+ /// auto_login: Credentials replayed on every connect. Defaults to
`AutoLogin.disabled()`.
+ /// reconnection: Reconnection policy. Defaults to
`WebSocketReconnectionConfig()`.
+ /// heartbeat_interval: Interval of heartbeats sent by the client.
Defaults to 5 seconds.
+ /// framing: Frame- and buffer-level options. Defaults to
`WebSocketFramingConfig()`.
+ /// tls_enabled: Whether to connect over TLS. Defaults to disabled.
+ /// tls_domain: Domain to validate the certificate against. Defaults to
+ /// `localhost`. Empty means it is taken from the IP
`server_address`
+ /// resolves to.
+ /// tls_ca_file: Path to the CA file for TLS. Read only when
`tls_enabled`
+ /// and `tls_validate_certificate` are both on; with either one
off it
+ /// is kept but never consulted, so pairing it with
+ /// `tls_validate_certificate=False` pins nothing.
+ /// tls_validate_certificate: Whether to validate the server
certificate.
+ /// Defaults to `False`; only the TCP transport validates by
default.
+ /// Disabling this accepts any certificate the server presents,
+ /// including self-signed and mismatched ones, and takes precedence
+ /// over `tls_ca_file`.
+ ///
+ /// Raises:
+ /// ValueError: If `server_address` is not a valid `host:port` pair,
if a
+ /// duration is negative, or if `heartbeat_interval` is zero.
+ #[new]
+ #[pyo3(signature = (
+ *,
+ server_address=None,
+ auto_login=None,
+ reconnection=None,
+ heartbeat_interval=None,
+ framing=None,
+ tls_enabled=None,
+ tls_domain=None,
+ tls_ca_file=None,
+ tls_validate_certificate=None,
+ ))]
+ #[allow(clippy::too_many_arguments)]
+ fn new(
+ #[gen_stub(override_type(type_repr = "builtins.str | None"))]
server_address: Option<
+ String,
+ >,
+ #[gen_stub(override_type(type_repr = "AutoLogin | None"))] auto_login:
Option<AutoLogin>,
+ #[gen_stub(override_type(type_repr = "WebSocketReconnectionConfig |
None"))]
+ reconnection: Option<WebSocketReconnectionConfig>,
+ #[gen_stub(override_type(type_repr = "datetime.timedelta | None",
imports=("datetime")))]
+ heartbeat_interval: Option<Py<PyDelta>>,
+ #[gen_stub(override_type(type_repr = "WebSocketFramingConfig |
None"))] framing: Option<
+ WebSocketFramingConfig,
+ >,
+ #[gen_stub(override_type(type_repr = "builtins.bool | None"))]
tls_enabled: Option<bool>,
+ #[gen_stub(override_type(type_repr = "builtins.str | None"))]
tls_domain: Option<String>,
+ #[gen_stub(override_type(type_repr = "builtins.str | None"))]
tls_ca_file: Option<String>,
+ #[gen_stub(override_type(type_repr = "builtins.bool | None"))]
+ tls_validate_certificate: Option<bool>,
+ ) -> PyResult<Self> {
+ // The builder starts from `WebSocketClientConfig::default()`, and its
+ // `build()` trims and validates the address whether or not one was
set here.
+ let mut builder = WebSocketClientConfigBuilder::new();
+ if let Some(server_address) = server_address {
+ builder = builder.with_server_address(server_address);
+ }
+ let mut inner = builder
+ .build()
+ .map_err(|e| invalid_address("server_address", e))?;
+ if let Some(auto_login) = auto_login {
+ inner.auto_login = auto_login.inner;
+ }
+ if let Some(reconnection) = reconnection {
+ inner.reconnection = reconnection.inner;
+ }
+ if let Some(heartbeat_interval) = heartbeat_interval {
+ inner.heartbeat_interval = reject_zero(
+ py_delta_to_iggy_duration(&heartbeat_interval)?,
+ "heartbeat_interval",
+ )?;
+ }
+ if let Some(framing) = framing {
+ inner.ws_config = framing.inner;
+ }
+ if let Some(tls_enabled) = tls_enabled {
+ inner.tls_enabled = tls_enabled;
+ }
+ if let Some(tls_domain) = tls_domain {
+ inner.tls_domain = tls_domain;
+ }
+ if tls_ca_file.is_some() {
+ inner.tls_ca_file = tls_ca_file;
+ }
+ if let Some(tls_validate_certificate) = tls_validate_certificate {
+ inner.tls_validate_certificate = tls_validate_certificate;
+ }
+
+ Ok(Self {
+ inner: Arc::new(inner),
+ })
+ }
+
+ #[getter]
+ fn server_address(&self) -> String {
+ self.inner.server_address.clone()
+ }
+
+ #[getter]
+ fn auto_login(&self) -> AutoLogin {
+ AutoLogin {
+ inner: self.inner.auto_login.clone(),
+ }
+ }
+
+ #[getter]
+ fn reconnection(&self) -> WebSocketReconnectionConfig {
+ WebSocketReconnectionConfig {
+ inner: self.inner.reconnection.clone(),
+ }
+ }
+
+ #[gen_stub(override_return_type(type_repr = "datetime.timedelta",
imports=("datetime")))]
+ #[getter]
+ fn heartbeat_interval<'a>(&self, py: Python<'a>) -> PyResult<Bound<'a,
PyDelta>> {
+ iggy_duration_to_py_delta(py, self.inner.heartbeat_interval.get())
+ }
+
+ #[getter]
+ fn framing(&self) -> WebSocketFramingConfig {
+ WebSocketFramingConfig {
+ inner: self.inner.ws_config.clone(),
+ }
+ }
+
+ #[getter]
+ fn tls_enabled(&self) -> bool {
+ self.inner.tls_enabled
+ }
+
+ #[getter]
+ fn tls_domain(&self) -> String {
+ self.inner.tls_domain.clone()
+ }
+
+ #[gen_stub(override_return_type(type_repr = "builtins.str | None"))]
+ #[getter]
+ fn tls_ca_file(&self) -> Option<String> {
+ self.inner.tls_ca_file.clone()
+ }
+
+ #[getter]
+ fn tls_validate_certificate(&self) -> bool {
+ self.inner.tls_validate_certificate
+ }
+
+ fn __repr__(&self) -> String {
+ let tls_ca_file = match &self.inner.tls_ca_file {
+ Some(tls_ca_file) => format!("{tls_ca_file:?}"),
+ None => "None".to_owned(),
+ };
+ format!(
+ "WebSocketConfig(server_address={:?}, auto_login={},
reconnection={}, heartbeat_interval={}, framing={}, tls_enabled={},
tls_domain={:?}, tls_ca_file={tls_ca_file}, tls_validate_certificate={})",
+ self.inner.server_address,
+ self.auto_login().__repr__(),
+ self.reconnection().__repr__(),
+ duration_repr(self.inner.heartbeat_interval.get()),
+ self.framing().__repr__(),
+ python_bool(self.inner.tls_enabled),
+ self.inner.tls_domain,
+ python_bool(self.inner.tls_validate_certificate),
+ )
+ }
+}
+
fn python_bool(value: bool) -> &'static str {
if value { "True" } else { "False" }
}
@@ -1020,8 +1483,24 @@ fn varint_param(value: i64, parameter: &str) ->
PyResult<u64> {
Ok(value)
}
+/// Converts a Python int to the unsigned pointer-sized integer a WebSocket
+/// framing field expects, naming the parameter in the error so a caller can
+/// tell which argument was out of range. Extracted as an `i128` rather than an
+/// `i64` so the whole `usize` range survives the way in:
`max_write_buffer_size`
+/// defaults to `usize::MAX`, which an `i64` parameter would refuse to take
back
+/// with an unnamed `OverflowError`, breaking `eval(repr(config))`.
+fn usize_param(value: i128, parameter: &str) -> PyResult<usize> {
+ usize::try_from(value).map_err(|_| {
+ PyValueError::new_err(format!(
+ "'{parameter}' must be between 0 and {}",
+ usize::MAX
+ ))
+ })
+}
+
/// What `IggyClient(...)` accepts: a bare `host:port`, a full `TcpConfig`, a
-/// `QuicConfig` for the QUIC transport, or an `HttpConfig` for the HTTP
transport.
+/// `QuicConfig` for the QUIC transport, an `HttpConfig` for the HTTP
transport,
+/// or a `WebSocketConfig` for the WebSocket transport.
#[derive(FromPyObject)]
pub enum PyClientConfig {
#[pyo3(transparent)]
@@ -1030,7 +1509,51 @@ pub enum PyClientConfig {
Quic(QuicConfig),
#[pyo3(transparent)]
Http(HttpConfig),
+ #[pyo3(transparent)]
+ WebSocket(WebSocketConfig),
#[pyo3(transparent, annotation = "str")]
ServerAddress(String),
}
-impl_stub_type!(PyClientConfig = TcpConfig | QuicConfig | HttpConfig | String);
+impl_stub_type!(PyClientConfig = TcpConfig | QuicConfig | HttpConfig |
WebSocketConfig | String);
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ /// Mirrors the literal in `WebSocketFramingConfig::new`'s signature.
+ const DEFAULT_MAX_MESSAGE_SIZE: usize = 64 << 20;
+
+ /// Mirrors the literal in `WebSocketFramingConfig::new`'s signature.
+ const DEFAULT_MAX_FRAME_SIZE: usize = 16 << 20;
+
+ /// The signature defaults have to be literals for the generated stub to
stay
+ /// valid Python, so nothing but this test stops them drifting from the SDK
+ /// (and so from tungstenite) on a dependency bump.
+ #[test]
+ fn defaults_should_match_the_sdk() {
+ let defaults = RustWebSocketFramingConfig::default();
+
+ assert_eq!(
+ defaults.max_message_size,
+ Some(DEFAULT_MAX_MESSAGE_SIZE),
+ "'max_message_size' drifted from the SDK, update the literal in \
+ WebSocketFramingConfig::new's signature too"
+ );
+ assert_eq!(
+ defaults.max_frame_size,
+ Some(DEFAULT_MAX_FRAME_SIZE),
+ "'max_frame_size' drifted from the SDK, update the literal in \
+ WebSocketFramingConfig::new's signature too"
+ );
+ assert!(
+ defaults.write_buffer_size.is_some(),
+ "'write_buffer_size' lost its default, so the write buffer
invariant \
+ check in WebSocketFramingConfig::new would stop running"
+ );
+ assert!(
+ defaults.max_write_buffer_size.is_some(),
+ "'max_write_buffer_size' lost its default, so the write buffer \
+ invariant check in WebSocketFramingConfig::new would stop running"
+ );
+ }
+}
diff --git a/foreign/python/src/lib.rs b/foreign/python/src/lib.rs
index 78b777918..dc238836e 100644
--- a/foreign/python/src/lib.rs
+++ b/foreign/python/src/lib.rs
@@ -34,6 +34,7 @@ mod user_headers;
use client::IggyClient;
use config::{
AutoLogin, HttpConfig, QuicConfig, QuicReconnectionConfig, TcpConfig,
TcpReconnectionConfig,
+ WebSocketConfig, WebSocketFramingConfig, WebSocketReconnectionConfig,
};
use consumer::{
AutoCommit, AutoCommitAfter, AutoCommitWhen, Consumer, ConsumerGroup,
ConsumerGroupDetails,
@@ -65,6 +66,9 @@ fn apache_iggy(_py: Python, m: &Bound<'_, PyModule>) ->
PyResult<()> {
m.add_class::<QuicConfig>()?;
m.add_class::<QuicReconnectionConfig>()?;
m.add_class::<HttpConfig>()?;
+ m.add_class::<WebSocketConfig>()?;
+ m.add_class::<WebSocketReconnectionConfig>()?;
+ m.add_class::<WebSocketFramingConfig>()?;
m.add_class::<StreamDetails>()?;
m.add_class::<Stream>()?;
m.add_class::<Stats>()?;
diff --git a/foreign/python/tests/test_websocket_config.py
b/foreign/python/tests/test_websocket_config.py
new file mode 100644
index 000000000..555e758fa
--- /dev/null
+++ b/foreign/python/tests/test_websocket_config.py
@@ -0,0 +1,522 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you 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.
+
+"""
+Tests for the WebSocket client configuration surface.
+
+`WebSocketConfig` and `WebSocketReconnectionConfig` mirror the Rust SDK
+types the same way `TcpConfig`/`TcpReconnectionConfig` do, so most of these
+assert that a value set from Python survives to the getters and that unset
+fields fall back to the Rust defaults. `WebSocketFramingConfig` is new: it
+wraps the tungstenite frame- and buffer-level options the Rust SDK nests
+under `ws_config`, exposed here as a `framing` argument rather than flat
+kwargs, mirroring the Rust struct's own nesting. `AutoLogin` is
+transport-agnostic and already covered by `test_client_config.py`.
+"""
+
+import ast
+import socket
+import sys
+import threading
+from collections.abc import Callable
+from datetime import timedelta
+
+import pytest
+
+from apache_iggy import (
+ AutoLogin,
+ IggyClient,
+ WebSocketConfig,
+ WebSocketFramingConfig,
+ WebSocketReconnectionConfig,
+)
+
+from .utils import get_websocket_server_config, wait_for_ping, wait_for_server
+
+# tungstenite's own `WebSocketConfig::default()` values, mirrored here so a
+# dependency bump that moves one fails loudly instead of silently redefining
the
+# Python surface.
+TUNGSTENITE_READ_BUFFER_SIZE = 128 * 1024
+TUNGSTENITE_WRITE_BUFFER_SIZE = 128 * 1024
+# `usize::MAX`, tungstenite's "unbounded" write buffer.
+TUNGSTENITE_MAX_WRITE_BUFFER_SIZE = sys.maxsize * 2 + 1
+TUNGSTENITE_MAX_MESSAGE_SIZE = 64 << 20
+TUNGSTENITE_MAX_FRAME_SIZE = 16 << 20
+
+# The accept thread only has to come back from a connection the client already
+# dialed, so anything beyond this is a hang, not slowness.
+ACCEPT_JOIN_TIMEOUT_SECONDS = 5.0
+
+
+def _accept_and_close(listener: socket.socket) -> None:
+ """Accept one connection and drop it, failing any WebSocket handshake."""
+ try:
+ connection, _ = listener.accept()
+ except OSError:
+ return
+ connection.close()
+
+
[email protected]
+class TestWebSocketReconnectionConfig:
+ """Test the reconnection policy."""
+
+ def test_defaults_match_the_rust_sdk(self):
+ """Test that an unconfigured policy reconnects forever, one second
apart."""
+ reconnection = WebSocketReconnectionConfig()
+
+ assert reconnection.enabled is True
+ assert reconnection.max_retries is None
+ assert reconnection.interval == timedelta(seconds=1)
+ assert reconnection.reestablish_after == timedelta(seconds=5)
+
+ def test_every_field_round_trips(self):
+ """Test that each configured field is readable back unchanged."""
+ reconnection = WebSocketReconnectionConfig(
+ enabled=False,
+ max_retries=10,
+ interval=timedelta(milliseconds=250),
+ reestablish_after=timedelta(seconds=30),
+ )
+
+ assert reconnection.enabled is False
+ assert reconnection.max_retries == 10
+ assert reconnection.interval == timedelta(milliseconds=250)
+ assert reconnection.reestablish_after == timedelta(seconds=30)
+
+ def test_arguments_are_keyword_only(self):
+ """Test that the adjacent flags cannot be passed positionally."""
+ with pytest.raises(TypeError):
+ # pyrefly: ignore # bad-argument-count
+ WebSocketReconnectionConfig(True)
+
+ @pytest.mark.parametrize(
+ "construct",
+ [
+ lambda duration: WebSocketReconnectionConfig(interval=duration),
+ lambda duration:
WebSocketReconnectionConfig(reestablish_after=duration),
+ ],
+ ids=["interval", "reestablish_after"],
+ )
+ @pytest.mark.parametrize(
+ "negative",
+ [timedelta(microseconds=-1), timedelta(seconds=-1),
timedelta(days=-1)],
+ )
+ def test_negative_duration_is_rejected(
+ self,
+ construct: Callable[[timedelta], WebSocketReconnectionConfig],
+ negative: timedelta,
+ ):
+ """Test that a negative duration fails at construction, not at
connect."""
+ with pytest.raises(ValueError, match="negative"):
+ construct(negative)
+
+ @pytest.mark.parametrize("out_of_range", [-1, 2**32])
+ def test_out_of_range_max_retries_is_rejected(self, out_of_range: int):
+ """Test that a retry count outside the wire range names the argument.
+
+ The conversion pyo3 does on its own raises OverflowError, which is not
a
+ ValueError and so escapes the handler a caller wraps construction in.
+ """
+ with pytest.raises(ValueError, match="max_retries"):
+ WebSocketReconnectionConfig(max_retries=out_of_range)
+
+ def test_zero_reestablish_after_is_allowed(self):
+ """Test that a zero cooldown is legal and readable back."""
+ reconnection =
WebSocketReconnectionConfig(reestablish_after=timedelta(0))
+
+ assert reconnection.reestablish_after == timedelta(0)
+
+ @pytest.mark.parametrize(
+ "kwargs",
+ [
+ {},
+ {"max_retries": 5},
+ {"enabled": False},
+ ],
+ ids=["unlimited_retries", "bounded_retries", "reconnection_disabled"],
+ )
+ def test_zero_interval_is_rejected(self, kwargs: dict):
+ """Test that a zero interval fails whatever the retry policy is.
+
+ The interval is a delay between passes, so zero reconnects in a
+ continuous loop.
+ """
+ with pytest.raises(ValueError, match="zero"):
+ WebSocketReconnectionConfig(interval=timedelta(0), **kwargs)
+
+ def test_very_long_interval_round_trips(self):
+ """Test that an interval beyond 68 years survives the i32 boundary."""
+ reconnection =
WebSocketReconnectionConfig(interval=timedelta(days=30_000))
+
+ assert reconnection.interval == timedelta(days=30_000)
+
+ def test_maximum_interval_round_trips(self):
+ """Test that the largest timedelta survives the day conversion."""
+ reconnection =
WebSocketReconnectionConfig(interval=timedelta(days=999_999_999))
+
+ assert reconnection.interval == timedelta(days=999_999_999)
+
+
[email protected]
+class TestWebSocketFramingConfig:
+ """Test the frame- and buffer-level options."""
+
+ def test_defaults_match_tungstenite(self):
+ """Test that unconfigured sizes fall back to the tungstenite defaults.
+
+ Every size defaults to a concrete value; an omitted argument keeps that
+ default, while an explicit `None` clears the limit. See
+ `test_explicit_none_clears_the_limit`.
+ """
+ framing = WebSocketFramingConfig()
+
+ assert framing.read_buffer_size == TUNGSTENITE_READ_BUFFER_SIZE
+ assert framing.write_buffer_size == TUNGSTENITE_WRITE_BUFFER_SIZE
+ assert framing.max_write_buffer_size ==
TUNGSTENITE_MAX_WRITE_BUFFER_SIZE
+ assert framing.max_message_size == TUNGSTENITE_MAX_MESSAGE_SIZE
+ assert framing.max_frame_size == TUNGSTENITE_MAX_FRAME_SIZE
+ assert framing.accept_unmasked_frames is False
+
+ def test_every_field_round_trips(self):
+ """Test that each configured field is readable back unchanged."""
+ framing = WebSocketFramingConfig(
+ read_buffer_size=4096,
+ write_buffer_size=4096,
+ max_write_buffer_size=8192,
+ max_message_size=16384,
+ max_frame_size=16384,
+ accept_unmasked_frames=True,
+ )
+
+ assert framing.read_buffer_size == 4096
+ assert framing.write_buffer_size == 4096
+ assert framing.max_write_buffer_size == 8192
+ assert framing.max_message_size == 16384
+ assert framing.max_frame_size == 16384
+ assert framing.accept_unmasked_frames is True
+
+ @pytest.mark.parametrize("field", ["max_message_size", "max_frame_size"])
+ def test_explicit_none_clears_the_limit(self, field: str):
+ """Test that an explicit `None` lifts the size limit.
+
+ An omitted argument and an explicit `None` both reach Rust as
+ `Option::None`, so without a sentinel the constructor cannot tell them
+ apart and a caller asking for no limit would silently keep the default.
+ """
+ framing = WebSocketFramingConfig(**{field: None})
+
+ assert getattr(framing, field) is None
+
+ @pytest.mark.parametrize("field", ["max_message_size", "max_frame_size"])
+ def test_omitting_the_argument_keeps_the_default_limit(self, field: str):
+ """Test that omitting the argument is not the same as passing
`None`."""
+ framing = WebSocketFramingConfig()
+
+ assert getattr(framing, field) is not None
+
+ def test_clearing_one_limit_leaves_the_other_alone(self):
+ """Test that the two sentinels are independent."""
+ framing = WebSocketFramingConfig(max_message_size=None)
+
+ assert framing.max_message_size is None
+ assert framing.max_frame_size is not None
+
+ def test_arguments_are_keyword_only(self):
+ """Test that the first field cannot be passed positionally."""
+ with pytest.raises(TypeError):
+ # pyrefly: ignore # bad-argument-count
+ WebSocketFramingConfig(4096)
+
+ def test_repr_shows_every_field_as_python(self):
+ """Test that repr covers every field and parses as Python."""
+ framing = WebSocketFramingConfig(
+ read_buffer_size=4096,
+ accept_unmasked_frames=True,
+ )
+
+ printed = repr(framing)
+
+ assert "read_buffer_size=4096" in printed
+ assert "accept_unmasked_frames=True" in printed
+ ast.parse(printed)
+
+ def test_default_repr_is_constructible(self):
+ """Test that the default repr survives being fed back to the
constructor.
+
+ `max_write_buffer_size` defaults to `usize::MAX`, so this only holds
+ because the constructor takes the sizes as `i128`. An `i64` parameter
+ rejects its own default with an `OverflowError`, which `ast.parse`
+ alone would not catch.
+ """
+ printed = repr(WebSocketFramingConfig())
+
+ assert repr(eval(printed)) == printed # noqa: S307
+
+ @pytest.mark.parametrize(
+ "field",
+ [
+ "read_buffer_size",
+ "write_buffer_size",
+ "max_write_buffer_size",
+ "max_message_size",
+ "max_frame_size",
+ ],
+ )
+ def test_negative_size_is_rejected(self, field: str):
+ """Test that a negative size names the argument that caused it."""
+ with pytest.raises(ValueError, match=field):
+ # pyrefly: ignore # bad-argument-type
+ WebSocketFramingConfig(**{field: -1})
+
+ def
test_max_write_buffer_size_not_greater_than_write_buffer_size_is_rejected(self):
+ """Test that a non-increasing write buffer pair fails at construction.
+
+ tungstenite enforces `max_write_buffer_size > write_buffer_size` with
an
+ `assert!` when the connection is established, which would otherwise
+ surface as an unrecoverable Rust panic instead of a catchable error.
+ """
+ with pytest.raises(ValueError, match="max_write_buffer_size"):
+ WebSocketFramingConfig(write_buffer_size=1000,
max_write_buffer_size=1000)
+
+ def
test_max_write_buffer_size_below_the_default_write_buffer_size_is_rejected(
+ self,
+ ):
+ """Test that the invariant is checked against the default too.
+
+ Setting only `max_write_buffer_size` below the untouched default
+ `write_buffer_size` (128 KiB) must fail the same way as setting both.
+ """
+ with pytest.raises(ValueError, match="max_write_buffer_size"):
+ WebSocketFramingConfig(max_write_buffer_size=1000)
+
+
[email protected]
+class TestWebSocketConfig:
+ """Test the transport configuration."""
+
+ def test_defaults_match_the_rust_sdk(self):
+ """Test that an unconfigured transport matches the Rust SDK
defaults."""
+ config = WebSocketConfig()
+
+ assert config.server_address == "127.0.0.1:8092"
+ assert config.auto_login.enabled is False
+ assert config.reconnection.enabled is True
+ assert config.heartbeat_interval == timedelta(seconds=5)
+ assert config.tls_enabled is False
+ assert config.tls_domain == "localhost"
+ assert config.tls_ca_file is None
+ # Unlike TCP and QUIC, WebSocket does not validate the server
+ # certificate by default.
+ assert config.tls_validate_certificate is False
+
+ def test_every_field_round_trips(self):
+ """Test that each configured field is readable back unchanged."""
+ config = WebSocketConfig(
+ server_address="127.0.0.1:8093",
+ auto_login=AutoLogin.username_password("iggy", "iggy"),
+ reconnection=WebSocketReconnectionConfig(max_retries=3),
+ heartbeat_interval=timedelta(seconds=15),
+ framing=WebSocketFramingConfig(read_buffer_size=4096),
+ tls_enabled=True,
+ tls_domain="example.com",
+ tls_ca_file="ca.pem",
+ tls_validate_certificate=True,
+ )
+
+ assert config.server_address == "127.0.0.1:8093"
+ assert config.auto_login.username == "iggy"
+ assert config.reconnection.max_retries == 3
+ assert config.heartbeat_interval == timedelta(seconds=15)
+ assert config.framing.read_buffer_size == 4096
+ assert config.tls_enabled is True
+ assert config.tls_domain == "example.com"
+ assert config.tls_ca_file == "ca.pem"
+ assert config.tls_validate_certificate is True
+
+ def test_arguments_are_keyword_only(self):
+ """Test that the address cannot be passed positionally."""
+ with pytest.raises(TypeError):
+ # pyrefly: ignore # bad-argument-count
+ WebSocketConfig("127.0.0.1:8092")
+
+ def test_repr_hides_the_password(self):
+ """Test that the password does not leak through repr."""
+ config = WebSocketConfig(
+ auto_login=AutoLogin.username_password("iggy", "secret")
+ )
+
+ assert "secret" not in repr(config)
+
+ def test_repr_shows_every_field_as_python(self):
+ """Test that repr covers the TLS fields and parses as Python."""
+ config = WebSocketConfig(
+ heartbeat_interval=timedelta(seconds=15),
+ tls_enabled=True,
+ tls_domain="example.com",
+ tls_ca_file="ca.pem",
+ tls_validate_certificate=True,
+ )
+
+ printed = repr(config)
+
+ assert 'tls_domain="example.com"' in printed
+ assert 'tls_ca_file="ca.pem"' in printed
+ assert "tls_validate_certificate=True" in printed
+ assert "heartbeat_interval=datetime.timedelta(seconds=15)" in printed
+ ast.parse(printed)
+
+ @pytest.mark.parametrize(
+ "invalid_address",
+ ["", "127.0.0.1", "127.0.0.1:not-a-port", "127.0.0.1:70000",
"::1:8092"],
+ )
+ def test_invalid_server_address_is_rejected(self, invalid_address: str):
+ """Test that a malformed address fails at construction, naming
itself."""
+ with pytest.raises(ValueError, match="server_address"):
+ WebSocketConfig(server_address=invalid_address)
+
+ def test_negative_heartbeat_interval_is_rejected(self):
+ """Test that a negative heartbeat interval fails at construction."""
+ with pytest.raises(ValueError, match="negative"):
+ WebSocketConfig(heartbeat_interval=timedelta(seconds=-3))
+
+ def test_zero_heartbeat_interval_is_rejected(self):
+ """Test that a zero heartbeat interval fails at construction.
+
+ Nothing downstream reads zero as "disabled"; it heartbeats in a
+ continuous loop for as long as the client lives.
+ """
+ with pytest.raises(ValueError, match="zero"):
+ WebSocketConfig(heartbeat_interval=timedelta(0))
+
+
[email protected]
+class TestWebSocketClientConstruction:
+ """Test that `IggyClient(...)` accepts a `WebSocketConfig`."""
+
+ @pytest.mark.asyncio
+ async def test_accepts_a_config(self):
+ """Test that the resulting client is actually WebSocket, not silently
TCP.
+
+ `IggyClient(...)` is not None for either union arm, so that alone never
+ pinned the transport, and neither does the error text: both arms report
+ a failed dial as `Cannot establish connection`. What separates them is
+ the outcome against a plain TCP listener. WebSocket has to complete an
+ HTTP upgrade handshake, which the listener below refuses by closing at
+ once, while a client that regressed to the TCP arm needs nothing beyond
+ the accepted socket and would connect. So the raise itself is the
proof.
+
+ The listener is a real socket on an ephemeral loopback port, so nothing
+ here depends on a sysctl, a privileged port, or a running server.
+ `max_retries=0` keeps the failure immediate instead of retrying
+ forever, which is the default.
+
+ The accept thread is joined before the block ends so it is out of
+ `accept()` before the listener closes under it.
+ """
+ with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as listener:
+ listener.bind(("127.0.0.1", 0))
+ listener.listen(1)
+ port = listener.getsockname()[1]
+ accept_thread = threading.Thread(
+ target=lambda: _accept_and_close(listener), daemon=True
+ )
+ accept_thread.start()
+
+ client = IggyClient(
+ WebSocketConfig(
+ server_address=f"127.0.0.1:{port}",
+ reconnection=WebSocketReconnectionConfig(max_retries=0),
+ )
+ )
+
+ with pytest.raises(RuntimeError, match="Cannot establish
connection"):
+ await client.connect()
+
+ accept_thread.join(timeout=ACCEPT_JOIN_TIMEOUT_SECONDS)
+
+ def test_accepts_the_default_config(self):
+ """Test that an explicit default `WebSocketConfig` is accepted."""
+ assert IggyClient(WebSocketConfig()) is not None
+
+
[email protected]
+class TestAutoLoginAgainstServer:
+ """Test that configured credentials are actually replayed on connect."""
+
+ @pytest.mark.asyncio
+ async def test_auto_login_authenticates_without_login_user(self,
unique_name):
+ """Test that a privileged call succeeds without a manual
login_user()."""
+ host, port = get_websocket_server_config()
+ wait_for_server(host, port)
+
+ client = IggyClient(
+ WebSocketConfig(
+ server_address=f"{host}:{port}",
+ auto_login=AutoLogin.username_password("iggy", "iggy"),
+ # The default reconnection policy retries forever: a missing
+ # listener would hang this test until the CI timeout instead
+ # of failing.
+ reconnection=WebSocketReconnectionConfig(enabled=False),
+ )
+ )
+ await client.connect()
+ await wait_for_ping(client)
+
+ stream_name = unique_name()
+ await client.create_stream(stream_name)
+ assert await client.get_stream(stream_name) is not None
+
+ @pytest.mark.asyncio
+ async def test_without_auto_login_a_privileged_call_is_unauthenticated(
+ self, unique_name
+ ):
+ """Test that the same call fails when no credentials are configured."""
+ host, port = get_websocket_server_config()
+ wait_for_server(host, port)
+
+ client = IggyClient(
+ WebSocketConfig(
+ server_address=f"{host}:{port}",
+ # The default reconnection policy retries forever: a missing
+ # listener would hang this test until the CI timeout instead
+ # of failing.
+ reconnection=WebSocketReconnectionConfig(enabled=False),
+ )
+ )
+ await client.connect()
+ await wait_for_ping(client)
+
+ with pytest.raises(RuntimeError):
+ await client.create_stream(unique_name())
+
+ @pytest.mark.asyncio
+ async def test_wrong_auto_login_credentials_fail(self):
+ """Test that bad configured credentials surface as a connect
failure."""
+ host, port = get_websocket_server_config()
+ wait_for_server(host, port)
+
+ client = IggyClient(
+ WebSocketConfig(
+ server_address=f"{host}:{port}",
+ auto_login=AutoLogin.username_password("iggy",
"invalid-password"),
+ reconnection=WebSocketReconnectionConfig(enabled=False),
+ )
+ )
+
+ with pytest.raises(RuntimeError):
+ await client.connect()
diff --git a/foreign/python/tests/utils.py b/foreign/python/tests/utils.py
index 5d7654cbf..b005dcabe 100644
--- a/foreign/python/tests/utils.py
+++ b/foreign/python/tests/utils.py
@@ -35,6 +35,7 @@ MAX_PASSWORD_BYTES = 100
DEFAULT_TCP_PORT = 8090
DEFAULT_QUIC_PORT = 8080
DEFAULT_HTTP_PORT = 3000
+DEFAULT_WEBSOCKET_PORT = 8092
def get_transport_config(port_env_var: str, default_port: int) -> tuple[str,
int]:
@@ -96,6 +97,16 @@ def get_http_server_config() -> tuple[str, int]:
return get_transport_config("IGGY_SERVER_HTTP_PORT", DEFAULT_HTTP_PORT)
+def get_websocket_server_config() -> tuple[str, int]:
+ """
+ Get WebSocket server configuration from environment variables or defaults.
+
+ Returns:
+ tuple: (host, port) for the Iggy server
+ """
+ return get_transport_config("IGGY_SERVER_WS_PORT", DEFAULT_WEBSOCKET_PORT)
+
+
def wait_for_server(host: str, port: int, timeout: int = 60, interval: int =
2) -> None:
"""
Wait for the server to become available.
