You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by gk...@apache.org on 2025/10/09 18:56:54 UTC
(iggy) branch io_uring_tpc updated: feat(io_uring): Add WebSocket server support to Iggy server (#2250)
This is an automated email from the ASF dual-hosted git repository.
gkoszyk pushed a commit to branch io_uring_tpc
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/io_uring_tpc by this push:
new fd1b5d612 feat(io_uring): Add WebSocket server support to Iggy server (#2250)
fd1b5d612 is described below
commit fd1b5d6126b81ac34fb702ee6c94fc1d6c59ef16
Author: Krishna Vishal <kr...@gmail.com>
AuthorDate: Fri Oct 10 00:26:49 2025 +0530
feat(io_uring): Add WebSocket server support to Iggy server (#2250)
This PR introduces WebSocket server support to Iggy, enabling WebSockets
alongside the existing TCP, QUIC, and HTTP transports. WS server is
basically similar to how TCP server works but only websocket protocol
for transport.
- Extended iggy's client SDK with `WebSocketClient` implementation, with
similar API as TCP/QUIC/HTTP clients. Auto-login support, connection
string parsing, custom heart beat interval etc.
- Built and integrated a new WebSocket library
([`compio-ws`](https://github.com/krishvishal/compio-ws)) specifically
designed for compio's io_uring-based runtime. It uses tungstenite under
the hood to handle the protocol and compio to drive the IO.
- Added `WebSocket` transport to integration tests.
- Added `websocket-test` to `examples`. We can test some commands using
it to test websocket functionality.
Limitation:
- `compio-ws` can not handle the case where the message is larger than
its internal buffers in
[SyncStream](https://docs.rs/compio/latest/compio/io/compat/struct.SyncStream.html)
. This will be fixed in the future.
- Currently only works with plain websocket connections. I'm working on
adding `TLS` connection support.
---
Cargo.lock | 448 +++++++++--------
Cargo.toml | 21 +-
core/bench/report/src/types/transport.rs | 3 +
core/bench/src/utils/client_factory.rs | 4 +
core/bench/src/utils/mod.rs | 1 +
core/bench/src/utils/server_starter.rs | 18 +
core/common/Cargo.toml | 1 +
core/common/src/error/iggy_error.rs | 12 +
core/common/src/lib.rs | 4 +
core/common/src/types/args/mod.rs | 63 +++
core/common/src/types/configuration/mod.rs | 1 +
core/common/src/types/configuration/transport.rs | 6 +
.../configuration/{ => websocket_config}/mod.rs | 9 +-
.../websocket_config/websocket_client_config.rs | 186 +++++++
.../websocket_client_config_builder.rs | 134 +++++
.../websocket_client_reconnection_config.rs | 61 +++
.../websocket_connection_string_options.rs | 209 ++++++++
core/configs/server.toml | 5 +
core/integration/src/lib.rs | 3 +
core/integration/src/test_server.rs | 23 +-
core/integration/src/websocket_client.rs | 53 ++
.../tests/cli/system/test_me_command.rs | 5 +
core/integration/tests/server/cg.rs | 2 +-
core/integration/tests/server/general.rs | 2 +-
core/integration/tests/server/mod.rs | 5 +
core/sdk/Cargo.toml | 2 +
core/sdk/src/client_provider.rs | 45 +-
core/sdk/src/client_wrappers/binary_client.rs | 4 +
.../src/client_wrappers/binary_cluster_client.rs | 1 +
.../binary_consumer_group_client.rs | 31 ++
.../binary_consumer_offset_client.rs | 15 +
.../src/client_wrappers/binary_message_client.rs | 23 +
.../src/client_wrappers/binary_partition_client.rs | 10 +
.../binary_personal_access_token_client.rs | 8 +
.../src/client_wrappers/binary_segment_client.rs | 5 +
.../src/client_wrappers/binary_stream_client.rs | 6 +
.../src/client_wrappers/binary_system_client.rs | 7 +
.../sdk/src/client_wrappers/binary_topic_client.rs | 30 ++
core/sdk/src/client_wrappers/binary_user_client.rs | 19 +
core/sdk/src/client_wrappers/client_wrapper.rs | 2 +
core/sdk/src/clients/client.rs | 4 +
core/sdk/src/clients/client_builder.rs | 6 +
core/sdk/src/lib.rs | 1 +
core/sdk/src/prelude.rs | 4 +-
core/sdk/src/{lib.rs => websocket/mod.rs} | 13 +-
core/sdk/src/websocket/websocket_client.rs | 539 +++++++++++++++++++++
.../src/websocket/websocket_connection_stream.rs | 219 +++++++++
.../websocket_stream.rs} | 27 +-
core/server/Cargo.toml | 3 +
core/server/src/binary/mapper.rs | 1 +
core/server/src/binary/sender.rs | 7 +
core/server/src/configs/defaults.rs | 17 +
core/server/src/configs/mod.rs | 1 +
core/server/src/configs/server.rs | 2 +
core/server/src/configs/websocket.rs | 107 ++++
core/server/src/lib.rs | 1 +
core/server/src/shard/builder.rs | 1 +
core/server/src/shard/mod.rs | 14 +
core/server/src/shard/tasks/continuous/mod.rs | 3 +-
.../shard/tasks/continuous/websocket_server.rs} | 33 +-
core/server/src/shard/transmission/event.rs | 3 +
core/server/src/websocket/connection_handler.rs | 140 ++++++
core/server/src/websocket/mod.rs | 6 +
core/server/src/websocket/websocket_listener.rs | 163 +++++++
core/server/src/websocket/websocket_sender.rs | 176 +++++++
core/server/src/websocket/websocket_server.rs | 53 ++
examples/rust/src/shared/args.rs | 32 ++
67 files changed, 2803 insertions(+), 260 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index 78044f1bf..c0fa1fec5 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -225,9 +225,9 @@ dependencies = [
[[package]]
name = "addr2line"
-version = "0.24.2"
+version = "0.25.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "dfbe277e56a376000877090da837660b4427aad530e3028d44e0bffe4f89a1c1"
+checksum = "1b5d307320b3181d6d7954e663bd7c774a838b8220fe0593c86d9fb09f498b4b"
dependencies = [
"gimli",
]
@@ -492,9 +492,9 @@ dependencies = [
[[package]]
name = "async-compression"
-version = "0.4.30"
+version = "0.4.32"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "977eb15ea9efd848bb8a4a1a2500347ed7f0bf794edf0dc3ddcf439f43d36b23"
+checksum = "5a89bce6054c720275ac2432fbba080a66a2106a44a1b804553930ca6909f4e0"
dependencies = [
"compression-codecs",
"compression-core",
@@ -613,7 +613,7 @@ dependencies = [
"crc32fast",
"futures-lite",
"pin-project",
- "thiserror 2.0.16",
+ "thiserror 2.0.17",
]
[[package]]
@@ -657,9 +657,9 @@ checksum = "c08606f8c3cbf4ce6ec8e28fb0014a2c086708fe954eaa885384a6165172e7e8"
[[package]]
name = "aws-lc-rs"
-version = "1.14.0"
+version = "1.14.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "94b8ff6c09cd57b16da53641caa860168b88c172a5ee163b0288d3d6eea12786"
+checksum = "879b6c89592deb404ba4dc0ae6b58ffd1795c78991cbb5b8bc441c48a070440d"
dependencies = [
"aws-lc-sys",
"zeroize",
@@ -667,15 +667,16 @@ dependencies = [
[[package]]
name = "aws-lc-sys"
-version = "0.31.0"
+version = "0.32.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0e44d16778acaf6a9ec9899b92cebd65580b83f685446bf2e1f5d3d732f99dcd"
+checksum = "1ba2e2516bdf37af57fc6ff047855f54abad0066e5c4fdaaeb76dabb2e05bcf5"
dependencies = [
"bindgen",
"cc",
"cmake",
"dunce",
"fs_extra",
+ "libloading",
]
[[package]]
@@ -707,11 +708,11 @@ dependencies = [
[[package]]
name = "axum"
-version = "0.8.4"
+version = "0.8.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "021e862c184ae977658b36c4500f7feac3221ca5da43e3f25bd04ab6c79a29b5"
+checksum = "98e529aee37b5c8206bb4bf4c44797127566d72f76952c970bd3d1e85de8f4e2"
dependencies = [
- "axum-core 0.5.2",
+ "axum-core 0.5.4",
"axum-macros",
"bytes",
"form_urlencoded",
@@ -727,8 +728,7 @@ dependencies = [
"mime",
"percent-encoding",
"pin-project-lite",
- "rustversion",
- "serde",
+ "serde_core",
"serde_json",
"serde_path_to_error",
"serde_urlencoded",
@@ -762,9 +762,9 @@ dependencies = [
[[package]]
name = "axum-core"
-version = "0.5.2"
+version = "0.5.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "68464cd0412f486726fb3373129ef5d2993f90c34bc2bc1c1e9943b2f4fc7ca6"
+checksum = "0ac7a6beb1182c7e30253ee75c3e918080bfb83f5a3023bcdf7209d85fd147e6"
dependencies = [
"bytes",
"futures-core",
@@ -773,7 +773,6 @@ dependencies = [
"http-body-util",
"mime",
"pin-project-lite",
- "rustversion",
"sync_wrapper",
"tower-layer",
"tower-service",
@@ -815,9 +814,9 @@ dependencies = [
[[package]]
name = "backtrace"
-version = "0.3.75"
+version = "0.3.76"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6806a6321ec58106fea15becdad98371e28d92ccbc7c8f1b3b6dd724fe8f1002"
+checksum = "bb531853791a215d7c62a30daf0dde835f381ab5de4589cfe7c649d2cbe92bd6"
dependencies = [
"addr2line",
"cfg-if",
@@ -825,7 +824,7 @@ dependencies = [
"miniz_oxide",
"object",
"rustc-demangle",
- "windows-targets 0.52.6",
+ "windows-link 0.2.0",
]
[[package]]
@@ -875,7 +874,7 @@ dependencies = [
"gloo 0.11.0",
"js-sys",
"serde_json",
- "thiserror 2.0.16",
+ "thiserror 2.0.17",
"uuid",
"wasm-bindgen",
"web-sys",
@@ -906,7 +905,7 @@ dependencies = [
"rand 0.9.2",
"serde",
"serde_json",
- "sysinfo 0.37.0",
+ "sysinfo 0.37.1",
"tracing",
"uuid",
]
@@ -1228,14 +1227,14 @@ dependencies = [
"semver",
"serde",
"serde_json",
- "thiserror 2.0.16",
+ "thiserror 2.0.17",
]
[[package]]
name = "cc"
-version = "1.2.38"
+version = "1.2.39"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "80f41ae168f955c12fb8960b057d70d0ca153fb83182b57d86380443527be7e9"
+checksum = "e1354349954c6fc9cb0deab020f27f783cf0b604e8bb754dc4658ecf0d29c35f"
dependencies = [
"find-msvc-tools",
"jobserver",
@@ -1335,9 +1334,9 @@ dependencies = [
[[package]]
name = "clap"
-version = "4.5.47"
+version = "4.5.48"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7eac00902d9d136acd712710d71823fb8ac8004ca445a89e73a41d45aa712931"
+checksum = "e2134bb3ea021b78629caa971416385309e0131b351b25e01dc16fb54e1b5fae"
dependencies = [
"clap_builder",
"clap_derive",
@@ -1345,9 +1344,9 @@ dependencies = [
[[package]]
name = "clap_builder"
-version = "4.5.47"
+version = "4.5.48"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2ad9bbf750e73b5884fb8a211a9424a1906c1e156724260fdae972f31d70e1d6"
+checksum = "c2ba64afa3c0a6df7fa517765e31314e983f51dda798ffba27b988194fb65dc9"
dependencies = [
"anstream",
"anstyle",
@@ -1398,7 +1397,7 @@ version = "0.3.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0fa961b519f0b462e3a3b4a34b64d119eeaca1d59af726fe450bbba07a9fc0a1"
dependencies = [
- "thiserror 2.0.16",
+ "thiserror 2.0.17",
]
[[package]]
@@ -1413,7 +1412,7 @@ version = "3.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "fde0e0ec90c9dfb3b4b1a0891a7dcd0e2bffde2f7efed5fe7c9bb00e5bfb915e"
dependencies = [
- "windows-sys 0.59.0",
+ "windows-sys 0.48.0",
]
[[package]]
@@ -1578,7 +1577,7 @@ dependencies = [
"quinn-proto",
"rustc-hash",
"rustls",
- "thiserror 2.0.16",
+ "thiserror 2.0.17",
"windows-sys 0.60.2",
]
@@ -1614,11 +1613,27 @@ dependencies = [
"rustls",
]
+[[package]]
+name = "compio-ws"
+version = "0.1.0"
+source = "git+https://github.com/krishvishal/compio-ws?branch=main#b6e2b4c45a341228c54b9a8b1414da1cc785ca85"
+dependencies = [
+ "compio",
+ "compio-buf",
+ "compio-io",
+ "compio-net",
+ "compio-runtime",
+ "compio-tls",
+ "rustls",
+ "rustls-pemfile",
+ "tungstenite",
+]
+
[[package]]
name = "compression-codecs"
-version = "0.4.30"
+version = "0.4.31"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "485abf41ac0c8047c07c87c72c8fb3eb5197f6e9d7ded615dfd1a00ae00a0f64"
+checksum = "ef8a506ec4b81c460798f572caead636d57d3d7e940f998160f52bd254bf2d23"
dependencies = [
"bzip2",
"compression-core",
@@ -1645,9 +1660,9 @@ dependencies = [
[[package]]
name = "config"
-version = "0.15.16"
+version = "0.15.17"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "cef036f0ecf99baef11555578630e2cca559909b4c50822dbba828c252d21c49"
+checksum = "680d3ac2fe066c43300ec831c978871e50113a708d58ab13d231bd92deca5adb"
dependencies = [
"async-trait",
"convert_case 0.6.0",
@@ -1991,7 +2006,7 @@ checksum = "881c5d0a13b2f1498e2306e82cbada78390e152d4b1378fb28a84f4dcd0dc4f3"
dependencies = [
"dispatch",
"nix",
- "windows-sys 0.61.0",
+ "windows-sys 0.61.1",
]
[[package]]
@@ -2072,7 +2087,7 @@ dependencies = [
"send_wrapper",
"serde",
"serde_urlencoded",
- "thiserror 2.0.16",
+ "thiserror 2.0.17",
"url",
]
@@ -2081,8 +2096,8 @@ name = "cyper-axum"
version = "0.4.0"
source = "git+https://github.com/krishvishal/cyper.git?rev=cd75e266df6ab0a9b9474eb7dda1735650d17db6#cd75e266df6ab0a9b9474eb7dda1735650d17db6"
dependencies = [
- "axum 0.8.4",
- "axum-core 0.5.2",
+ "axum 0.8.5",
+ "axum-core 0.5.4",
"compio",
"compio-log 0.1.0 (registry+https://github.com/rust-lang/crates.io-index)",
"cyper-core",
@@ -2196,6 +2211,12 @@ dependencies = [
"parking_lot_core 0.9.11",
]
+[[package]]
+name = "data-encoding"
+version = "2.9.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2a2330da5de22e8a3cb63252ce2abb30116bf5265e89c0e01bc17015ce30a476"
+
[[package]]
name = "dbus"
version = "0.9.9"
@@ -2237,12 +2258,12 @@ dependencies = [
[[package]]
name = "deranged"
-version = "0.5.3"
+version = "0.5.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d630bccd429a5bb5a64b5e94f693bfc48c9f8566418fda4c494cc94f911f87cc"
+checksum = "a41953f86f8a05768a6cda24def994fd2f424b04ec5c719cf89989779f199071"
dependencies = [
"powerfmt",
- "serde",
+ "serde_core",
]
[[package]]
@@ -2378,7 +2399,7 @@ dependencies = [
"libc",
"option-ext",
"redox_users",
- "windows-sys 0.61.0",
+ "windows-sys 0.61.1",
]
[[package]]
@@ -2600,7 +2621,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb"
dependencies = [
"libc",
- "windows-sys 0.61.0",
+ "windows-sys 0.61.1",
]
[[package]]
@@ -2750,9 +2771,9 @@ checksum = "1ced73b1dacfc750a6db6c0a0c3a3853c8b41997e2e2c563dc90804ae6867959"
[[package]]
name = "flatbuffers"
-version = "25.2.10"
+version = "25.9.23"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1045398c1bfd89168b5fd3f1fc11f6e70b34f6f66300c87d44d3de849463abf1"
+checksum = "09b6620799e7340ebd9968d2e0708eb82cf1971e9a16821e2091b6d6e475eed5"
dependencies = [
"bitflags 2.9.4",
"rustc_version",
@@ -2982,20 +3003,6 @@ dependencies = [
"slab",
]
-[[package]]
-name = "generator"
-version = "0.8.7"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "605183a538e3e2a9c1038635cc5c2d194e2ee8fd0d1b66b8349fad7dbacce5a2"
-dependencies = [
- "cc",
- "cfg-if",
- "libc",
- "log",
- "rustversion",
- "windows 0.61.3",
-]
-
[[package]]
name = "generic-array"
version = "0.14.7"
@@ -3062,9 +3069,9 @@ dependencies = [
[[package]]
name = "gimli"
-version = "0.31.1"
+version = "0.32.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "07e28edb80900c19c28f1072f2e8aeca7fa06b23cd4169cefe1af5aa3260783f"
+checksum = "e629b9b98ef3dd8afe6ca2bd0f89306cec16d43d907889945bc5d6687f2f13c7"
[[package]]
name = "git2"
@@ -3602,7 +3609,7 @@ dependencies = [
"pest_derive",
"serde",
"serde_json",
- "thiserror 2.0.16",
+ "thiserror 2.0.17",
]
[[package]]
@@ -3646,12 +3653,6 @@ dependencies = [
"foldhash",
]
-[[package]]
-name = "hashbrown"
-version = "0.16.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5419bdc4f6a9207fbeba6d11b604d481addf78ecd10c11ad51e76c2f6482748d"
-
[[package]]
name = "hashlink"
version = "0.10.0"
@@ -3898,7 +3899,7 @@ dependencies = [
"libc",
"percent-encoding",
"pin-project-lite",
- "socket2 0.6.0",
+ "socket2 0.5.10",
"tokio",
"tower-service",
"tracing",
@@ -3916,7 +3917,7 @@ dependencies = [
"js-sys",
"log",
"wasm-bindgen",
- "windows-core 0.62.0",
+ "windows-core 0.62.1",
]
[[package]]
@@ -4066,8 +4067,10 @@ dependencies = [
"serde",
"tokio",
"tokio-rustls",
+ "tokio-tungstenite",
"tracing",
"trait-variant",
+ "tungstenite",
"webpki-roots 1.0.2",
]
@@ -4092,7 +4095,7 @@ dependencies = [
"rand 0.9.2",
"rayon",
"serde",
- "sysinfo 0.37.0",
+ "sysinfo 0.37.1",
"tokio",
"toml 0.9.7",
"tracing",
@@ -4119,7 +4122,7 @@ dependencies = [
"serde",
"serde_json",
"tempfile",
- "thiserror 2.0.16",
+ "thiserror 2.0.17",
"tokio",
"tracing",
"tracing-subscriber",
@@ -4141,7 +4144,7 @@ dependencies = [
"iggy_binary_protocol",
"keyring",
"passterm",
- "thiserror 2.0.16",
+ "thiserror 2.0.17",
"tokio",
"tracing",
"tracing-appender",
@@ -4152,7 +4155,7 @@ dependencies = [
name = "iggy-connectors"
version = "0.1.0"
dependencies = [
- "axum 0.8.4",
+ "axum 0.8.5",
"axum-server",
"config",
"dashmap",
@@ -4170,7 +4173,7 @@ dependencies = [
"serde_json",
"serde_yml",
"strum",
- "thiserror 2.0.16",
+ "thiserror 2.0.17",
"tokio",
"toml 0.9.7",
"tower-http",
@@ -4182,7 +4185,7 @@ dependencies = [
name = "iggy-mcp"
version = "0.1.0"
dependencies = [
- "axum 0.8.4",
+ "axum 0.8.5",
"axum-server",
"config",
"dotenvy",
@@ -4192,7 +4195,7 @@ dependencies = [
"serde",
"serde_json",
"strum",
- "thiserror 2.0.16",
+ "thiserror 2.0.17",
"tokio",
"tower-http",
"tracing",
@@ -4233,6 +4236,7 @@ dependencies = [
"chrono",
"clap",
"comfy-table",
+ "compio-ws",
"crc32fast",
"derive_more 2.0.1",
"fast-async-mutex",
@@ -4243,7 +4247,7 @@ dependencies = [
"serde_json",
"serde_with",
"strum",
- "thiserror 2.0.16",
+ "thiserror 2.0.17",
"tokio",
"tracing",
]
@@ -4338,7 +4342,7 @@ dependencies = [
"serde_json",
"simd-json",
"strum_macros",
- "thiserror 2.0.16",
+ "thiserror 2.0.17",
"tokio",
"tracing",
"tracing-subscriber",
@@ -4372,7 +4376,7 @@ dependencies = [
"rand 0.9.2",
"serde",
"serde_json",
- "thiserror 2.0.16",
+ "thiserror 2.0.17",
"tokio",
"tracing",
"tracing-subscriber",
@@ -4407,7 +4411,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f8a9aa791c7b5a71b636b7a68207fdebf171ddfc593d9c8506ec4cbc527b6a84"
dependencies = [
"implicit-clone-derive",
- "indexmap 2.11.4",
+ "indexmap 1.9.3",
]
[[package]]
@@ -4438,7 +4442,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4b0f83760fb341a774ed326568e19f5a863af4a952def8c39f9ab92fd95b88e5"
dependencies = [
"equivalent",
- "hashbrown 0.16.0",
+ "hashbrown 0.15.5",
"serde",
"serde_core",
]
@@ -4674,9 +4678,9 @@ dependencies = [
[[package]]
name = "js-sys"
-version = "0.3.80"
+version = "0.3.81"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "852f13bec5eba4ba9afbeb93fd7c13fe56147f055939ae21c43a29a0ecb2702e"
+checksum = "ec48937a97411dcb524a265206ccd4c90bb711fca92b2792c407f268825b9305"
dependencies = [
"once_cell",
"wasm-bindgen",
@@ -4821,9 +4825,9 @@ checksum = "2c4a545a15244c7d945065b5d392b2d2d7f21526fba56ce51467b06ed445e8f7"
[[package]]
name = "libc"
-version = "0.2.175"
+version = "0.2.176"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6a82ae493e598baaea5209805c49bbf2ea7de956d50d7da0da1164f9c6d28543"
+checksum = "58f929b4d672ea937a23a1ab494143d968337a5f47e56d0815df1e0890ddf174"
[[package]]
name = "libdbus-sys"
@@ -4854,14 +4858,14 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "07033963ba89ebaf1584d767badaa2e8fcec21aedea6b8c0346d487d49c28667"
dependencies = [
"cfg-if",
- "windows-targets 0.53.3",
+ "windows-targets 0.48.5",
]
[[package]]
name = "liblzma"
-version = "0.4.4"
+version = "0.4.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "10bf66f4598dc77ff96677c8e763655494f00ff9c1cf79e2eb5bb07bc31f807d"
+checksum = "73c36d08cad03a3fbe2c4e7bb3a9e84c57e4ee4135ed0b065cade3d98480c648"
dependencies = [
"liblzma-sys",
]
@@ -5036,19 +5040,6 @@ dependencies = [
"logos-codegen",
]
-[[package]]
-name = "loom"
-version = "0.7.2"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "419e0dc8046cb947daa77eb95ae174acfbddb7673b4151f56d1eed8e93fbfaca"
-dependencies = [
- "cfg-if",
- "generator",
- "scoped-tls",
- "tracing",
- "tracing-subscriber",
-]
-
[[package]]
name = "lru-slab"
version = "0.1.2"
@@ -5114,9 +5105,9 @@ dependencies = [
[[package]]
name = "memchr"
-version = "2.7.5"
+version = "2.7.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "32a282da65faaf38286cf3be983213fcf1d2e2a58700e808f83f4ea9a4804bc0"
+checksum = "f52b00d39961fc5b2736ea853c9cc86238e165017a493d1d5c8eac6bdc4cc273"
[[package]]
name = "miette"
@@ -5220,23 +5211,22 @@ dependencies = [
[[package]]
name = "moka"
-version = "0.12.10"
+version = "0.12.11"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a9321642ca94a4282428e6ea4af8cc2ca4eac48ac7a6a4ea8f33f76d0ce70926"
+checksum = "8261cd88c312e0004c1d51baad2980c66528dfdb2bee62003e643a4d8f86b077"
dependencies = [
"async-lock",
"crossbeam-channel",
"crossbeam-epoch",
"crossbeam-utils",
+ "equivalent",
"event-listener",
"futures-util",
- "loom",
"parking_lot 0.12.4",
"portable-atomic",
"rustc_version",
"smallvec",
"tagptr",
- "thiserror 1.0.69",
"uuid",
]
@@ -5504,9 +5494,9 @@ dependencies = [
[[package]]
name = "object"
-version = "0.36.7"
+version = "0.37.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "62948e14d923ea95ea2c7c86c71013138b66525b86bdc08d2dcc262bdb497b87"
+checksum = "ff76201f031d8863c38aa7f905eca4f53abbfa15f609db4277d44cd8938f33fe"
dependencies = [
"memchr",
]
@@ -5633,7 +5623,7 @@ dependencies = [
"futures-sink",
"js-sys",
"pin-project-lite",
- "thiserror 2.0.16",
+ "thiserror 2.0.17",
"tracing",
]
@@ -5676,7 +5666,7 @@ dependencies = [
"opentelemetry_sdk",
"prost 0.13.5",
"reqwest",
- "thiserror 2.0.16",
+ "thiserror 2.0.17",
"tokio",
"tonic 0.13.1",
"tracing",
@@ -5713,7 +5703,7 @@ dependencies = [
"percent-encoding",
"rand 0.9.2",
"serde_json",
- "thiserror 2.0.16",
+ "thiserror 2.0.17",
]
[[package]]
@@ -5931,7 +5921,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "21e0a3a33733faeaf8651dfee72dd0f388f0c8e5ad496a3478fa5a922f49cfa8"
dependencies = [
"memchr",
- "thiserror 2.0.16",
+ "thiserror 2.0.17",
"ucd-trie",
]
@@ -6049,7 +6039,7 @@ dependencies = [
"hermit-abi",
"pin-project-lite",
"rustix",
- "windows-sys 0.61.0",
+ "windows-sys 0.61.1",
]
[[package]]
@@ -6321,9 +6311,9 @@ dependencies = [
[[package]]
name = "prost-reflect"
-version = "0.16.1"
+version = "0.16.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3c9ae1e4084d9737646934f5ac36a8d77f9a82f962f43e267a382fde4f2a903d"
+checksum = "89a3ac73ec9a9118131a4594c9d336631a07852220a1d0ae03ee36b04503a063"
dependencies = [
"logos",
"miette",
@@ -6361,7 +6351,7 @@ dependencies = [
"prost-reflect",
"prost-types 0.14.1",
"protox-parse",
- "thiserror 2.0.16",
+ "thiserror 2.0.17",
]
[[package]]
@@ -6373,7 +6363,7 @@ dependencies = [
"logos",
"miette",
"prost-types 0.14.1",
- "thiserror 2.0.16",
+ "thiserror 2.0.17",
]
[[package]]
@@ -6434,8 +6424,8 @@ dependencies = [
"quinn-udp",
"rustc-hash",
"rustls",
- "socket2 0.6.0",
- "thiserror 2.0.16",
+ "socket2 0.5.10",
+ "thiserror 2.0.17",
"tokio",
"tracing",
"web-time",
@@ -6458,7 +6448,7 @@ dependencies = [
"rustls-pki-types",
"rustls-platform-verifier",
"slab",
- "thiserror 2.0.16",
+ "thiserror 2.0.17",
"tinyvec",
"tracing",
"web-time",
@@ -6473,16 +6463,16 @@ dependencies = [
"cfg_aliases",
"libc",
"once_cell",
- "socket2 0.6.0",
+ "socket2 0.5.10",
"tracing",
"windows-sys 0.60.2",
]
[[package]]
name = "quote"
-version = "1.0.40"
+version = "1.0.41"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1885c039570dc00dcb4ff087a89e185fd56bae234ddc7f056a945bf36467248d"
+checksum = "ce25767e7b499d1b604768e7cde645d14cc8584231ea6b295e9c9eb22c02e1d1"
dependencies = [
"proc-macro2",
]
@@ -6589,9 +6579,9 @@ dependencies = [
[[package]]
name = "rcgen"
-version = "0.14.4"
+version = "0.14.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4c83367ba62b3f1dbd0f086ede4e5ebfb4713fb234dbbc5807772a31245ff46d"
+checksum = "5fae430c6b28f1ad601274e78b7dffa0546de0b73b4cd32f46723c0c2a16f7a5"
dependencies = [
"pem",
"ring",
@@ -6626,23 +6616,23 @@ checksum = "a4e608c6638b9c18977b00b475ac1f28d14e84b27d8d42f70e0bf1e3dec127ac"
dependencies = [
"getrandom 0.2.16",
"libredox",
- "thiserror 2.0.16",
+ "thiserror 2.0.17",
]
[[package]]
name = "ref-cast"
-version = "1.0.24"
+version = "1.0.25"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4a0ae411dbe946a674d89546582cea4ba2bb8defac896622d6496f14c23ba5cf"
+checksum = "f354300ae66f76f1c85c5f84693f0ce81d747e2c3f21a45fef496d89c960bf7d"
dependencies = [
"ref-cast-impl",
]
[[package]]
name = "ref-cast-impl"
-version = "1.0.24"
+version = "1.0.25"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1165225c21bff1f3bbce98f5a1f889949bc902d3575308cc7b0de30b4f6d27c7"
+checksum = "b7186006dcb21920990093f30e3dea63b7d6e977bf1256be20c3563a5db070da"
dependencies = [
"proc-macro2",
"quote",
@@ -6651,9 +6641,9 @@ dependencies = [
[[package]]
name = "regex"
-version = "1.11.2"
+version = "1.11.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "23d7fd106d8c02486a8d64e778353d1cffe08ce79ac2e82f540c86d0facf6912"
+checksum = "8b5288124840bee7b386bc413c487869b360b2b4ec421ea56425128692f2a82c"
dependencies = [
"aho-corasick",
"memchr",
@@ -6663,9 +6653,9 @@ dependencies = [
[[package]]
name = "regex-automata"
-version = "0.4.10"
+version = "0.4.11"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6b9458fa0bfeeac22b5ca447c63aaf45f28439a709ccd244698632f9aa6394d6"
+checksum = "833eb9ce86d40ef33cb1306d8accf7bc8ec2bfea4355cbdebb3df68b40925cad"
dependencies = [
"aho-corasick",
"memchr",
@@ -6842,7 +6832,7 @@ version = "0.6.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "41ab0892f4938752b34ae47cb53910b1b0921e55e77ddb6e44df666cab17939f"
dependencies = [
- "axum 0.8.4",
+ "axum 0.8.5",
"base64 0.22.1",
"bytes",
"chrono",
@@ -6859,7 +6849,7 @@ dependencies = [
"serde",
"serde_json",
"sse-stream",
- "thiserror 2.0.16",
+ "thiserror 2.0.17",
"tokio",
"tokio-stream",
"tokio-util",
@@ -6976,7 +6966,7 @@ dependencies = [
"errno",
"libc",
"linux-raw-sys",
- "windows-sys 0.61.0",
+ "windows-sys 0.61.1",
]
[[package]]
@@ -7004,7 +6994,7 @@ dependencies = [
"openssl-probe",
"rustls-pki-types",
"schannel",
- "security-framework 3.4.0",
+ "security-framework 3.5.1",
]
[[package]]
@@ -7041,7 +7031,7 @@ dependencies = [
"rustls-native-certs",
"rustls-platform-verifier-android",
"rustls-webpki",
- "security-framework 3.4.0",
+ "security-framework 3.5.1",
"security-framework-sys",
"webpki-root-certs",
"windows-sys 0.59.0",
@@ -7120,7 +7110,7 @@ version = "0.1.28"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "891d81b926048e76efe18581bf793546b4c0eaf8448d72be8de2bbee5fd166e1"
dependencies = [
- "windows-sys 0.61.0",
+ "windows-sys 0.61.1",
]
[[package]]
@@ -7221,9 +7211,9 @@ dependencies = [
[[package]]
name = "security-framework"
-version = "3.4.0"
+version = "3.5.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "60b369d18893388b345804dc0007963c99b7d665ae71d275812d828c6f089640"
+checksum = "b3297343eaf830f66ede390ea39da1d462b6b0c1b000f420d0a83f898bbbe6ef"
dependencies = [
"bitflags 2.9.4",
"core-foundation 0.10.1",
@@ -7249,7 +7239,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5b55fb86dfd3a2f5f76ea78310a88f96c4ea21a3031f8d212443d56123fd0521"
dependencies = [
"libc",
- "windows-sys 0.61.0",
+ "windows-sys 0.61.1",
]
[[package]]
@@ -7273,9 +7263,9 @@ dependencies = [
[[package]]
name = "serde"
-version = "1.0.225"
+version = "1.0.228"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "fd6c24dee235d0da097043389623fb913daddf92c76e9f5a1db88607a0bcbd1d"
+checksum = "9a8e94ea7f378bd32cbbd37198a4a91436180c5bb472411e48b5ec2e2124ae9e"
dependencies = [
"serde_core",
"serde_derive",
@@ -7317,18 +7307,18 @@ dependencies = [
[[package]]
name = "serde_core"
-version = "1.0.225"
+version = "1.0.228"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "659356f9a0cb1e529b24c01e43ad2bdf520ec4ceaf83047b83ddcc2251f96383"
+checksum = "41d385c7d4ca58e59fc732af25c3983b67ac852c1a25000afe1175de458b67ad"
dependencies = [
"serde_derive",
]
[[package]]
name = "serde_derive"
-version = "1.0.225"
+version = "1.0.228"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0ea936adf78b1f766949a4977b91d2f5595825bd6ec079aa9543ad2685fc4516"
+checksum = "d540f220d3187173da220f885ab66608367b6574e925011a9353e4badda91d79"
dependencies = [
"proc-macro2",
"quote",
@@ -7402,9 +7392,9 @@ dependencies = [
[[package]]
name = "serde_with"
-version = "3.14.0"
+version = "3.14.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f2c45cd61fefa9db6f254525d46e392b852e0e61d9a1fd36e5bd183450a556d5"
+checksum = "c522100790450cf78eeac1507263d0a350d4d5b30df0c8e1fe051a10c22b376e"
dependencies = [
"base64 0.22.1",
"chrono",
@@ -7422,11 +7412,11 @@ dependencies = [
[[package]]
name = "serde_with_macros"
-version = "3.14.0"
+version = "3.14.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "de90945e6565ce0d9a25098082ed4ee4002e047cb59892c318d66821e14bb30f"
+checksum = "327ada00f7d64abaac1e55a6911e90cf665aa051b9a561c7006c157f4633135e"
dependencies = [
- "darling 0.20.11",
+ "darling 0.21.3",
"proc-macro2",
"quote",
"syn 2.0.106",
@@ -7482,7 +7472,7 @@ dependencies = [
"argon2",
"async-channel",
"async_zip",
- "axum 0.8.4",
+ "axum 0.8.5",
"axum-server",
"bincode 2.0.1",
"blake3",
@@ -7490,9 +7480,11 @@ dependencies = [
"chrono",
"clap",
"compio",
+ "compio-io",
"compio-net",
"compio-quic",
"compio-tls",
+ "compio-ws",
"console-subscriber",
"crossbeam",
"ctrlc",
@@ -7541,16 +7533,17 @@ dependencies = [
"socket2 0.6.0",
"static-toml",
"strum",
- "sysinfo 0.37.0",
+ "sysinfo 0.37.1",
"tempfile",
"test-case",
- "thiserror 2.0.16",
+ "thiserror 2.0.17",
"toml 0.9.7",
"tower-http",
"tracing",
"tracing-appender",
"tracing-opentelemetry",
"tracing-subscriber",
+ "tungstenite",
"twox-hash",
"ulid",
"uuid",
@@ -7657,7 +7650,7 @@ checksum = "297f631f50729c8c99b84667867963997ec0b50f32b2a7dbcab828ef0541e8bb"
dependencies = [
"num-bigint",
"num-traits",
- "thiserror 2.0.16",
+ "thiserror 2.0.17",
"time",
]
@@ -7810,7 +7803,7 @@ dependencies = [
"serde_json",
"sha2",
"smallvec",
- "thiserror 2.0.16",
+ "thiserror 2.0.17",
"tokio",
"tokio-stream",
"tracing",
@@ -7895,7 +7888,7 @@ dependencies = [
"smallvec",
"sqlx-core",
"stringprep",
- "thiserror 2.0.16",
+ "thiserror 2.0.17",
"tracing",
"uuid",
"whoami",
@@ -7934,7 +7927,7 @@ dependencies = [
"smallvec",
"sqlx-core",
"stringprep",
- "thiserror 2.0.16",
+ "thiserror 2.0.17",
"tracing",
"uuid",
"whoami",
@@ -7960,7 +7953,7 @@ dependencies = [
"serde",
"serde_urlencoded",
"sqlx-core",
- "thiserror 2.0.16",
+ "thiserror 2.0.17",
"tracing",
"url",
"uuid",
@@ -8133,9 +8126,9 @@ dependencies = [
[[package]]
name = "sysinfo"
-version = "0.37.0"
+version = "0.37.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "07cec4dc2d2e357ca1e610cfb07de2fa7a10fc3e9fe89f72545f3d244ea87753"
+checksum = "3bddd368fda2f82ead69c03d46d351987cfa0c2a57abfa37a017f3aa3e9bf69a"
dependencies = [
"libc",
"memchr",
@@ -8159,15 +8152,15 @@ checksum = "55937e1799185b12863d447f42597ed69d9928686b8d88a1df17376a097d8369"
[[package]]
name = "tempfile"
-version = "3.22.0"
+version = "3.23.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "84fa4d11fadde498443cca10fd3ac23c951f0dc59e080e9f4b93d4df4e4eea53"
+checksum = "2d31c77bdf42a745371d260a26ca7163f1e0924b64afa0b688e61b5a9fa02f16"
dependencies = [
"fastrand",
"getrandom 0.3.3",
"once_cell",
"rustix",
- "windows-sys 0.61.0",
+ "windows-sys 0.61.1",
]
[[package]]
@@ -8241,11 +8234,11 @@ dependencies = [
[[package]]
name = "thiserror"
-version = "2.0.16"
+version = "2.0.17"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3467d614147380f2e4e374161426ff399c91084acd2363eaf549172b3d5e60c0"
+checksum = "f63587ca0f12b72a0600bcba1d40081f830876000bb46dd2337a3051618f4fc8"
dependencies = [
- "thiserror-impl 2.0.16",
+ "thiserror-impl 2.0.17",
]
[[package]]
@@ -8261,9 +8254,9 @@ dependencies = [
[[package]]
name = "thiserror-impl"
-version = "2.0.16"
+version = "2.0.17"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6c5e1be1c48b9172ee610da68fd9cd2770e7a4056cb3fc98710ee6906f0c7960"
+checksum = "3ff15c8ecd7de3849db632e14d18d2571fa09dfc5ed93479bc4485c7a517c913"
dependencies = [
"proc-macro2",
"quote",
@@ -8380,9 +8373,9 @@ dependencies = [
[[package]]
name = "tokio-rustls"
-version = "0.26.3"
+version = "0.26.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "05f63835928ca123f1bef57abbcd23bb2ba0ac9ae1235f1e65bda0d06e7786bd"
+checksum = "1729aa945f29d91ba541258c8df89027d5792d85a8841fb65e8bf0f4ede4ef61"
dependencies = [
"rustls",
"tokio",
@@ -8399,6 +8392,18 @@ dependencies = [
"tokio",
]
+[[package]]
+name = "tokio-tungstenite"
+version = "0.27.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "489a59b6730eda1b0171fcfda8b121f4bee2b35cba8645ca35c5f7ba3eb736c1"
+dependencies = [
+ "futures-util",
+ "log",
+ "tokio",
+ "tungstenite",
+]
+
[[package]]
name = "tokio-util"
version = "0.7.16"
@@ -8739,11 +8744,9 @@ dependencies = [
"once_cell",
"regex-automata",
"sharded-slab",
- "smallvec",
"thread_local",
"tracing",
"tracing-core",
- "tracing-log",
]
[[package]]
@@ -8763,6 +8766,23 @@ version = "0.2.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b"
+[[package]]
+name = "tungstenite"
+version = "0.27.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "eadc29d668c91fcc564941132e17b28a7ceb2f3ebf0b9dae3e03fd7a6748eb0d"
+dependencies = [
+ "bytes",
+ "data-encoding",
+ "http 1.3.1",
+ "httparse",
+ "log",
+ "rand 0.9.2",
+ "sha1",
+ "thiserror 2.0.17",
+ "utf-8",
+]
+
[[package]]
name = "twox-hash"
version = "2.1.2"
@@ -8932,6 +8952,12 @@ version = "2.1.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "daf8dba3b7eb870caf1ddeed7bc9d2a049f3cfdfae7cb521b087cc33ae4c49da"
+[[package]]
+name = "utf-8"
+version = "0.7.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "09cc8ee72d2a9becf2f2febe0205bbed8fc6615b7cb429ad062dc7b7ddd036a9"
+
[[package]]
name = "utf8-width"
version = "0.1.7"
@@ -9109,9 +9135,9 @@ checksum = "b8dad83b4f25e74f184f64c43b150b91efe7647395b42289f38e50566d82855b"
[[package]]
name = "wasm-bindgen"
-version = "0.2.103"
+version = "0.2.104"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ab10a69fbd0a177f5f649ad4d8d3305499c42bab9aef2f7ff592d0ec8f833819"
+checksum = "c1da10c01ae9f1ae40cbfac0bac3b1e724b320abfcf52229f80b547c0d250e2d"
dependencies = [
"cfg-if",
"once_cell",
@@ -9122,9 +9148,9 @@ dependencies = [
[[package]]
name = "wasm-bindgen-backend"
-version = "0.2.103"
+version = "0.2.104"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0bb702423545a6007bbc368fde243ba47ca275e549c8a28617f56f6ba53b1d1c"
+checksum = "671c9a5a66f49d8a47345ab942e2cb93c7d1d0339065d4f8139c486121b43b19"
dependencies = [
"bumpalo",
"log",
@@ -9136,9 +9162,9 @@ dependencies = [
[[package]]
name = "wasm-bindgen-futures"
-version = "0.4.53"
+version = "0.4.54"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a0b221ff421256839509adbb55998214a70d829d3a28c69b4a6672e9d2a42f67"
+checksum = "7e038d41e478cc73bae0ff9b36c60cff1c98b8f38f8d7e8061e79ee63608ac5c"
dependencies = [
"cfg-if",
"js-sys",
@@ -9149,9 +9175,9 @@ dependencies = [
[[package]]
name = "wasm-bindgen-macro"
-version = "0.2.103"
+version = "0.2.104"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "fc65f4f411d91494355917b605e1480033152658d71f722a90647f56a70c88a0"
+checksum = "7ca60477e4c59f5f2986c50191cd972e3a50d8a95603bc9434501cf156a9a119"
dependencies = [
"quote",
"wasm-bindgen-macro-support",
@@ -9159,9 +9185,9 @@ dependencies = [
[[package]]
name = "wasm-bindgen-macro-support"
-version = "0.2.103"
+version = "0.2.104"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ffc003a991398a8ee604a401e194b6b3a39677b3173d6e74495eb51b82e99a32"
+checksum = "9f07d2f20d4da7b26400c9f4a0511e6e0345b040694e8a75bd41d578fa4421d7"
dependencies = [
"proc-macro2",
"quote",
@@ -9172,9 +9198,9 @@ dependencies = [
[[package]]
name = "wasm-bindgen-shared"
-version = "0.2.103"
+version = "0.2.104"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "293c37f4efa430ca14db3721dfbe48d8c33308096bd44d80ebaa775ab71ba1cf"
+checksum = "bad67dc8b2a1a6e5448428adec4c3e84c43e561d8c9ee8a9e5aabeb193ec41d1"
dependencies = [
"unicode-ident",
]
@@ -9209,9 +9235,9 @@ dependencies = [
[[package]]
name = "web-sys"
-version = "0.3.80"
+version = "0.3.81"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "fbe734895e869dc429d78c4b433f8d17d95f8d05317440b4fad5ab2d33e596dc"
+checksum = "9367c417a924a74cae129e6a2ae3b47fabb1f8995595ab474029da749a8be120"
dependencies = [
"js-sys",
"wasm-bindgen",
@@ -9293,7 +9319,7 @@ version = "0.1.11"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22"
dependencies = [
- "windows-sys 0.61.0",
+ "windows-sys 0.48.0",
]
[[package]]
@@ -9352,8 +9378,8 @@ version = "0.61.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c0fdd3ddb90610c7638aa2b3a3ab2904fb9e5cdbecc643ddb3647212781c4ae3"
dependencies = [
- "windows-implement 0.60.0",
- "windows-interface 0.59.1",
+ "windows-implement 0.60.1",
+ "windows-interface 0.59.2",
"windows-link 0.1.3",
"windows-result 0.3.4",
"windows-strings 0.4.2",
@@ -9361,12 +9387,12 @@ dependencies = [
[[package]]
name = "windows-core"
-version = "0.62.0"
+version = "0.62.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "57fe7168f7de578d2d8a05b07fd61870d2e73b4020e9f49aa00da8471723497c"
+checksum = "6844ee5416b285084d3d3fffd743b925a6c9385455f64f6d4fa3031c4c2749a9"
dependencies = [
- "windows-implement 0.60.0",
- "windows-interface 0.59.1",
+ "windows-implement 0.60.1",
+ "windows-interface 0.59.2",
"windows-link 0.2.0",
"windows-result 0.4.0",
"windows-strings 0.5.0",
@@ -9396,9 +9422,9 @@ dependencies = [
[[package]]
name = "windows-implement"
-version = "0.60.0"
+version = "0.60.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a47fddd13af08290e67f4acabf4b459f647552718f683a7b415d290ac744a836"
+checksum = "edb307e42a74fb6de9bf3a02d9712678b22399c87e6fa869d6dfcd8c1b7754e0"
dependencies = [
"proc-macro2",
"quote",
@@ -9418,9 +9444,9 @@ dependencies = [
[[package]]
name = "windows-interface"
-version = "0.59.1"
+version = "0.59.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "bd9211b69f8dcdfa817bfd14bf1c97c9188afa36f4750130fcdf3f400eca9fa8"
+checksum = "c0abd1ddbc6964ac14db11c7213d6532ef34bd9aa042c2e5935f59d7908b46a5"
dependencies = [
"proc-macro2",
"quote",
@@ -9536,14 +9562,14 @@ version = "0.60.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f2f500e4d28234f72040990ec9d39e3a6b950f9f22d3dba18416c35882612bcb"
dependencies = [
- "windows-targets 0.53.3",
+ "windows-targets 0.53.4",
]
[[package]]
name = "windows-sys"
-version = "0.61.0"
+version = "0.61.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e201184e40b2ede64bc2ea34968b28e33622acdbbf37104f0e4a33f7abe657aa"
+checksum = "6f109e41dd4a3c848907eb83d5a42ea98b3769495597450cf6d153507b166f0f"
dependencies = [
"windows-link 0.2.0",
]
@@ -9596,11 +9622,11 @@ dependencies = [
[[package]]
name = "windows-targets"
-version = "0.53.3"
+version = "0.53.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d5fe6031c4041849d7c496a8ded650796e7b6ecc19df1a431c1a363342e5dc91"
+checksum = "2d42b7b7f66d2a06854650af09cfdf8713e427a439c97ad65a6375318033ac4b"
dependencies = [
- "windows-link 0.1.3",
+ "windows-link 0.2.0",
"windows_aarch64_gnullvm 0.53.0",
"windows_aarch64_msvc 0.53.0",
"windows_i686_gnu 0.53.0",
diff --git a/Cargo.toml b/Cargo.toml
index c37e8557a..247c2946a 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -157,7 +157,15 @@ cyper = { git = "https://github.com/krishvishal/cyper.git", rev = "cd75e266df6ab
cyper-axum = { git = "https://github.com/krishvishal/cyper", rev = "cd75e266df6ab0a9b9474eb7dda1735650d17db6" }
compio-net = { git = "https://github.com/compio-rs/compio.git", rev = "fe4243f0b6811ebc325afd081c9b087b4d9817be" }
compio-quic = { git = "https://github.com/compio-rs/compio.git", rev = "fe4243f0b6811ebc325afd081c9b087b4d9817be" }
-compio-tls = { git = "https://github.com/compio-rs/compio.git", rev = "fe4243f0b6811ebc325afd081c9b087b4d9817be", features = ["rustls"] }
+compio-tls = { git = "https://github.com/compio-rs/compio.git", rev = "fe4243f0b6811ebc325afd081c9b087b4d9817be", features = [
+ "rustls",
+] }
+compio-io = { git = "https://github.com/compio-rs/compio.git", rev = "fe4243f0b6811ebc325afd081c9b087b4d9817be" }
+compio-ws = { git = "https://github.com/krishvishal/compio-ws", branch = "main", features = [
+ "rustls",
+] }
+tungstenite = "0.27.0"
+tokio-tungstenite = "0.27.0"
tokio-rustls = "0.26.2"
tokio-util = { version = "0.7.16", features = ["compat"] }
toml = "0.9.5"
@@ -188,3 +196,14 @@ zip = "5.1.1"
[profile.release]
lto = true
codegen-units = 1
+
+[patch.crates-io]
+compio = { git = "https://github.com/compio-rs/compio.git", rev = "fe4243f0b6811ebc325afd081c9b087b4d9817be" }
+compio-driver = { git = "https://github.com/compio-rs/compio.git", rev = "fe4243f0b6811ebc325afd081c9b087b4d9817be" }
+compio-fs = { git = "https://github.com/compio-rs/compio.git", rev = "fe4243f0b6811ebc325afd081c9b087b4d9817be" }
+compio-io = { git = "https://github.com/compio-rs/compio.git", rev = "fe4243f0b6811ebc325afd081c9b087b4d9817be" }
+compio-net = { git = "https://github.com/compio-rs/compio.git", rev = "fe4243f0b6811ebc325afd081c9b087b4d9817be" }
+compio-runtime = { git = "https://github.com/compio-rs/compio.git", rev = "fe4243f0b6811ebc325afd081c9b087b4d9817be" }
+compio-tls = { git = "https://github.com/compio-rs/compio.git", rev = "fe4243f0b6811ebc325afd081c9b087b4d9817be" }
+compio-quic = { git = "https://github.com/compio-rs/compio.git", rev = "fe4243f0b6811ebc325afd081c9b087b4d9817be" }
+compio-buf = { git = "https://github.com/compio-rs/compio.git", rev = "fe4243f0b6811ebc325afd081c9b087b4d9817be" }
diff --git a/core/bench/report/src/types/transport.rs b/core/bench/report/src/types/transport.rs
index 73f036646..36bb5074d 100644
--- a/core/bench/report/src/types/transport.rs
+++ b/core/bench/report/src/types/transport.rs
@@ -31,4 +31,7 @@ pub enum BenchmarkTransport {
#[display("QUIC")]
#[serde(rename = "quic")]
Quic,
+ #[display("WEBSOCKET")]
+ #[serde(rename = "websocket")]
+ WebSocket,
}
diff --git a/core/bench/src/utils/client_factory.rs b/core/bench/src/utils/client_factory.rs
index 010d276a5..513d22300 100644
--- a/core/bench/src/utils/client_factory.rs
+++ b/core/bench/src/utils/client_factory.rs
@@ -23,6 +23,7 @@ use integration::http_client::HttpClientFactory;
use integration::quic_client::QuicClientFactory;
use integration::tcp_client::TcpClientFactory;
use integration::test_server::ClientFactory;
+use integration::websocket_client::WebSocketClientFactory;
use std::sync::Arc;
pub fn create_client_factory(args: &IggyBenchArgs) -> Arc<dyn ClientFactory> {
@@ -48,5 +49,8 @@ pub fn create_client_factory(args: &IggyBenchArgs) -> Arc<dyn ClientFactory> {
TransportProtocol::Quic => Arc::new(QuicClientFactory {
server_addr: args.server_address().to_owned(),
}),
+ TransportProtocol::WebSocket => Arc::new(WebSocketClientFactory {
+ server_addr: args.server_address().to_owned(),
+ }),
}
}
diff --git a/core/bench/src/utils/mod.rs b/core/bench/src/utils/mod.rs
index 8c07a9f1b..2559a6a3a 100644
--- a/core/bench/src/utils/mod.rs
+++ b/core/bench/src/utils/mod.rs
@@ -117,6 +117,7 @@ pub fn params_from_args_and_metrics(
TransportProtocol::Tcp => BenchmarkTransport::Tcp,
TransportProtocol::Quic => BenchmarkTransport::Quic,
TransportProtocol::Http => BenchmarkTransport::Http,
+ TransportProtocol::WebSocket => BenchmarkTransport::WebSocket,
};
let server_address = args.server_address().to_string();
let remark = args.remark();
diff --git a/core/bench/src/utils/server_starter.rs b/core/bench/src/utils/server_starter.rs
index 049e84924..1ee66d9b6 100644
--- a/core/bench/src/utils/server_starter.rs
+++ b/core/bench/src/utils/server_starter.rs
@@ -30,6 +30,7 @@ struct ServerConfig {
http: ConfigAddress,
tcp: ConfigAddress,
quic: ConfigAddress,
+ websocket: ConfigAddress,
}
#[derive(Debug, Deserialize)]
@@ -150,6 +151,23 @@ async fn evaluate_server_start_condition(args: &IggyBenchArgs) -> (bool, HashMap
envs,
)
}
+ TransportProtocol::WebSocket => {
+ let args_websocket_address = args.server_address().parse::<SocketAddr>().unwrap();
+ let config_websocket_address = default_config
+ .websocket
+ .address
+ .parse::<SocketAddr>()
+ .unwrap();
+ let envs = HashMap::from([(
+ "IGGY_WEBSOCKET_ADDRESS".to_owned(),
+ default_config.websocket.address.clone(),
+ )]);
+ (
+ addresses_are_equivalent(&args_websocket_address, &config_websocket_address)
+ && !is_tcp_addr_in_use(&args_websocket_address).await,
+ envs,
+ )
+ }
}
}
diff --git a/core/common/Cargo.toml b/core/common/Cargo.toml
index 48affb57d..c1d0e5e1e 100644
--- a/core/common/Cargo.toml
+++ b/core/common/Cargo.toml
@@ -40,6 +40,7 @@ bon = { workspace = true }
byte-unit = { workspace = true }
bytes = { workspace = true }
chrono = { workspace = true }
+compio-ws = { workspace = true }
clap = { workspace = true }
comfy-table = { workspace = true }
crc32fast = { workspace = true }
diff --git a/core/common/src/error/iggy_error.rs b/core/common/src/error/iggy_error.rs
index 92af86010..326c059ea 100644
--- a/core/common/src/error/iggy_error.rs
+++ b/core/common/src/error/iggy_error.rs
@@ -69,6 +69,8 @@ pub enum IggyError {
CannotOpenDatabase(String) = 19,
#[error("Resource with key: {0} was not found.")]
ResourceNotFound(String) = 20,
+ #[error("Cannot close WebSocket connection")]
+ CannotCloseWebSocketConnection(String) = 21,
#[error("Stale client")]
StaleClient = 30,
#[error("TCP error")]
@@ -175,6 +177,16 @@ pub enum IggyError {
CannotCreateEndpoint = 305,
#[error("Cannot parse URL")]
CannotParseUrl = 306,
+ #[error("WebSocket error")]
+ WebSocketError = 400,
+ #[error("WebSocket connection error")]
+ WebSocketConnectionError = 401,
+ #[error("WebSocket close error")]
+ WebSocketCloseError = 402,
+ #[error("WebSocket receive error")]
+ WebSocketReceiveError = 403,
+ #[error("WebSocket send error")]
+ WebSocketSendError = 404,
#[error("Cannot create streams directory, Path: {0}")]
CannotCreateStreamsDirectory(String) = 1000,
#[error("Cannot create stream with ID: {0} directory, Path: {1}")]
diff --git a/core/common/src/lib.rs b/core/common/src/lib.rs
index 29874c388..55c4c70f8 100644
--- a/core/common/src/lib.rs
+++ b/core/common/src/lib.rs
@@ -66,6 +66,10 @@ pub use types::configuration::tcp_config::tcp_client_config::*;
pub use types::configuration::tcp_config::tcp_client_config_builder::*;
pub use types::configuration::tcp_config::tcp_client_reconnection_config::*;
pub use types::configuration::tcp_config::tcp_connection_string_options::*;
+pub use types::configuration::websocket_config::websocket_client_config::*;
+pub use types::configuration::websocket_config::websocket_client_config_builder::*;
+pub use types::configuration::websocket_config::websocket_client_reconnection_config::*;
+pub use types::configuration::websocket_config::websocket_connection_string_options::*;
pub use types::configuration::transport::TransportProtocol;
pub use types::consumer::consumer_group::*;
pub use types::consumer::consumer_kind::*;
diff --git a/core/common/src/types/args/mod.rs b/core/common/src/types/args/mod.rs
index 3010faa50..220be6e16 100644
--- a/core/common/src/types/args/mod.rs
+++ b/core/common/src/types/args/mod.rs
@@ -198,6 +198,27 @@ pub struct ArgsOptional {
#[arg(long, default_missing_value(Some("true")), num_args(0..1))]
#[serde(skip_serializing_if = "Option::is_none")]
pub quic_validate_certificate: Option<bool>,
+
+ /// The optional server address for the WebSocket transport
+ ///
+ /// [default: 127.0.0.1:8095]
+ #[arg(long)]
+ #[serde(skip_serializing_if = "Option::is_none")]
+ pub websocket_server_address: Option<String>,
+
+ /// The optional number of max reconnect retries for the WebSocket transport
+ ///
+ /// [default: 3]
+ #[arg(long)]
+ #[serde(skip_serializing_if = "Option::is_none")]
+ pub websocket_reconnection_max_retries: Option<u32>,
+
+ /// The optional reconnect interval for the WebSocket transport
+ ///
+ /// [default: "1s"]
+ #[arg(long)]
+ #[serde(skip_serializing_if = "Option::is_none")]
+ pub websocket_reconnection_interval: Option<String>,
}
/// The arguments used by the `ClientProviderConfig` to create a client.
@@ -301,11 +322,30 @@ pub struct Args {
/// The optional heartbeat interval for the QUIC transport
pub quic_heartbeat_interval: String,
+
+ /// The optional server address for the WebSocket transport
+ pub websocket_server_address: String,
+
+ /// The optional number of maximum reconnect retries for the WebSocket transport
+ pub websocket_reconnection_enabled: bool,
+
+ /// The optional number of maximum reconnect retries for the WebSocket transport
+ pub websocket_reconnection_max_retries: Option<u32>,
+
+ /// The optional reconnect interval for the WebSocket transport
+ pub websocket_reconnection_interval: String,
+
+ /// The optional re-establish after last connection interval for WebSocket
+ pub websocket_reconnection_reestablish_after: String,
+
+ /// The optional heartbeat interval for the WebSocket transport
+ pub websocket_heartbeat_interval: String,
}
const QUIC_TRANSPORT: &str = "quic";
const HTTP_TRANSPORT: &str = "http";
const TCP_TRANSPORT: &str = "tcp";
+const WEBSOCKET_TRANSPORT: &str = "websocket";
impl Args {
pub fn get_server_address(&self) -> Option<String> {
@@ -318,6 +358,10 @@ impl Args {
.replace("localhost", "127.0.0.1"),
),
TCP_TRANSPORT => Some(self.tcp_server_address.replace("localhost", "127.0.0.1")),
+ WEBSOCKET_TRANSPORT => Some(
+ self.websocket_server_address
+ .replace("localhost", "127.0.0.1"),
+ ),
_ => None,
}
}
@@ -359,6 +403,12 @@ impl Default for Args {
quic_max_idle_timeout: 10000,
quic_validate_certificate: false,
quic_heartbeat_interval: "5s".to_string(),
+ websocket_server_address: "127.0.0.1:8095".to_string(),
+ websocket_reconnection_enabled: true,
+ websocket_reconnection_max_retries: None,
+ websocket_reconnection_interval: "1s".to_string(),
+ websocket_reconnection_reestablish_after: "5s".to_string(),
+ websocket_heartbeat_interval: "5s".to_string(),
}
}
}
@@ -447,6 +497,19 @@ impl From<Vec<ArgsOptional>> for Args {
if let Some(quic_validate_certificate) = optional_args.quic_validate_certificate {
args.quic_validate_certificate = quic_validate_certificate;
}
+ if let Some(websocket_server_address) = optional_args.websocket_server_address {
+ args.websocket_server_address = websocket_server_address;
+ }
+ if let Some(websocket_reconnection_retries) =
+ optional_args.websocket_reconnection_max_retries
+ {
+ args.websocket_reconnection_max_retries = Some(websocket_reconnection_retries);
+ }
+ if let Some(websocket_reconnection_interval) =
+ optional_args.websocket_reconnection_interval
+ {
+ args.websocket_reconnection_interval = websocket_reconnection_interval;
+ }
}
args
diff --git a/core/common/src/types/configuration/mod.rs b/core/common/src/types/configuration/mod.rs
index 75f03ed22..727cce158 100644
--- a/core/common/src/types/configuration/mod.rs
+++ b/core/common/src/types/configuration/mod.rs
@@ -19,4 +19,5 @@ pub(crate) mod auth_config;
pub(crate) mod http_config;
pub(crate) mod quic_config;
pub(crate) mod tcp_config;
+pub(crate) mod websocket_config;
pub(crate) mod transport;
diff --git a/core/common/src/types/configuration/transport.rs b/core/common/src/types/configuration/transport.rs
index 356295f13..0114fdc3a 100644
--- a/core/common/src/types/configuration/transport.rs
+++ b/core/common/src/types/configuration/transport.rs
@@ -32,6 +32,8 @@ pub enum TransportProtocol {
Quic = 2,
#[strum(to_string = "http")]
Http = 3,
+ #[strum(to_string = "websocket")]
+ WebSocket = 4,
}
impl TransportProtocol {
@@ -48,6 +50,7 @@ impl TryFrom<u8> for TransportProtocol {
v if v == TransportProtocol::Tcp as u8 => Ok(TransportProtocol::Tcp),
v if v == TransportProtocol::Quic as u8 => Ok(TransportProtocol::Quic),
v if v == TransportProtocol::Http as u8 => Ok(TransportProtocol::Http),
+ v if v == TransportProtocol::WebSocket as u8 => Ok(TransportProtocol::WebSocket),
_ => Err(IggyError::InvalidCommand),
}
}
@@ -98,6 +101,9 @@ impl<'de> Deserialize<'de> for TransportProtocol {
v if v == TransportProtocol::Tcp as u8 => Ok(TransportProtocol::Tcp),
v if v == TransportProtocol::Quic as u8 => Ok(TransportProtocol::Quic),
v if v == TransportProtocol::Http as u8 => Ok(TransportProtocol::Http),
+ v if v == TransportProtocol::WebSocket as u8 => {
+ Ok(TransportProtocol::WebSocket)
+ }
_ => Err(serde::de::Error::custom(format!(
"invalid transport protocol number: {}",
value
diff --git a/core/common/src/types/configuration/mod.rs b/core/common/src/types/configuration/websocket_config/mod.rs
similarity index 82%
copy from core/common/src/types/configuration/mod.rs
copy to core/common/src/types/configuration/websocket_config/mod.rs
index 75f03ed22..46de921e5 100644
--- a/core/common/src/types/configuration/mod.rs
+++ b/core/common/src/types/configuration/websocket_config/mod.rs
@@ -15,8 +15,7 @@
// specific language governing permissions and limitations
// under the License.
-pub(crate) mod auth_config;
-pub(crate) mod http_config;
-pub(crate) mod quic_config;
-pub(crate) mod tcp_config;
-pub(crate) mod transport;
+pub mod websocket_client_config;
+pub mod websocket_client_config_builder;
+pub mod websocket_client_reconnection_config;
+pub mod websocket_connection_string_options;
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
new file mode 100644
index 000000000..34cf20f53
--- /dev/null
+++ b/core/common/src/types/configuration/websocket_config/websocket_client_config.rs
@@ -0,0 +1,186 @@
+/* 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.
+ */
+
+use crate::types::configuration::auth_config::connection_string::ConnectionString;
+use crate::types::configuration::websocket_config::websocket_connection_string_options::WebSocketConnectionStringOptions;
+use crate::{AutoLogin, IggyDuration, WebSocketClientReconnectionConfig};
+use compio_ws::WebSocketConfig as TungsteniteConfig;
+use std::fmt::{Display, Formatter};
+use std::str::FromStr;
+
+/// Configuration for the WebSocket client.
+#[derive(Debug, Clone)]
+pub struct WebSocketClientConfig {
+ /// The address of the Iggy server.
+ pub server_address: String,
+ /// Whether to automatically login user after establishing connection.
+ pub auto_login: AutoLogin,
+ /// Whether to automatically reconnect when disconnected.
+ pub reconnection: WebSocketClientReconnectionConfig,
+ /// Interval of heartbeats sent by the client
+ pub heartbeat_interval: IggyDuration,
+ /// WebSocket-specific configuration.
+ pub ws_config: WebSocketConfig,
+}
+
+/// WebSocket-specific configuration that maps to tungstenite options.
+/// Uses same structure as server-side WebSocketConfig but for client.
+#[derive(Debug, Clone)]
+pub struct WebSocketConfig {
+ /// Read buffer size.
+ pub read_buffer_size: Option<usize>,
+ /// Write buffer size.
+ pub write_buffer_size: Option<usize>,
+ /// Maximum write buffer size.
+ pub max_write_buffer_size: Option<usize>,
+ /// Maximum message size.
+ pub max_message_size: Option<usize>,
+ /// Maximum frame size.
+ pub max_frame_size: Option<usize>,
+ /// Accept unmasked frames (client should typically keep as false).
+ pub accept_unmasked_frames: bool,
+}
+
+impl Default for WebSocketClientConfig {
+ fn default() -> Self {
+ WebSocketClientConfig {
+ server_address: "127.0.0.1:8080".to_string(),
+ auto_login: AutoLogin::Disabled,
+ reconnection: WebSocketClientReconnectionConfig::default(),
+ heartbeat_interval: IggyDuration::from_str("5s").unwrap(),
+ ws_config: WebSocketConfig::default(),
+ }
+ }
+}
+
+impl Default for WebSocketConfig {
+ fn default() -> Self {
+ // Use tungstenite defaults
+ let tungstenite_config = TungsteniteConfig::default();
+ WebSocketConfig {
+ read_buffer_size: Some(tungstenite_config.read_buffer_size),
+ write_buffer_size: Some(tungstenite_config.write_buffer_size),
+ max_write_buffer_size: Some(tungstenite_config.max_write_buffer_size),
+ max_message_size: tungstenite_config.max_message_size,
+ max_frame_size: tungstenite_config.max_frame_size,
+ accept_unmasked_frames: false,
+ }
+ }
+}
+
+impl WebSocketConfig {
+ /// Convert to tungstenite WebSocketConfig so we can use tungstenite defaults
+ pub fn to_tungstenite_config(&self) -> Option<TungsteniteConfig> {
+ if self.read_buffer_size.is_none()
+ && self.write_buffer_size.is_none()
+ && self.max_write_buffer_size.is_none()
+ && self.max_message_size.is_none()
+ && self.max_frame_size.is_none()
+ && !self.accept_unmasked_frames
+ {
+ return None;
+ }
+
+ let mut config = TungsteniteConfig::default();
+
+ if let Some(read_buf_size) = self.read_buffer_size {
+ config = config.read_buffer_size(read_buf_size);
+ }
+
+ if let Some(write_buf_size) = self.write_buffer_size {
+ config = config.write_buffer_size(write_buf_size);
+ }
+
+ if let Some(max_write_buf_size) = self.max_write_buffer_size {
+ 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));
+ }
+
+ config = config.accept_unmasked_frames(self.accept_unmasked_frames);
+
+ Some(config)
+ }
+}
+
+impl From<ConnectionString<WebSocketConnectionStringOptions>> for WebSocketClientConfig {
+ fn from(connection_string: ConnectionString<WebSocketConnectionStringOptions>) -> Self {
+ let options = connection_string.options();
+
+ let mut ws_config = WebSocketConfig::default();
+
+ // Apply connection string options to WebSocket config
+ if let Some(read_buf_size) = options.read_buffer_size() {
+ ws_config.read_buffer_size = Some(read_buf_size);
+ }
+ if let Some(write_buf_size) = options.write_buffer_size() {
+ ws_config.write_buffer_size = Some(write_buf_size);
+ }
+ if let Some(max_write_buf_size) = options.max_write_buffer_size() {
+ ws_config.max_write_buffer_size = Some(max_write_buf_size);
+ }
+ if let Some(max_msg_size) = options.max_message_size() {
+ ws_config.max_message_size = Some(max_msg_size);
+ }
+ if let Some(max_frame_size) = options.max_frame_size() {
+ ws_config.max_frame_size = Some(max_frame_size);
+ }
+ if let Some(accept_unmasked) = options.accept_unmasked_frames() {
+ ws_config.accept_unmasked_frames = accept_unmasked;
+ }
+
+ WebSocketClientConfig {
+ server_address: connection_string.server_address().into(),
+ auto_login: connection_string.auto_login().to_owned(),
+ reconnection: connection_string.options().reconnection().to_owned(),
+ heartbeat_interval: connection_string.options().heartbeat_interval(),
+ ws_config,
+ }
+ }
+}
+
+impl Display for WebSocketClientConfig {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ write!(
+ f,
+ "{{ server_address: {}, reconnection: {}, heartbeat_interval: {}, ws_config: {} }}",
+ self.server_address, self.reconnection, self.heartbeat_interval, self.ws_config
+ )
+ }
+}
+
+impl Display for WebSocketConfig {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ write!(
+ f,
+ "{{ read_buffer_size: {:?}, write_buffer_size: {:?}, max_write_buffer_size: {:?}, max_message_size: {:?}, max_frame_size: {:?}, accept_unmasked_frames: {} }}",
+ self.read_buffer_size,
+ self.write_buffer_size,
+ self.max_write_buffer_size,
+ self.max_message_size,
+ self.max_frame_size,
+ self.accept_unmasked_frames
+ )
+ }
+}
diff --git a/core/common/src/types/configuration/websocket_config/websocket_client_config_builder.rs b/core/common/src/types/configuration/websocket_config/websocket_client_config_builder.rs
new file mode 100644
index 000000000..ca581829b
--- /dev/null
+++ b/core/common/src/types/configuration/websocket_config/websocket_client_config_builder.rs
@@ -0,0 +1,134 @@
+/* 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.
+ */
+
+use crate::{AutoLogin, IggyDuration, IggyError, WebSocketClientConfig};
+use std::net::SocketAddr;
+
+/// Builder for the WebSocket client configuration.
+/// Allows configuring the WebSocket client with custom settings or using defaults:
+/// - `server_address`: Default is "127.0.0.1:8080"
+/// - `auto_login`: Default is AutoLogin::Disabled.
+/// - `reconnection`: Default is enabled unlimited retries and 1 second interval.
+/// - `heartbeat_interval`: Default is 5 seconds.
+/// - WebSocket-specific options with sensible defaults from tungstenite.
+#[derive(Debug, Default)]
+pub struct WebSocketClientConfigBuilder {
+ config: WebSocketClientConfig,
+}
+
+impl WebSocketClientConfigBuilder {
+ pub fn new() -> Self {
+ WebSocketClientConfigBuilder::default()
+ }
+
+ /// Sets the server address for the WebSocket client.
+ pub fn with_server_address(mut self, server_address: String) -> Self {
+ self.config.server_address = server_address;
+ self
+ }
+
+ /// Sets the auto sign in during connection.
+ pub fn with_auto_sign_in(mut self, auto_sign_in: AutoLogin) -> Self {
+ self.config.auto_login = auto_sign_in;
+ self
+ }
+
+ pub fn with_enabled_reconnection(mut self) -> Self {
+ self.config.reconnection.enabled = true;
+ self
+ }
+
+ /// Sets the number of retries when connecting to the server.
+ pub fn with_reconnection_max_retries(mut self, max_retries: Option<u32>) -> Self {
+ self.config.reconnection.max_retries = max_retries;
+ self
+ }
+
+ /// Sets the interval between retries when connecting to the server.
+ pub fn with_reconnection_interval(mut self, interval: IggyDuration) -> Self {
+ self.config.reconnection.interval = interval;
+ self
+ }
+
+ /// Sets the time to wait before attempting to reestablish connection.
+ pub fn with_reestablish_after(mut self, reestablish_after: IggyDuration) -> Self {
+ self.config.reconnection.reestablish_after = reestablish_after;
+ self
+ }
+
+ /// Sets the heartbeat interval.
+ pub fn with_heartbeat_interval(mut self, heartbeat_interval: IggyDuration) -> Self {
+ self.config.heartbeat_interval = heartbeat_interval;
+ self
+ }
+
+ /// Sets the read buffer size.
+ pub fn with_read_buffer_size(mut self, size: usize) -> Self {
+ self.config.ws_config.read_buffer_size = Some(size);
+ self
+ }
+
+ /// Sets the write buffer size.
+ pub fn with_write_buffer_size(mut self, size: usize) -> Self {
+ self.config.ws_config.write_buffer_size = Some(size);
+ self
+ }
+
+ /// Sets the maximum write buffer size.
+ pub fn with_max_write_buffer_size(mut self, size: usize) -> Self {
+ self.config.ws_config.max_write_buffer_size = Some(size);
+ self
+ }
+
+ /// Sets the maximum message size.
+ pub fn with_max_message_size(mut self, size: usize) -> Self {
+ self.config.ws_config.max_message_size = Some(size);
+ self
+ }
+
+ /// Sets the maximum frame size.
+ pub fn with_max_frame_size(mut self, size: usize) -> Self {
+ self.config.ws_config.max_frame_size = Some(size);
+ self
+ }
+
+ /// Sets whether to accept unmasked frames.
+ /// Note: Clients should typically keep this as false for RFC compliance.
+ pub fn with_accept_unmasked_frames(mut self, accept: bool) -> Self {
+ self.config.ws_config.accept_unmasked_frames = accept;
+ self
+ }
+
+ /// Builds the WebSocket client configuration.
+ pub fn build(self) -> Result<WebSocketClientConfig, IggyError> {
+ let addr = self.config.server_address.trim();
+
+ // Check if it's a valid socket address or host:port format
+ if addr.parse::<SocketAddr>().is_err() {
+ let (host, port) = addr.rsplit_once(':').unwrap_or((addr, ""));
+ if port.is_empty() || port.parse::<u16>().is_err() {
+ return Err(IggyError::InvalidIpAddress(
+ host.to_owned(),
+ port.to_owned(),
+ ));
+ }
+ }
+
+ Ok(self.config)
+ }
+}
diff --git a/core/common/src/types/configuration/websocket_config/websocket_client_reconnection_config.rs b/core/common/src/types/configuration/websocket_config/websocket_client_reconnection_config.rs
new file mode 100644
index 000000000..c7a86aa90
--- /dev/null
+++ b/core/common/src/types/configuration/websocket_config/websocket_client_reconnection_config.rs
@@ -0,0 +1,61 @@
+/* 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.
+ */
+
+use crate::IggyDuration;
+use serde::{Deserialize, Serialize};
+use std::fmt::{Display, Formatter};
+use std::str::FromStr;
+
+/// WebSocket client reconnection configuration.
+/// It consists of the following fields:
+/// - `enabled`: whether to enable reconnection.
+/// - `max_retries`: the maximum number of retries. If None, will retry infinitely.
+/// - `interval`: the interval between retries.
+/// - `reestablish_after`: the time to wait before attempting to reestablish connection.
+#[derive(Debug, Clone, Deserialize, Serialize)]
+pub struct WebSocketClientReconnectionConfig {
+ /// Whether to enable reconnection.
+ pub enabled: bool,
+ /// The maximum number of retries. If None, will retry infinitely.
+ pub max_retries: Option<u32>,
+ /// The interval between retries.
+ pub interval: IggyDuration,
+ /// The time to wait before attempting to reestablish connection.
+ pub reestablish_after: IggyDuration,
+}
+
+impl Default for WebSocketClientReconnectionConfig {
+ fn default() -> Self {
+ WebSocketClientReconnectionConfig {
+ enabled: true,
+ max_retries: None,
+ interval: IggyDuration::from_str("1s").unwrap(),
+ reestablish_after: IggyDuration::from_str("5s").unwrap(),
+ }
+ }
+}
+
+impl Display for WebSocketClientReconnectionConfig {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ write!(
+ f,
+ "{{ enabled: {}, max_retries: {:?}, interval: {}, reestablish_after: {} }}",
+ self.enabled, self.max_retries, self.interval, self.reestablish_after
+ )
+ }
+}
diff --git a/core/common/src/types/configuration/websocket_config/websocket_connection_string_options.rs b/core/common/src/types/configuration/websocket_config/websocket_connection_string_options.rs
new file mode 100644
index 000000000..60da518ea
--- /dev/null
+++ b/core/common/src/types/configuration/websocket_config/websocket_connection_string_options.rs
@@ -0,0 +1,209 @@
+// 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.
+
+use crate::{ConnectionStringOptions, IggyDuration, IggyError, WebSocketClientReconnectionConfig};
+use std::str::FromStr;
+
+#[derive(Debug, Clone)]
+pub struct WebSocketConnectionStringOptions {
+ heartbeat_interval: IggyDuration,
+ reconnection: WebSocketClientReconnectionConfig,
+
+ read_buffer_size: Option<usize>,
+ write_buffer_size: Option<usize>,
+ max_write_buffer_size: Option<usize>,
+ max_message_size: Option<usize>,
+ max_frame_size: Option<usize>,
+
+ accept_unmasked_frames: Option<bool>,
+
+ tls_enabled: bool,
+ tls_domain: String,
+ tls_ca_file: Option<String>,
+ tls_validate_certificate: bool,
+}
+
+impl WebSocketConnectionStringOptions {
+ pub fn heartbeat_interval(&self) -> IggyDuration {
+ self.heartbeat_interval
+ }
+
+ pub fn reconnection(&self) -> &WebSocketClientReconnectionConfig {
+ &self.reconnection
+ }
+
+ pub fn read_buffer_size(&self) -> Option<usize> {
+ self.read_buffer_size
+ }
+
+ pub fn write_buffer_size(&self) -> Option<usize> {
+ self.write_buffer_size
+ }
+
+ pub fn max_write_buffer_size(&self) -> Option<usize> {
+ self.max_write_buffer_size
+ }
+
+ pub fn max_message_size(&self) -> Option<usize> {
+ self.max_message_size
+ }
+
+ pub fn max_frame_size(&self) -> Option<usize> {
+ self.max_frame_size
+ }
+
+ pub fn accept_unmasked_frames(&self) -> Option<bool> {
+ self.accept_unmasked_frames
+ }
+
+ pub fn tls_enabled(&self) -> bool {
+ self.tls_enabled
+ }
+
+ pub fn tls_domain(&self) -> &str {
+ &self.tls_domain
+ }
+
+ pub fn tls_ca_file(&self) -> &Option<String> {
+ &self.tls_ca_file
+ }
+
+ pub fn tls_validate_certificate(&self) -> bool {
+ self.tls_validate_certificate
+ }
+}
+
+impl ConnectionStringOptions for WebSocketConnectionStringOptions {
+ fn retries(&self) -> Option<u32> {
+ self.reconnection.max_retries
+ }
+
+ fn heartbeat_interval(&self) -> IggyDuration {
+ self.heartbeat_interval
+ }
+
+ fn parse_options(options_str: &str) -> Result<Self, IggyError> {
+ let mut parsed_options = WebSocketConnectionStringOptions::default();
+
+ if options_str.is_empty() {
+ return Ok(parsed_options);
+ }
+
+ for option in options_str.split('&') {
+ let parts: Vec<&str> = option.split('=').collect();
+ if parts.len() != 2 {
+ return Err(IggyError::InvalidConnectionString);
+ }
+
+ match parts[0] {
+ "heartbeat_interval" => {
+ parsed_options.heartbeat_interval = IggyDuration::from_str(parts[1])
+ .map_err(|_| IggyError::InvalidConnectionString)?;
+ }
+ "reconnection_retries" => {
+ let retries = match parts[1] {
+ "unlimited" => None,
+ val => Some(
+ val.parse::<u32>()
+ .map_err(|_| IggyError::InvalidConnectionString)?,
+ ),
+ };
+ parsed_options.reconnection.max_retries = retries;
+ }
+ "reconnection_interval" => {
+ parsed_options.reconnection.interval = IggyDuration::from_str(parts[1])
+ .map_err(|_| IggyError::InvalidConnectionString)?;
+ }
+ "reestablish_after" => {
+ parsed_options.reconnection.reestablish_after =
+ IggyDuration::from_str(parts[1])
+ .map_err(|_| IggyError::InvalidConnectionString)?;
+ }
+
+ "read_buffer_size" => {
+ let size = parts[1]
+ .parse::<usize>()
+ .map_err(|_| IggyError::InvalidConnectionString)?;
+ parsed_options.read_buffer_size = Some(size);
+ }
+ "write_buffer_size" => {
+ let size = parts[1]
+ .parse::<usize>()
+ .map_err(|_| IggyError::InvalidConnectionString)?;
+ parsed_options.write_buffer_size = Some(size);
+ }
+ "max_write_buffer_size" => {
+ let size = parts[1]
+ .parse::<usize>()
+ .map_err(|_| IggyError::InvalidConnectionString)?;
+ parsed_options.max_write_buffer_size = Some(size);
+ }
+ "max_message_size" => {
+ let size = parts[1]
+ .parse::<usize>()
+ .map_err(|_| IggyError::InvalidConnectionString)?;
+ parsed_options.max_message_size = Some(size);
+ }
+ "max_frame_size" => {
+ let size = parts[1]
+ .parse::<usize>()
+ .map_err(|_| IggyError::InvalidConnectionString)?;
+ parsed_options.max_frame_size = Some(size);
+ }
+
+ "accept_unmasked_frames" => {
+ parsed_options.accept_unmasked_frames = Some(parts[1] == "true");
+ }
+
+ "tls" => {
+ parsed_options.tls_enabled = parts[1] == "true";
+ }
+ "tls_domain" => {
+ parsed_options.tls_domain = parts[1].to_string();
+ }
+ "tls_ca_file" => {
+ parsed_options.tls_ca_file = Some(parts[1].to_string());
+ }
+ "tls_validate_certificate" => {
+ parsed_options.tls_validate_certificate = parts[1] == "true";
+ }
+
+ _ => return Err(IggyError::InvalidConnectionString),
+ }
+ }
+ Ok(parsed_options)
+ }
+}
+
+impl Default for WebSocketConnectionStringOptions {
+ fn default() -> Self {
+ WebSocketConnectionStringOptions {
+ heartbeat_interval: IggyDuration::from_str("5s").unwrap(),
+ reconnection: WebSocketClientReconnectionConfig::default(),
+ read_buffer_size: None,
+ write_buffer_size: None,
+ max_write_buffer_size: None,
+ max_message_size: None,
+ max_frame_size: None,
+ accept_unmasked_frames: None,
+ tls_enabled: false,
+ tls_domain: "localhost".to_string(),
+ tls_ca_file: None,
+ tls_validate_certificate: true,
+ }
+ }
+}
diff --git a/core/configs/server.toml b/core/configs/server.toml
index 3d25b0cef..2de2dd11f 100644
--- a/core/configs/server.toml
+++ b/core/configs/server.toml
@@ -600,3 +600,8 @@ address = "127.0.0.1:8091"
# - numeric value (e.g. 4): Use 4 shards (4 threads pinned to cores 0, 1, 2, 3)
# - range (e.g. "5..8"): Use 3 shards with affinity to cores 5, 6, 7
cpu_allocation = "all"
+
+[websocket]
+
+enabled = true
+address = "0.0.0.0:8092"
diff --git a/core/integration/src/lib.rs b/core/integration/src/lib.rs
index 88c51ac10..699229f22 100644
--- a/core/integration/src/lib.rs
+++ b/core/integration/src/lib.rs
@@ -24,6 +24,9 @@ pub mod http_client;
pub mod quic_client;
#[allow(deprecated)]
pub mod tcp_client;
+#[allow(deprecated)]
+pub mod websocket_client;
+
#[allow(deprecated)]
pub mod test_mcp_server;
#[allow(deprecated)]
diff --git a/core/integration/src/test_server.rs b/core/integration/src/test_server.rs
index a16b5ce4f..a3ac828db 100644
--- a/core/integration/src/test_server.rs
+++ b/core/integration/src/test_server.rs
@@ -59,7 +59,6 @@ pub trait ClientFactory: Sync + Send {
fn transport(&self) -> TransportProtocol;
fn server_addr(&self) -> String;
}
-
#[derive(Display, Debug)]
enum ServerProtocolAddr {
#[display("RAW_TCP:{_0}")]
@@ -70,6 +69,9 @@ enum ServerProtocolAddr {
#[display("QUIC_UDP:{_0}")]
QuicUdp(SocketAddr),
+
+ #[display("WEBSOCKET:{_0}")]
+ WebSocket(SocketAddr),
}
#[derive(Debug)]
@@ -323,6 +325,9 @@ impl TestServer {
ServerProtocolAddr::QuicUdp(addr) => {
("IGGY_QUIC_ADDRESS".to_string(), addr.to_string())
}
+ ServerProtocolAddr::WebSocket(addr) => {
+ ("IGGY_WEBSOCKET_ADDRESS".to_string(), addr.to_string())
+ }
};
self.envs.entry(key.0).or_insert(key.1);
@@ -405,11 +410,18 @@ impl TestServer {
panic!("Http address port is 0!");
}
+ let websocket_addr: SocketAddr = config.websocket.address.parse().unwrap();
+ if websocket_addr.port() == 0 {
+ panic!("WebSocket address port is 0!");
+ }
+
self.server_addrs
.push(ServerProtocolAddr::QuicUdp(quic_addr));
self.server_addrs.push(ServerProtocolAddr::RawTcp(tcp_addr));
self.server_addrs
.push(ServerProtocolAddr::HttpTcp(http_addr));
+ self.server_addrs
+ .push(ServerProtocolAddr::WebSocket(websocket_addr));
} else {
panic!(
"Failed to load config from file {config_path} in {MAX_PORT_WAIT_DURATION_S} s!"
@@ -488,6 +500,15 @@ impl TestServer {
None
}
}
+
+ pub fn get_websocket_addr(&self) -> Option<String> {
+ for server_protocol_addr in &self.server_addrs {
+ if let ServerProtocolAddr::WebSocket(a) = server_protocol_addr {
+ return Some(a.to_string());
+ }
+ }
+ None
+ }
}
impl Drop for TestServer {
diff --git a/core/integration/src/websocket_client.rs b/core/integration/src/websocket_client.rs
new file mode 100644
index 000000000..cc0cc0826
--- /dev/null
+++ b/core/integration/src/websocket_client.rs
@@ -0,0 +1,53 @@
+/* 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.
+ */
+
+use crate::test_server::ClientFactory;
+use async_trait::async_trait;
+use iggy::prelude::{Client, ClientWrapper, WebSocketClientConfig};
+use iggy::websocket::websocket_client::WebSocketClient;
+use iggy_common::TransportProtocol;
+use std::sync::Arc;
+
+#[derive(Debug, Clone)]
+pub struct WebSocketClientFactory {
+ pub server_addr: String,
+}
+
+#[async_trait]
+impl ClientFactory for WebSocketClientFactory {
+ async fn create_client(&self) -> ClientWrapper {
+ let config = WebSocketClientConfig {
+ server_address: self.server_addr.clone(),
+ ..WebSocketClientConfig::default()
+ };
+ let client = WebSocketClient::create(Arc::new(config)).unwrap();
+ Client::connect(&client).await.unwrap();
+ ClientWrapper::WebSocket(client)
+ }
+
+ fn transport(&self) -> TransportProtocol {
+ TransportProtocol::WebSocket
+ }
+
+ fn server_addr(&self) -> String {
+ self.server_addr.clone()
+ }
+}
+
+unsafe impl Send for WebSocketClientFactory {}
+unsafe impl Sync for WebSocketClientFactory {}
diff --git a/core/integration/tests/cli/system/test_me_command.rs b/core/integration/tests/cli/system/test_me_command.rs
index 9730217cc..19dfeb628 100644
--- a/core/integration/tests/cli/system/test_me_command.rs
+++ b/core/integration/tests/cli/system/test_me_command.rs
@@ -44,6 +44,7 @@ impl TransportProtocolExt for TransportProtocol {
match self {
TransportProtocol::Tcp => vec!["--transport", "tcp"],
TransportProtocol::Quic => vec!["--transport", "quic"],
+ TransportProtocol::WebSocket => vec!["--transport", "websocket"],
// Note: HTTP is not supported for the 'me' command
TransportProtocol::Http => {
panic!("HTTP transport is not supported for the 'me' command")
@@ -117,6 +118,10 @@ impl IggyCmdTestCase for TestMeCmd {
"--quic-server-address".into(),
server.get_quic_udp_addr().unwrap(),
],
+ TransportProtocol::WebSocket => vec![
+ "--websocket-server-address".into(),
+ server.get_websocket_addr().unwrap(),
+ ],
TransportProtocol::Http => {
panic!("HTTP transport is not supported for the 'me' command")
}
diff --git a/core/integration/tests/server/cg.rs b/core/integration/tests/server/cg.rs
index 529cfefeb..28cef8b24 100644
--- a/core/integration/tests/server/cg.rs
+++ b/core/integration/tests/server/cg.rs
@@ -24,7 +24,7 @@ use test_case::test_matrix;
// Consumer group scenarios do not support HTTP
#[test_matrix(
- [TransportProtocol::Tcp, TransportProtocol::Quic],
+ [TransportProtocol::Tcp, TransportProtocol::Quic, TransportProtocol::WebSocket],
[
join_scenario(),
single_client_scenario(),
diff --git a/core/integration/tests/server/general.rs b/core/integration/tests/server/general.rs
index e8a0b5ffa..5f67c5c06 100644
--- a/core/integration/tests/server/general.rs
+++ b/core/integration/tests/server/general.rs
@@ -25,7 +25,7 @@ use test_case::test_matrix;
// TODO: Include other trasnsport protocols
#[test_matrix(
- [TransportProtocol::Http, TransportProtocol::Tcp],
+ [TransportProtocol::Http, TransportProtocol::Tcp, TransportProtocol::WebSocket],
[
system_scenario(),
user_scenario(),
diff --git a/core/integration/tests/server/mod.rs b/core/integration/tests/server/mod.rs
index fe7911da6..335d2d269 100644
--- a/core/integration/tests/server/mod.rs
+++ b/core/integration/tests/server/mod.rs
@@ -27,6 +27,7 @@ use integration::{
quic_client::QuicClientFactory,
tcp_client::TcpClientFactory,
test_server::{ClientFactory, TestServer},
+ websocket_client::WebSocketClientFactory,
};
use scenarios::{
bench_scenario, consumer_group_join_scenario,
@@ -95,6 +96,10 @@ async fn run_scenario(transport: TransportProtocol, scenario: ScenarioFn) {
let server_addr = test_server.get_http_api_addr().unwrap();
Box::new(HttpClientFactory { server_addr })
}
+ TransportProtocol::WebSocket => {
+ let server_addr = test_server.get_websocket_addr().unwrap();
+ Box::new(WebSocketClientFactory { server_addr })
+ }
};
scenario(&*client_factory).await;
diff --git a/core/sdk/Cargo.toml b/core/sdk/Cargo.toml
index c6b267298..8dbb97afc 100644
--- a/core/sdk/Cargo.toml
+++ b/core/sdk/Cargo.toml
@@ -56,6 +56,8 @@ reqwest-retry = { workspace = true }
rustls = { workspace = true }
serde = { workspace = true }
tokio = { workspace = true }
+tungstenite = { workspace = true }
+tokio-tungstenite = { workspace = true }
tokio-rustls = { workspace = true }
tracing = { workspace = true }
trait-variant = { workspace = true }
diff --git a/core/sdk/src/client_provider.rs b/core/sdk/src/client_provider.rs
index a91ab9b53..eca6dd13d 100644
--- a/core/sdk/src/client_provider.rs
+++ b/core/sdk/src/client_provider.rs
@@ -22,12 +22,15 @@ use crate::clients::client::IggyClient;
use crate::http::http_client::HttpClient;
use crate::prelude::{
ClientError, HttpClientConfig, IggyDuration, QuicClientConfig, QuicClientReconnectionConfig,
- TcpClientConfig, TcpClientReconnectionConfig,
+ TcpClientConfig, TcpClientReconnectionConfig, WebSocketClient,
};
use crate::quic::quic_client::QuicClient;
use crate::tcp::tcp_client::TcpClient;
use iggy_binary_protocol::Client;
-use iggy_common::{AutoLogin, Credentials, TransportProtocol};
+use iggy_common::{
+ AutoLogin, Credentials, TransportProtocol, WebSocketClientConfig,
+ WebSocketClientReconnectionConfig, WebSocketConfig,
+};
use std::str::FromStr;
use std::sync::Arc;
@@ -47,6 +50,8 @@ pub struct ClientProviderConfig {
pub quic: Option<Arc<QuicClientConfig>>,
/// The optional configuration for the TCP transport.
pub tcp: Option<Arc<TcpClientConfig>>,
+ /// The optional configuration for the WebSocket transport.
+ pub websocket: Option<Arc<WebSocketClientConfig>>,
}
impl Default for ClientProviderConfig {
@@ -56,6 +61,7 @@ impl Default for ClientProviderConfig {
http: Some(Arc::new(HttpClientConfig::default())),
quic: Some(Arc::new(QuicClientConfig::default())),
tcp: Some(Arc::new(TcpClientConfig::default())),
+ websocket: Some(Arc::new(WebSocketClientConfig::default())),
}
}
}
@@ -79,6 +85,7 @@ impl ClientProviderConfig {
http: None,
quic: None,
tcp: None,
+ websocket: None,
};
match config.transport {
TransportProtocol::Quic => {
@@ -151,6 +158,32 @@ impl ClientProviderConfig {
},
}));
}
+ TransportProtocol::WebSocket => {
+ config.websocket = Some(Arc::new(WebSocketClientConfig {
+ server_address: args.websocket_server_address,
+ heartbeat_interval: IggyDuration::from_str(&args.websocket_heartbeat_interval)
+ .unwrap(),
+ reconnection: WebSocketClientReconnectionConfig {
+ enabled: args.websocket_reconnection_enabled,
+ max_retries: args.websocket_reconnection_max_retries,
+ interval: IggyDuration::from_str(&args.websocket_reconnection_interval)
+ .unwrap(),
+ reestablish_after: IggyDuration::from_str(
+ &args.websocket_reconnection_reestablish_after,
+ )
+ .unwrap(),
+ },
+ auto_login: if auto_login {
+ AutoLogin::Enabled(Credentials::UsernamePassword(
+ args.username,
+ args.password,
+ ))
+ } else {
+ AutoLogin::Disabled
+ },
+ ws_config: WebSocketConfig::default(),
+ }));
+ }
}
Ok(config)
@@ -202,5 +235,13 @@ pub async fn get_raw_client(
};
Ok(ClientWrapper::Tcp(client))
}
+ TransportProtocol::WebSocket => {
+ let websocket_config = config.websocket.as_ref().unwrap();
+ let client = WebSocketClient::create(websocket_config.clone())?;
+ if establish_connection {
+ Client::connect(&client).await?
+ };
+ Ok(ClientWrapper::WebSocket(client))
+ }
}
}
diff --git a/core/sdk/src/client_wrappers/binary_client.rs b/core/sdk/src/client_wrappers/binary_client.rs
index 41d5744e9..ee650c777 100644
--- a/core/sdk/src/client_wrappers/binary_client.rs
+++ b/core/sdk/src/client_wrappers/binary_client.rs
@@ -30,6 +30,7 @@ impl Client for ClientWrapper {
ClientWrapper::Http(client) => client.connect().await,
ClientWrapper::Tcp(client) => client.connect().await,
ClientWrapper::Quic(client) => client.connect().await,
+ ClientWrapper::WebSocket(client) => client.connect().await,
}
}
@@ -39,6 +40,7 @@ impl Client for ClientWrapper {
ClientWrapper::Http(client) => client.disconnect().await,
ClientWrapper::Tcp(client) => client.disconnect().await,
ClientWrapper::Quic(client) => client.disconnect().await,
+ ClientWrapper::WebSocket(client) => client.disconnect().await,
}
}
@@ -48,6 +50,7 @@ impl Client for ClientWrapper {
ClientWrapper::Http(client) => client.shutdown().await,
ClientWrapper::Tcp(client) => client.shutdown().await,
ClientWrapper::Quic(client) => client.shutdown().await,
+ ClientWrapper::WebSocket(client) => client.shutdown().await,
}
}
@@ -57,6 +60,7 @@ impl Client for ClientWrapper {
ClientWrapper::Http(client) => client.subscribe_events().await,
ClientWrapper::Tcp(client) => client.subscribe_events().await,
ClientWrapper::Quic(client) => client.subscribe_events().await,
+ ClientWrapper::WebSocket(client) => client.subscribe_events().await,
}
}
}
diff --git a/core/sdk/src/client_wrappers/binary_cluster_client.rs b/core/sdk/src/client_wrappers/binary_cluster_client.rs
index 964c2a7e3..ca3cd3478 100644
--- a/core/sdk/src/client_wrappers/binary_cluster_client.rs
+++ b/core/sdk/src/client_wrappers/binary_cluster_client.rs
@@ -29,6 +29,7 @@ impl ClusterClient for ClientWrapper {
ClientWrapper::Http(client) => client.get_cluster_metadata().await,
ClientWrapper::Quic(client) => client.get_cluster_metadata().await,
ClientWrapper::Tcp(client) => client.get_cluster_metadata().await,
+ ClientWrapper::WebSocket(client) => client.get_cluster_metadata().await,
}
}
}
diff --git a/core/sdk/src/client_wrappers/binary_consumer_group_client.rs b/core/sdk/src/client_wrappers/binary_consumer_group_client.rs
index 4c2c9ecd8..d0f3d74b7 100644
--- a/core/sdk/src/client_wrappers/binary_consumer_group_client.rs
+++ b/core/sdk/src/client_wrappers/binary_consumer_group_client.rs
@@ -51,6 +51,11 @@ impl ConsumerGroupClient for ClientWrapper {
.get_consumer_group(stream_id, topic_id, group_id)
.await
}
+ ClientWrapper::WebSocket(client) => {
+ client
+ .get_consumer_group(stream_id, topic_id, group_id)
+ .await
+ }
}
}
@@ -64,6 +69,9 @@ impl ConsumerGroupClient for ClientWrapper {
ClientWrapper::Http(client) => client.get_consumer_groups(stream_id, topic_id).await,
ClientWrapper::Tcp(client) => client.get_consumer_groups(stream_id, topic_id).await,
ClientWrapper::Quic(client) => client.get_consumer_groups(stream_id, topic_id).await,
+ ClientWrapper::WebSocket(client) => {
+ client.get_consumer_groups(stream_id, topic_id).await
+ }
}
}
@@ -94,6 +102,11 @@ impl ConsumerGroupClient for ClientWrapper {
.create_consumer_group(stream_id, topic_id, name)
.await
}
+ ClientWrapper::WebSocket(client) => {
+ client
+ .create_consumer_group(stream_id, topic_id, name)
+ .await
+ }
}
}
@@ -124,6 +137,11 @@ impl ConsumerGroupClient for ClientWrapper {
.delete_consumer_group(stream_id, topic_id, group_id)
.await
}
+ ClientWrapper::WebSocket(client) => {
+ client
+ .delete_consumer_group(stream_id, topic_id, group_id)
+ .await
+ }
}
}
@@ -154,6 +172,11 @@ impl ConsumerGroupClient for ClientWrapper {
.join_consumer_group(stream_id, topic_id, group_id)
.await
}
+ ClientWrapper::WebSocket(client) => {
+ client
+ .join_consumer_group(stream_id, topic_id, group_id)
+ .await
+ }
}
}
@@ -184,6 +207,11 @@ impl ConsumerGroupClient for ClientWrapper {
.leave_consumer_group(stream_id, topic_id, group_id)
.await
}
+ ClientWrapper::WebSocket(client) => {
+ client
+ .leave_consumer_group(stream_id, topic_id, group_id)
+ .await
+ }
}
}
}
@@ -204,6 +232,9 @@ impl AsyncDrop for ClientWrapper {
ClientWrapper::Quic(client) => {
let _ = client.logout_user().await;
}
+ ClientWrapper::WebSocket(client) => {
+ let _ = client.logout_user().await;
+ }
}
}
}
diff --git a/core/sdk/src/client_wrappers/binary_consumer_offset_client.rs b/core/sdk/src/client_wrappers/binary_consumer_offset_client.rs
index 668f120bc..d934f7de9 100644
--- a/core/sdk/src/client_wrappers/binary_consumer_offset_client.rs
+++ b/core/sdk/src/client_wrappers/binary_consumer_offset_client.rs
@@ -52,6 +52,11 @@ impl ConsumerOffsetClient for ClientWrapper {
.store_consumer_offset(consumer, stream_id, topic_id, partition_id, offset)
.await
}
+ ClientWrapper::WebSocket(client) => {
+ client
+ .store_consumer_offset(consumer, stream_id, topic_id, partition_id, offset)
+ .await
+ }
}
}
@@ -83,6 +88,11 @@ impl ConsumerOffsetClient for ClientWrapper {
.get_consumer_offset(consumer, stream_id, topic_id, partition_id)
.await
}
+ ClientWrapper::WebSocket(client) => {
+ client
+ .get_consumer_offset(consumer, stream_id, topic_id, partition_id)
+ .await
+ }
}
}
@@ -114,6 +124,11 @@ impl ConsumerOffsetClient for ClientWrapper {
.delete_consumer_offset(consumer, stream_id, topic_id, partition_id)
.await
}
+ ClientWrapper::WebSocket(client) => {
+ client
+ .delete_consumer_offset(consumer, stream_id, topic_id, partition_id)
+ .await
+ }
}
}
}
diff --git a/core/sdk/src/client_wrappers/binary_message_client.rs b/core/sdk/src/client_wrappers/binary_message_client.rs
index 01e0c1664..48cffbb00 100644
--- a/core/sdk/src/client_wrappers/binary_message_client.rs
+++ b/core/sdk/src/client_wrappers/binary_message_client.rs
@@ -88,6 +88,19 @@ impl MessageClient for ClientWrapper {
)
.await
}
+ ClientWrapper::WebSocket(client) => {
+ client
+ .poll_messages(
+ stream_id,
+ topic_id,
+ partition_id,
+ consumer,
+ strategy,
+ count,
+ auto_commit,
+ )
+ .await
+ }
}
}
@@ -119,6 +132,11 @@ impl MessageClient for ClientWrapper {
.send_messages(stream_id, topic_id, partitioning, messages)
.await
}
+ ClientWrapper::WebSocket(client) => {
+ client
+ .send_messages(stream_id, topic_id, partitioning, messages)
+ .await
+ }
}
}
@@ -150,6 +168,11 @@ impl MessageClient for ClientWrapper {
.flush_unsaved_buffer(stream_id, topic_id, partitioning_id, fsync)
.await
}
+ ClientWrapper::WebSocket(client) => {
+ client
+ .flush_unsaved_buffer(stream_id, topic_id, partitioning_id, fsync)
+ .await
+ }
}
}
}
diff --git a/core/sdk/src/client_wrappers/binary_partition_client.rs b/core/sdk/src/client_wrappers/binary_partition_client.rs
index 26ccad1d8..83e60ed8e 100644
--- a/core/sdk/src/client_wrappers/binary_partition_client.rs
+++ b/core/sdk/src/client_wrappers/binary_partition_client.rs
@@ -50,6 +50,11 @@ impl PartitionClient for ClientWrapper {
.create_partitions(stream_id, topic_id, partitions_count)
.await
}
+ ClientWrapper::WebSocket(client) => {
+ client
+ .create_partitions(stream_id, topic_id, partitions_count)
+ .await
+ }
}
}
@@ -80,6 +85,11 @@ impl PartitionClient for ClientWrapper {
.delete_partitions(stream_id, topic_id, partitions_count)
.await
}
+ ClientWrapper::WebSocket(client) => {
+ client
+ .delete_partitions(stream_id, topic_id, partitions_count)
+ .await
+ }
}
}
}
diff --git a/core/sdk/src/client_wrappers/binary_personal_access_token_client.rs b/core/sdk/src/client_wrappers/binary_personal_access_token_client.rs
index 58c6fbed5..388f52450 100644
--- a/core/sdk/src/client_wrappers/binary_personal_access_token_client.rs
+++ b/core/sdk/src/client_wrappers/binary_personal_access_token_client.rs
@@ -32,6 +32,7 @@ impl PersonalAccessTokenClient for ClientWrapper {
ClientWrapper::Http(client) => client.get_personal_access_tokens().await,
ClientWrapper::Tcp(client) => client.get_personal_access_tokens().await,
ClientWrapper::Quic(client) => client.get_personal_access_tokens().await,
+ ClientWrapper::WebSocket(client) => client.get_personal_access_tokens().await,
}
}
@@ -45,6 +46,9 @@ impl PersonalAccessTokenClient for ClientWrapper {
ClientWrapper::Http(client) => client.create_personal_access_token(name, expiry).await,
ClientWrapper::Tcp(client) => client.create_personal_access_token(name, expiry).await,
ClientWrapper::Quic(client) => client.create_personal_access_token(name, expiry).await,
+ ClientWrapper::WebSocket(client) => {
+ client.create_personal_access_token(name, expiry).await
+ }
}
}
@@ -54,6 +58,7 @@ impl PersonalAccessTokenClient for ClientWrapper {
ClientWrapper::Http(client) => client.delete_personal_access_token(name).await,
ClientWrapper::Tcp(client) => client.delete_personal_access_token(name).await,
ClientWrapper::Quic(client) => client.delete_personal_access_token(name).await,
+ ClientWrapper::WebSocket(client) => client.delete_personal_access_token(name).await,
}
}
@@ -66,6 +71,9 @@ impl PersonalAccessTokenClient for ClientWrapper {
ClientWrapper::Http(client) => client.login_with_personal_access_token(token).await,
ClientWrapper::Tcp(client) => client.login_with_personal_access_token(token).await,
ClientWrapper::Quic(client) => client.login_with_personal_access_token(token).await,
+ ClientWrapper::WebSocket(client) => {
+ client.login_with_personal_access_token(token).await
+ }
}
}
}
diff --git a/core/sdk/src/client_wrappers/binary_segment_client.rs b/core/sdk/src/client_wrappers/binary_segment_client.rs
index 0ec082ff0..f0e3b8cff 100644
--- a/core/sdk/src/client_wrappers/binary_segment_client.rs
+++ b/core/sdk/src/client_wrappers/binary_segment_client.rs
@@ -51,6 +51,11 @@ impl SegmentClient for ClientWrapper {
.delete_segments(stream_id, topic_id, partition_id, segments_count)
.await
}
+ ClientWrapper::WebSocket(client) => {
+ client
+ .delete_segments(stream_id, topic_id, partition_id, segments_count)
+ .await
+ }
}
}
}
diff --git a/core/sdk/src/client_wrappers/binary_stream_client.rs b/core/sdk/src/client_wrappers/binary_stream_client.rs
index 782916680..a14d26c0a 100644
--- a/core/sdk/src/client_wrappers/binary_stream_client.rs
+++ b/core/sdk/src/client_wrappers/binary_stream_client.rs
@@ -29,6 +29,7 @@ impl StreamClient for ClientWrapper {
ClientWrapper::Http(client) => client.get_stream(stream_id).await,
ClientWrapper::Tcp(client) => client.get_stream(stream_id).await,
ClientWrapper::Quic(client) => client.get_stream(stream_id).await,
+ ClientWrapper::WebSocket(client) => client.get_stream(stream_id).await,
}
}
@@ -38,6 +39,7 @@ impl StreamClient for ClientWrapper {
ClientWrapper::Http(client) => client.get_streams().await,
ClientWrapper::Tcp(client) => client.get_streams().await,
ClientWrapper::Quic(client) => client.get_streams().await,
+ ClientWrapper::WebSocket(client) => client.get_streams().await,
}
}
@@ -47,6 +49,7 @@ impl StreamClient for ClientWrapper {
ClientWrapper::Http(client) => client.create_stream(name).await,
ClientWrapper::Tcp(client) => client.create_stream(name).await,
ClientWrapper::Quic(client) => client.create_stream(name).await,
+ ClientWrapper::WebSocket(client) => client.create_stream(name).await,
}
}
@@ -56,6 +59,7 @@ impl StreamClient for ClientWrapper {
ClientWrapper::Http(client) => client.update_stream(stream_id, name).await,
ClientWrapper::Tcp(client) => client.update_stream(stream_id, name).await,
ClientWrapper::Quic(client) => client.update_stream(stream_id, name).await,
+ ClientWrapper::WebSocket(client) => client.update_stream(stream_id, name).await,
}
}
@@ -65,6 +69,7 @@ impl StreamClient for ClientWrapper {
ClientWrapper::Http(client) => client.delete_stream(stream_id).await,
ClientWrapper::Tcp(client) => client.delete_stream(stream_id).await,
ClientWrapper::Quic(client) => client.delete_stream(stream_id).await,
+ ClientWrapper::WebSocket(client) => client.delete_stream(stream_id).await,
}
}
@@ -74,6 +79,7 @@ impl StreamClient for ClientWrapper {
ClientWrapper::Http(client) => client.purge_stream(stream_id).await,
ClientWrapper::Tcp(client) => client.purge_stream(stream_id).await,
ClientWrapper::Quic(client) => client.purge_stream(stream_id).await,
+ ClientWrapper::WebSocket(client) => client.purge_stream(stream_id).await,
}
}
}
diff --git a/core/sdk/src/client_wrappers/binary_system_client.rs b/core/sdk/src/client_wrappers/binary_system_client.rs
index 825dbd8e5..39363b726 100644
--- a/core/sdk/src/client_wrappers/binary_system_client.rs
+++ b/core/sdk/src/client_wrappers/binary_system_client.rs
@@ -32,6 +32,7 @@ impl SystemClient for ClientWrapper {
ClientWrapper::Http(client) => client.get_stats().await,
ClientWrapper::Tcp(client) => client.get_stats().await,
ClientWrapper::Quic(client) => client.get_stats().await,
+ ClientWrapper::WebSocket(client) => client.get_stats().await,
}
}
@@ -41,6 +42,7 @@ impl SystemClient for ClientWrapper {
ClientWrapper::Http(client) => client.get_me().await,
ClientWrapper::Tcp(client) => client.get_me().await,
ClientWrapper::Quic(client) => client.get_me().await,
+ ClientWrapper::WebSocket(client) => client.get_me().await,
}
}
@@ -50,6 +52,7 @@ impl SystemClient for ClientWrapper {
ClientWrapper::Http(client) => client.get_client(client_id).await,
ClientWrapper::Tcp(client) => client.get_client(client_id).await,
ClientWrapper::Quic(client) => client.get_client(client_id).await,
+ ClientWrapper::WebSocket(client) => client.get_client(client_id).await,
}
}
@@ -59,6 +62,7 @@ impl SystemClient for ClientWrapper {
ClientWrapper::Http(client) => client.get_clients().await,
ClientWrapper::Tcp(client) => client.get_clients().await,
ClientWrapper::Quic(client) => client.get_clients().await,
+ ClientWrapper::WebSocket(client) => client.get_clients().await,
}
}
@@ -68,6 +72,7 @@ impl SystemClient for ClientWrapper {
ClientWrapper::Http(client) => client.ping().await,
ClientWrapper::Tcp(client) => client.ping().await,
ClientWrapper::Quic(client) => client.ping().await,
+ ClientWrapper::WebSocket(client) => client.ping().await,
}
}
@@ -77,6 +82,7 @@ impl SystemClient for ClientWrapper {
ClientWrapper::Http(client) => client.heartbeat_interval().await,
ClientWrapper::Tcp(client) => client.heartbeat_interval().await,
ClientWrapper::Quic(client) => client.heartbeat_interval().await,
+ ClientWrapper::WebSocket(client) => client.heartbeat_interval().await,
}
}
@@ -90,6 +96,7 @@ impl SystemClient for ClientWrapper {
ClientWrapper::Http(client) => client.snapshot(compression, snapshot_types).await,
ClientWrapper::Tcp(client) => client.snapshot(compression, snapshot_types).await,
ClientWrapper::Quic(client) => client.snapshot(compression, snapshot_types).await,
+ ClientWrapper::WebSocket(client) => client.snapshot(compression, snapshot_types).await,
}
}
}
diff --git a/core/sdk/src/client_wrappers/binary_topic_client.rs b/core/sdk/src/client_wrappers/binary_topic_client.rs
index 0f8772261..9c8307fe0 100644
--- a/core/sdk/src/client_wrappers/binary_topic_client.rs
+++ b/core/sdk/src/client_wrappers/binary_topic_client.rs
@@ -35,6 +35,7 @@ impl TopicClient for ClientWrapper {
ClientWrapper::Http(client) => client.get_topic(stream_id, topic_id).await,
ClientWrapper::Tcp(client) => client.get_topic(stream_id, topic_id).await,
ClientWrapper::Quic(client) => client.get_topic(stream_id, topic_id).await,
+ ClientWrapper::WebSocket(client) => client.get_topic(stream_id, topic_id).await,
}
}
@@ -44,6 +45,7 @@ impl TopicClient for ClientWrapper {
ClientWrapper::Http(client) => client.get_topics(stream_id).await,
ClientWrapper::Tcp(client) => client.get_topics(stream_id).await,
ClientWrapper::Quic(client) => client.get_topics(stream_id).await,
+ ClientWrapper::WebSocket(client) => client.get_topics(stream_id).await,
}
}
@@ -110,6 +112,19 @@ impl TopicClient for ClientWrapper {
)
.await
}
+ ClientWrapper::WebSocket(client) => {
+ client
+ .create_topic(
+ stream_id,
+ name,
+ partitions_count,
+ compression_algorithm,
+ replication_factor,
+ message_expiry,
+ max_topic_size,
+ )
+ .await
+ }
}
}
@@ -176,6 +191,19 @@ impl TopicClient for ClientWrapper {
)
.await
}
+ ClientWrapper::WebSocket(client) => {
+ client
+ .update_topic(
+ stream_id,
+ topic_id,
+ name,
+ compression_algorithm,
+ replication_factor,
+ message_expiry,
+ max_topic_size,
+ )
+ .await
+ }
}
}
@@ -189,6 +217,7 @@ impl TopicClient for ClientWrapper {
ClientWrapper::Http(client) => client.delete_topic(stream_id, topic_id).await,
ClientWrapper::Tcp(client) => client.delete_topic(stream_id, topic_id).await,
ClientWrapper::Quic(client) => client.delete_topic(stream_id, topic_id).await,
+ ClientWrapper::WebSocket(client) => client.delete_topic(stream_id, topic_id).await,
}
}
@@ -202,6 +231,7 @@ impl TopicClient for ClientWrapper {
ClientWrapper::Http(client) => client.purge_topic(stream_id, topic_id).await,
ClientWrapper::Tcp(client) => client.purge_topic(stream_id, topic_id).await,
ClientWrapper::Quic(client) => client.purge_topic(stream_id, topic_id).await,
+ ClientWrapper::WebSocket(client) => client.purge_topic(stream_id, topic_id).await,
}
}
}
diff --git a/core/sdk/src/client_wrappers/binary_user_client.rs b/core/sdk/src/client_wrappers/binary_user_client.rs
index f7289c244..a51234f89 100644
--- a/core/sdk/src/client_wrappers/binary_user_client.rs
+++ b/core/sdk/src/client_wrappers/binary_user_client.rs
@@ -31,6 +31,7 @@ impl UserClient for ClientWrapper {
ClientWrapper::Http(client) => client.get_user(user_id).await,
ClientWrapper::Tcp(client) => client.get_user(user_id).await,
ClientWrapper::Quic(client) => client.get_user(user_id).await,
+ ClientWrapper::WebSocket(client) => client.get_user(user_id).await,
}
}
@@ -40,6 +41,7 @@ impl UserClient for ClientWrapper {
ClientWrapper::Http(client) => client.get_users().await,
ClientWrapper::Tcp(client) => client.get_users().await,
ClientWrapper::Quic(client) => client.get_users().await,
+ ClientWrapper::WebSocket(client) => client.get_users().await,
}
}
@@ -71,6 +73,11 @@ impl UserClient for ClientWrapper {
.create_user(username, password, status, permissions)
.await
}
+ ClientWrapper::WebSocket(client) => {
+ client
+ .create_user(username, password, status, permissions)
+ .await
+ }
}
}
@@ -80,6 +87,7 @@ impl UserClient for ClientWrapper {
ClientWrapper::Tcp(client) => client.delete_user(user_id).await,
ClientWrapper::Quic(client) => client.delete_user(user_id).await,
ClientWrapper::Iggy(client) => client.delete_user(user_id).await,
+ ClientWrapper::WebSocket(client) => client.delete_user(user_id).await,
}
}
@@ -94,6 +102,7 @@ impl UserClient for ClientWrapper {
ClientWrapper::Tcp(client) => client.update_user(user_id, username, status).await,
ClientWrapper::Quic(client) => client.update_user(user_id, username, status).await,
ClientWrapper::Iggy(client) => client.update_user(user_id, username, status).await,
+ ClientWrapper::WebSocket(client) => client.update_user(user_id, username, status).await,
}
}
@@ -107,6 +116,9 @@ impl UserClient for ClientWrapper {
ClientWrapper::Http(client) => client.update_permissions(user_id, permissions).await,
ClientWrapper::Tcp(client) => client.update_permissions(user_id, permissions).await,
ClientWrapper::Quic(client) => client.update_permissions(user_id, permissions).await,
+ ClientWrapper::WebSocket(client) => {
+ client.update_permissions(user_id, permissions).await
+ }
}
}
@@ -137,6 +149,11 @@ impl UserClient for ClientWrapper {
.change_password(user_id, current_password, new_password)
.await
}
+ ClientWrapper::WebSocket(client) => {
+ client
+ .change_password(user_id, current_password, new_password)
+ .await
+ }
}
}
@@ -146,6 +163,7 @@ impl UserClient for ClientWrapper {
ClientWrapper::Http(client) => client.login_user(username, password).await,
ClientWrapper::Tcp(client) => client.login_user(username, password).await,
ClientWrapper::Quic(client) => client.login_user(username, password).await,
+ ClientWrapper::WebSocket(client) => client.login_user(username, password).await,
}
}
@@ -155,6 +173,7 @@ impl UserClient for ClientWrapper {
ClientWrapper::Http(client) => client.logout_user().await,
ClientWrapper::Tcp(client) => client.logout_user().await,
ClientWrapper::Quic(client) => client.logout_user().await,
+ ClientWrapper::WebSocket(client) => client.logout_user().await,
}
}
}
diff --git a/core/sdk/src/client_wrappers/client_wrapper.rs b/core/sdk/src/client_wrappers/client_wrapper.rs
index d391bfda3..669b2545b 100644
--- a/core/sdk/src/client_wrappers/client_wrapper.rs
+++ b/core/sdk/src/client_wrappers/client_wrapper.rs
@@ -20,6 +20,7 @@ use crate::clients::client::IggyClient;
use crate::http::http_client::HttpClient;
use crate::quic::quic_client::QuicClient;
use crate::tcp::tcp_client::TcpClient;
+use crate::websocket::websocket_client::WebSocketClient;
#[allow(clippy::large_enum_variant)]
#[derive(Debug)]
@@ -28,4 +29,5 @@ pub enum ClientWrapper {
Http(HttpClient),
Tcp(TcpClient),
Quic(QuicClient),
+ WebSocket(WebSocketClient),
}
diff --git a/core/sdk/src/clients/client.rs b/core/sdk/src/clients/client.rs
index c13e3531e..15b39db2b 100644
--- a/core/sdk/src/clients/client.rs
+++ b/core/sdk/src/clients/client.rs
@@ -28,6 +28,7 @@ use crate::prelude::IggyError;
use crate::prelude::IggyProducerBuilder;
use crate::quic::quic_client::QuicClient;
use crate::tcp::tcp_client::TcpClient;
+use crate::websocket::websocket_client::WebSocketClient;
use async_broadcast::Receiver;
use async_trait::async_trait;
use iggy_binary_protocol::{Client, SystemClient};
@@ -93,6 +94,9 @@ impl IggyClient {
TransportProtocol::Http => Ok(IggyClient::new(ClientWrapper::Http(
HttpClient::from_connection_string(connection_string)?,
))),
+ TransportProtocol::WebSocket => Ok(IggyClient::new(ClientWrapper::WebSocket(
+ WebSocketClient::from_connection_string(connection_string)?,
+ ))),
}
}
diff --git a/core/sdk/src/clients/client_builder.rs b/core/sdk/src/clients/client_builder.rs
index 205dcd8f5..08199ce3b 100644
--- a/core/sdk/src/clients/client_builder.rs
+++ b/core/sdk/src/clients/client_builder.rs
@@ -25,6 +25,7 @@ use crate::prelude::{
};
use crate::quic::quic_client::QuicClient;
use crate::tcp::tcp_client::TcpClient;
+use crate::websocket::websocket_client::WebSocketClient;
use iggy_common::{ConnectionStringUtils, TransportProtocol};
use std::sync::Arc;
use tracing::error;
@@ -64,6 +65,11 @@ impl IggyClientBuilder {
connection_string,
)?));
}
+ TransportProtocol::WebSocket => {
+ builder.client = Some(ClientWrapper::WebSocket(
+ WebSocketClient::from_connection_string(connection_string)?,
+ ));
+ }
}
Ok(builder)
diff --git a/core/sdk/src/lib.rs b/core/sdk/src/lib.rs
index c315fd5e2..c68b7dd5a 100644
--- a/core/sdk/src/lib.rs
+++ b/core/sdk/src/lib.rs
@@ -26,3 +26,4 @@ pub mod prelude;
pub mod quic;
pub mod stream_builder;
pub mod tcp;
+pub mod websocket;
diff --git a/core/sdk/src/prelude.rs b/core/sdk/src/prelude.rs
index ab22827a8..5f4d03ce0 100644
--- a/core/sdk/src/prelude.rs
+++ b/core/sdk/src/prelude.rs
@@ -45,6 +45,7 @@ pub use crate::stream_builder::IggyStreamConsumer;
pub use crate::stream_builder::{IggyProducerConfig, IggyStreamProducer};
pub use crate::stream_builder::{IggyStream, IggyStreamConfig};
pub use crate::tcp::tcp_client::TcpClient;
+pub use crate::websocket::websocket_client::WebSocketClient;
pub use iggy_binary_protocol::{
Client, ClusterClient, ConsumerGroupClient, ConsumerOffsetClient, MessageClient,
PartitionClient, PersonalAccessTokenClient, SegmentClient, StreamClient, SystemClient,
@@ -63,7 +64,8 @@ pub use iggy_common::{
Sizeable, SnapshotCompression, Stats, Stream, StreamDetails, StreamPermissions,
SystemSnapshotType, TcpClientConfig, TcpClientConfigBuilder, TcpClientReconnectionConfig,
Topic, TopicDetails, TopicPermissions, TransportProtocol, UserId, UserStatus, Validatable,
- defaults, locking,
+ defaults, locking, WebSocketClientConfig,
+ WebSocketClientConfigBuilder, WebSocketClientReconnectionConfig
};
pub use iggy_common::{
IGGY_MESSAGE_CHECKSUM_OFFSET_RANGE, IGGY_MESSAGE_HEADER_SIZE,
diff --git a/core/sdk/src/lib.rs b/core/sdk/src/websocket/mod.rs
similarity index 80%
copy from core/sdk/src/lib.rs
copy to core/sdk/src/websocket/mod.rs
index c315fd5e2..40e3aa46b 100644
--- a/core/sdk/src/lib.rs
+++ b/core/sdk/src/websocket/mod.rs
@@ -16,13 +16,6 @@
* under the License.
*/
-#[allow(deprecated)]
-pub mod client_provider;
-pub mod client_wrappers;
-pub mod clients;
-pub mod consumer_ext;
-pub mod http;
-pub mod prelude;
-pub mod quic;
-pub mod stream_builder;
-pub mod tcp;
+pub mod websocket_client;
+pub(crate) mod websocket_connection_stream;
+pub(crate) mod websocket_stream;
diff --git a/core/sdk/src/websocket/websocket_client.rs b/core/sdk/src/websocket/websocket_client.rs
new file mode 100644
index 000000000..1cb4730db
--- /dev/null
+++ b/core/sdk/src/websocket/websocket_client.rs
@@ -0,0 +1,539 @@
+/* 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.
+ */
+
+use crate::websocket::websocket_connection_stream::WebSocketConnectionStream;
+use crate::{prelude::Client, websocket::websocket_stream::ConnectionStream};
+use async_broadcast::{Receiver, Sender, broadcast};
+use async_trait::async_trait;
+use bytes::{BufMut, Bytes, BytesMut};
+use iggy_binary_protocol::{BinaryClient, BinaryTransport, PersonalAccessTokenClient, UserClient};
+use iggy_common::{
+ AutoLogin, ClientState, Command, ConnectionString, Credentials, DiagnosticEvent, IggyDuration,
+ IggyError, IggyTimestamp, WebSocketClientConfig, WebSocketConnectionStringOptions,
+};
+use std::net::SocketAddr;
+use std::sync::Arc;
+use tokio::net::TcpStream;
+use tokio::sync::Mutex;
+use tokio::time::sleep;
+use tokio_tungstenite::{client_async_with_config, tungstenite::client::IntoClientRequest};
+use tracing::{debug, error, info, trace, warn};
+
+const REQUEST_INITIAL_BYTES_LENGTH: usize = 4;
+const RESPONSE_INITIAL_BYTES_LENGTH: usize = 8;
+const NAME: &str = "WebSocket";
+
+#[derive(Debug)]
+pub struct WebSocketClient {
+ stream: Arc<Mutex<Option<WebSocketConnectionStream>>>,
+ pub(crate) config: Arc<WebSocketClientConfig>,
+ pub(crate) state: Mutex<ClientState>,
+ client_address: Mutex<Option<SocketAddr>>,
+ events: (Sender<DiagnosticEvent>, Receiver<DiagnosticEvent>),
+ connected_at: Mutex<Option<IggyTimestamp>>,
+}
+
+impl Default for WebSocketClient {
+ fn default() -> Self {
+ WebSocketClient::create(Arc::new(WebSocketClientConfig::default())).unwrap()
+ }
+}
+
+#[async_trait]
+impl Client for WebSocketClient {
+ async fn connect(&self) -> Result<(), IggyError> {
+ WebSocketClient::connect(self).await
+ }
+
+ async fn disconnect(&self) -> Result<(), IggyError> {
+ WebSocketClient::disconnect(self).await
+ }
+
+ async fn shutdown(&self) -> Result<(), IggyError> {
+ WebSocketClient::shutdown(self).await
+ }
+
+ async fn subscribe_events(&self) -> Receiver<DiagnosticEvent> {
+ self.events.1.clone()
+ }
+}
+
+#[async_trait]
+impl BinaryTransport for WebSocketClient {
+ async fn get_state(&self) -> ClientState {
+ *self.state.lock().await
+ }
+
+ async fn set_state(&self, state: ClientState) {
+ *self.state.lock().await = state;
+ }
+
+ async fn publish_event(&self, event: DiagnosticEvent) {
+ if let Err(error) = self.events.0.broadcast(event).await {
+ error!("Failed to send a {} diagnostic event: {error}", NAME);
+ }
+ }
+
+ async fn send_with_response<T: Command>(&self, command: &T) -> Result<Bytes, IggyError> {
+ command.validate()?;
+ self.send_raw_with_response(command.code(), command.to_bytes())
+ .await
+ }
+
+ async fn send_raw_with_response(&self, code: u32, payload: Bytes) -> Result<Bytes, IggyError> {
+ let result = self.send_raw(code, payload.clone()).await;
+ if result.is_ok() {
+ return result;
+ }
+
+ let error = result.unwrap_err();
+ if !matches!(
+ error,
+ IggyError::Disconnected
+ | IggyError::EmptyResponse
+ | IggyError::Unauthenticated
+ | IggyError::StaleClient
+ ) {
+ return Err(error);
+ }
+
+ if !self.config.reconnection.enabled {
+ return Err(IggyError::Disconnected);
+ }
+
+ self.disconnect().await?;
+
+ {
+ let client_address = self.get_client_address_value().await;
+ info!(
+ "Reconnecting to the server: {} by client: {client_address}...",
+ self.config.server_address
+ );
+ }
+
+ self.connect().await?;
+ self.send_raw(code, payload).await
+ }
+
+ fn get_heartbeat_interval(&self) -> IggyDuration {
+ self.config.heartbeat_interval
+ }
+}
+
+impl BinaryClient for WebSocketClient {}
+
+impl WebSocketClient {
+ /// Create a new WebSocket client with the provided configuration.
+ pub fn create(config: Arc<WebSocketClientConfig>) -> Result<Self, IggyError> {
+ let (sender, receiver) = broadcast(1000);
+ Ok(WebSocketClient {
+ stream: Arc::new(Mutex::new(None)),
+ config,
+ state: Mutex::new(ClientState::Disconnected),
+ client_address: Mutex::new(None),
+ events: (sender, receiver),
+ connected_at: Mutex::new(None),
+ })
+ }
+
+ /// Create a new WebSocket client from a connection string.
+ pub fn from_connection_string(connection_string: &str) -> Result<Self, IggyError> {
+ let parsed_connection_string =
+ ConnectionString::<WebSocketConnectionStringOptions>::new(connection_string)?;
+ let config = WebSocketClientConfig::from(parsed_connection_string);
+ Self::create(Arc::new(config))
+ }
+
+ async fn get_client_address_value(&self) -> String {
+ let client_address = self.client_address.lock().await;
+ match client_address.as_ref() {
+ Some(address) => address.to_string(),
+ None => "unknown".to_string(),
+ }
+ }
+
+ async fn connect(&self) -> Result<(), IggyError> {
+ if self.get_state().await == ClientState::Connected {
+ return Ok(());
+ }
+
+ let mut retry_count = 0;
+
+ loop {
+ info!(
+ "{NAME} client is connecting to server: {}...",
+ self.config.server_address
+ );
+ self.set_state(ClientState::Connecting).await;
+
+ if retry_count > 0 {
+ let elapsed = self
+ .connected_at
+ .lock()
+ .await
+ .map(|ts| IggyTimestamp::now().as_micros() - ts.as_micros())
+ .unwrap_or(0);
+
+ let interval = self.config.reconnection.reestablish_after.as_micros();
+ debug!("Elapsed time since last connection: {}μs", elapsed);
+
+ if elapsed < interval {
+ let remaining =
+ IggyDuration::new(std::time::Duration::from_micros(interval - elapsed));
+ info!("Trying to connect to the server in: {remaining}");
+ sleep(remaining.get_duration()).await;
+ }
+ }
+
+ let server_addr = self
+ .config
+ .server_address
+ .parse::<SocketAddr>()
+ .map_err(|_| {
+ error!("Invalid server address: {}", self.config.server_address);
+ IggyError::InvalidConfiguration
+ })?;
+
+ let tcp_stream = match TcpStream::connect(&server_addr).await {
+ Ok(stream) => stream,
+ Err(error) => {
+ error!(
+ "Failed to connect to server: {}. Error: {}",
+ self.config.server_address, error
+ );
+
+ if !self.config.reconnection.enabled {
+ warn!("Automatic reconnection is disabled.");
+ return Err(IggyError::CannotEstablishConnection);
+ }
+
+ let unlimited_retries = self.config.reconnection.max_retries.is_none();
+ let max_retries = self.config.reconnection.max_retries.unwrap_or_default();
+ let max_retries_str = self
+ .config
+ .reconnection
+ .max_retries
+ .map(|r| r.to_string())
+ .unwrap_or_else(|| "unlimited".to_string());
+
+ let interval_str = self.config.reconnection.interval.as_human_time_string();
+ if unlimited_retries || retry_count < max_retries {
+ retry_count += 1;
+ info!(
+ "Retrying to connect to server ({retry_count}/{max_retries_str}): {} in: {interval_str}",
+ self.config.server_address,
+ );
+ sleep(self.config.reconnection.interval.get_duration()).await;
+ continue;
+ }
+
+ self.set_state(ClientState::Disconnected).await;
+ self.publish_event(DiagnosticEvent::Disconnected).await;
+ return Err(IggyError::CannotEstablishConnection);
+ }
+ };
+
+ let ws_url = format!("ws://{}", server_addr);
+
+ let request = ws_url.into_client_request().map_err(|e| {
+ error!("Failed to create WebSocket request: {}", e);
+ IggyError::InvalidConfiguration
+ })?;
+
+ let tungstenite_config = self.config.ws_config.to_tungstenite_config();
+
+ let (websocket_stream, response) =
+ match client_async_with_config(request, tcp_stream, tungstenite_config).await {
+ Ok(result) => result,
+ Err(error) => {
+ error!("WebSocket handshake failed: {}", error);
+
+ if !self.config.reconnection.enabled {
+ return Err(IggyError::WebSocketConnectionError);
+ }
+
+ let unlimited_retries = self.config.reconnection.max_retries.is_none();
+ let max_retries = self.config.reconnection.max_retries.unwrap_or_default();
+
+ if unlimited_retries || retry_count < max_retries {
+ retry_count += 1;
+ sleep(self.config.reconnection.interval.get_duration()).await;
+ continue;
+ }
+
+ return Err(IggyError::WebSocketConnectionError);
+ }
+ };
+
+ debug!(
+ "WebSocket connection established. Response status: {}",
+ response.status()
+ );
+
+ let connection_stream = WebSocketConnectionStream::new(server_addr, websocket_stream);
+
+ *self.stream.lock().await = Some(connection_stream);
+ *self.client_address.lock().await = Some(server_addr);
+ self.set_state(ClientState::Connected).await;
+ *self.connected_at.lock().await = Some(IggyTimestamp::now());
+ self.publish_event(DiagnosticEvent::Connected).await;
+
+ let now = IggyTimestamp::now();
+ info!(
+ "{NAME} client has connected to server: {} at: {now}",
+ server_addr
+ );
+
+ self.auto_login().await?;
+ return Ok(());
+ }
+ }
+
+ async fn auto_login(&self) -> Result<(), IggyError> {
+ let client_address = self.get_client_address_value().await;
+ match &self.config.auto_login {
+ AutoLogin::Disabled => {
+ info!("{NAME} client: {client_address} - automatic sign-in is disabled.");
+ Ok(())
+ }
+ AutoLogin::Enabled(credentials) => {
+ info!("{NAME} client: {client_address} is signing in...");
+ self.set_state(ClientState::Authenticating).await;
+ match credentials {
+ Credentials::UsernamePassword(username, password) => {
+ self.login_user(username, password).await?;
+ info!(
+ "{NAME} client: {client_address} has signed in with the user credentials, username: {username}",
+ );
+ Ok(())
+ }
+ Credentials::PersonalAccessToken(token) => {
+ self.login_with_personal_access_token(token).await?;
+ info!(
+ "{NAME} client: {client_address} has signed in with a personal access token.",
+ );
+ Ok(())
+ }
+ }
+ }
+ }
+ }
+
+ async fn disconnect(&self) -> Result<(), IggyError> {
+ if self.get_state().await == ClientState::Disconnected {
+ return Ok(());
+ }
+
+ let client_address = self.get_client_address_value().await;
+ info!("{NAME} client: {client_address} is disconnecting from server...");
+ self.set_state(ClientState::Disconnected).await;
+
+ if let Some(mut stream) = self.stream.lock().await.take() {
+ let _ = stream.shutdown().await;
+ }
+
+ self.publish_event(DiagnosticEvent::Disconnected).await;
+ let now = IggyTimestamp::now();
+ info!("{NAME} client: {client_address} has disconnected from server at: {now}.");
+ Ok(())
+ }
+
+ async fn shutdown(&self) -> Result<(), IggyError> {
+ if self.get_state().await == ClientState::Shutdown {
+ return Ok(());
+ }
+
+ let client_address = self.get_client_address_value().await;
+ info!("Shutting down the {NAME} client: {client_address}");
+
+ self.disconnect().await?;
+ self.set_state(ClientState::Shutdown).await;
+ self.publish_event(DiagnosticEvent::Shutdown).await;
+ info!("{NAME} client: {client_address} has been shutdown.");
+ Ok(())
+ }
+
+ async fn send_raw(&self, code: u32, payload: Bytes) -> Result<Bytes, IggyError> {
+ match self.get_state().await {
+ ClientState::Shutdown => {
+ trace!("Cannot send data. Client is shutdown.");
+ return Err(IggyError::ClientShutdown);
+ }
+ ClientState::Disconnected => {
+ trace!("Cannot send data. Client is not connected.");
+ return Err(IggyError::NotConnected);
+ }
+ ClientState::Connecting => {
+ trace!("Cannot send data. Client is still connecting.");
+ return Err(IggyError::NotConnected);
+ }
+ ClientState::Connected | ClientState::Authenticating | ClientState::Authenticated => {}
+ }
+
+ let mut stream_guard = self.stream.lock().await;
+ let stream = stream_guard.as_mut().ok_or_else(|| {
+ trace!("Cannot send data. Client is not connected.");
+ IggyError::NotConnected
+ })?;
+
+ let payload_length = payload.len() + REQUEST_INITIAL_BYTES_LENGTH;
+ let mut request = BytesMut::with_capacity(4 + REQUEST_INITIAL_BYTES_LENGTH + payload.len());
+ request.put_u32_le(payload_length as u32);
+ request.put_u32_le(code);
+ request.put_slice(&payload);
+
+ trace!(
+ "Sending {NAME} message with code: {}, payload size: {} bytes",
+ code,
+ payload.len()
+ );
+
+ stream.write(&request).await?;
+ stream.flush().await?;
+
+ let mut response_initial_buffer = vec![0u8; RESPONSE_INITIAL_BYTES_LENGTH];
+ stream.read(&mut response_initial_buffer).await?;
+
+ let status = u32::from_le_bytes([
+ response_initial_buffer[0],
+ response_initial_buffer[1],
+ response_initial_buffer[2],
+ response_initial_buffer[3],
+ ]);
+
+ let length = u32::from_le_bytes([
+ response_initial_buffer[4],
+ response_initial_buffer[5],
+ response_initial_buffer[6],
+ response_initial_buffer[7],
+ ]) as usize;
+
+ trace!(
+ "Received {NAME} response status: {}, length: {} bytes",
+ status, length
+ );
+
+ if status != 0 {
+ if length > 0 {
+ let mut error_buffer = vec![0u8; length];
+ stream.read(&mut error_buffer).await?;
+ let _error_message = String::from_utf8(error_buffer).unwrap_or_default();
+ // FIXME: add error message
+ return Err(IggyError::Error);
+ }
+ // FIXME: add error message
+ return Err(IggyError::Error);
+ }
+
+ if length == 0 {
+ return Ok(Bytes::new());
+ }
+
+ let mut response_buffer = vec![0u8; length];
+ stream.read(&mut response_buffer).await?;
+
+ trace!("Received {NAME} response payload, size: {} bytes", length);
+ Ok(Bytes::from(response_buffer))
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use std::str::FromStr;
+
+ #[test]
+ fn should_be_created_with_default_config() {
+ let client = WebSocketClient::default();
+ assert_eq!(client.config.server_address, "127.0.0.1:8080");
+ assert_eq!(
+ client.config.heartbeat_interval,
+ IggyDuration::from_str("5s").unwrap()
+ );
+ assert!(matches!(client.config.auto_login, AutoLogin::Disabled));
+ assert!(client.config.reconnection.enabled);
+ }
+
+ #[tokio::test]
+ async fn should_be_disconnected_by_default() {
+ let client = WebSocketClient::default();
+ assert_eq!(client.get_state().await, ClientState::Disconnected);
+ }
+
+ #[test]
+ fn should_succeed_from_connection_string() {
+ let connection_string = "iggy+websocket://user:secret@127.0.0.1:8080";
+ let client = WebSocketClient::from_connection_string(connection_string);
+ assert!(client.is_ok());
+ }
+
+ #[test]
+ fn should_create_with_custom_config() {
+ let config = WebSocketClientConfig {
+ server_address: "localhost:9090".to_string(),
+ heartbeat_interval: IggyDuration::from_str("10s").unwrap(),
+ ..Default::default()
+ };
+
+ let client = WebSocketClient::create(Arc::new(config));
+ assert!(client.is_ok());
+
+ let client = client.unwrap();
+ assert_eq!(client.config.server_address, "localhost:9090");
+ assert_eq!(
+ client.config.heartbeat_interval,
+ IggyDuration::from_str("10s").unwrap()
+ );
+ }
+
+ #[test]
+ fn should_fail_with_empty_connection_string() {
+ let value = "";
+ let client = WebSocketClient::from_connection_string(value);
+ assert!(client.is_err());
+ }
+
+ #[test]
+ fn should_fail_without_username() {
+ let connection_string = "iggy+websocket://:secret@127.0.0.1:8080";
+ let client = WebSocketClient::from_connection_string(connection_string);
+ assert!(client.is_err());
+ }
+
+ #[test]
+ fn should_fail_without_password() {
+ let connection_string = "iggy+websocket://user:@127.0.0.1:8080";
+ let client = WebSocketClient::from_connection_string(connection_string);
+ assert!(client.is_err());
+ }
+
+ #[test]
+ fn should_fail_without_server_address() {
+ let connection_string = "iggy+websocket://user:secret@:8080";
+ let client = WebSocketClient::from_connection_string(connection_string);
+ assert!(client.is_err());
+ }
+
+ #[test]
+ fn should_fail_with_invalid_options() {
+ let connection_string =
+ "iggy+websocket://user:secret@127.0.0.1:8080?invalid_option=invalid";
+ let client = WebSocketClient::from_connection_string(connection_string);
+ assert!(client.is_err());
+ }
+}
diff --git a/core/sdk/src/websocket/websocket_connection_stream.rs b/core/sdk/src/websocket/websocket_connection_stream.rs
new file mode 100644
index 000000000..e71a47283
--- /dev/null
+++ b/core/sdk/src/websocket/websocket_connection_stream.rs
@@ -0,0 +1,219 @@
+/* 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.
+ */
+
+use crate::websocket::websocket_stream::ConnectionStream;
+use async_trait::async_trait;
+use bytes::BytesMut;
+use futures_util::{SinkExt, StreamExt};
+use iggy_common::IggyError;
+use std::io::ErrorKind;
+use std::net::SocketAddr;
+use tokio::net::TcpStream;
+use tokio_tungstenite::{WebSocketStream, tungstenite::Message};
+use tracing::{debug, error, trace};
+
+#[derive(Debug)]
+pub struct WebSocketConnectionStream {
+ client_address: SocketAddr,
+ stream: WebSocketStream<TcpStream>,
+ read_buffer: BytesMut,
+}
+
+impl WebSocketConnectionStream {
+ pub fn new(client_address: SocketAddr, stream: WebSocketStream<TcpStream>) -> Self {
+ Self {
+ client_address,
+ stream,
+ read_buffer: BytesMut::new(),
+ }
+ }
+
+ /// Internal method to read a WebSocket message and buffer it
+ async fn read_message(&mut self) -> Result<(), IggyError> {
+ loop {
+ match self.stream.next().await {
+ Some(Ok(Message::Binary(data))) => {
+ trace!(
+ "Received WebSocket binary message from {}, size: {} bytes",
+ self.client_address,
+ data.len()
+ );
+ self.read_buffer.extend_from_slice(&data);
+ return Ok(());
+ }
+ Some(Ok(Message::Text(text))) => {
+ trace!(
+ "Received WebSocket text message from {}, converting to binary",
+ self.client_address
+ );
+ self.read_buffer.extend_from_slice(text.as_bytes());
+ return Ok(());
+ }
+ Some(Ok(Message::Ping(data))) => {
+ trace!(
+ "Received WebSocket ping from {}, sending pong",
+ self.client_address
+ );
+ if let Err(e) = self.stream.send(Message::Pong(data)).await {
+ error!(
+ "Failed to send WebSocket pong to {}: {}",
+ self.client_address, e
+ );
+ return Err(IggyError::WebSocketSendError);
+ }
+ continue;
+ }
+ Some(Ok(Message::Pong(_))) => {
+ trace!("Received WebSocket pong from {}", self.client_address);
+ continue;
+ }
+ Some(Ok(Message::Close(_))) => {
+ debug!(
+ "WebSocket connection closed by client: {}",
+ self.client_address
+ );
+ return Err(IggyError::ConnectionClosed);
+ }
+ Some(Ok(Message::Frame(_))) => {
+ // Raw frames - just continue
+ continue;
+ }
+ Some(Err(e)) => {
+ error!(
+ "Failed to read WebSocket message from {}: {}",
+ self.client_address, e
+ );
+ return match e {
+ tokio_tungstenite::tungstenite::Error::ConnectionClosed
+ | tokio_tungstenite::tungstenite::Error::AlreadyClosed => {
+ Err(IggyError::ConnectionClosed)
+ }
+ tokio_tungstenite::tungstenite::Error::Io(io_err) => match io_err.kind() {
+ ErrorKind::UnexpectedEof
+ | ErrorKind::ConnectionAborted
+ | ErrorKind::ConnectionReset => Err(IggyError::ConnectionClosed),
+ _ => Err(IggyError::WebSocketReceiveError),
+ },
+ _ => Err(IggyError::WebSocketReceiveError),
+ };
+ }
+ None => {
+ debug!("WebSocket stream ended for client: {}", self.client_address);
+ return Err(IggyError::ConnectionClosed);
+ }
+ }
+ }
+ }
+}
+
+#[async_trait]
+impl ConnectionStream for WebSocketConnectionStream {
+ async fn read(&mut self, buf: &mut [u8]) -> Result<usize, IggyError> {
+ let requested_bytes = buf.len();
+
+ // Keep reading until we have enough bytes
+ while self.read_buffer.len() < requested_bytes {
+ self.read_message().await?;
+ }
+
+ // Copy exactly the requested bytes
+ buf.copy_from_slice(&self.read_buffer[..requested_bytes]);
+
+ // Remove consumed bytes from buffer
+ let _consumed = self.read_buffer.split_to(requested_bytes);
+
+ trace!(
+ "Read {} bytes from WebSocket stream for client: {}",
+ requested_bytes, self.client_address
+ );
+
+ debug!(
+ "WebSocket read {} bytes: {:02x?}",
+ requested_bytes,
+ &buf[..requested_bytes.min(16)] // Log first 16 bytes
+ );
+
+ Ok(requested_bytes)
+ }
+
+ async fn write(&mut self, buf: &[u8]) -> Result<(), IggyError> {
+ trace!(
+ "Writing {} bytes to WebSocket stream for client: {}",
+ buf.len(),
+ self.client_address
+ );
+
+ debug!(
+ "WebSocket write {} bytes: {:02x?}",
+ buf.len(),
+ &buf[..buf.len().min(16)]
+ );
+
+ self.stream
+ .send(Message::Binary(buf.to_vec().into()))
+ .await
+ .map_err(|e| {
+ error!(
+ "Failed to write data to WebSocket connection for client: {}: {}",
+ self.client_address, e
+ );
+ match e {
+ tokio_tungstenite::tungstenite::Error::ConnectionClosed
+ | tokio_tungstenite::tungstenite::Error::AlreadyClosed => {
+ IggyError::ConnectionClosed
+ }
+ tokio_tungstenite::tungstenite::Error::Io(io_err) => match io_err.kind() {
+ ErrorKind::BrokenPipe
+ | ErrorKind::ConnectionAborted
+ | ErrorKind::ConnectionReset => IggyError::ConnectionClosed,
+ _ => IggyError::WebSocketSendError,
+ },
+ _ => IggyError::WebSocketSendError,
+ }
+ })
+ }
+
+ async fn flush(&mut self) -> Result<(), IggyError> {
+ trace!(
+ "Flushing WebSocket stream for client: {}",
+ self.client_address
+ );
+ Ok(())
+ }
+
+ async fn shutdown(&mut self) -> Result<(), IggyError> {
+ debug!(
+ "Shutting down WebSocket connection for client: {}",
+ self.client_address
+ );
+
+ let close_message =
+ Message::Close(Some(tokio_tungstenite::tungstenite::protocol::CloseFrame {
+ code: tokio_tungstenite::tungstenite::protocol::frame::coding::CloseCode::Normal,
+ reason: "Client requested shutdown".into(),
+ }));
+
+ self.stream.send(close_message).await.map_err(|e| {
+ error!(
+ "Failed to send close frame to WebSocket connection for client: {}: {}",
+ self.client_address, e
+ );
+ IggyError::WebSocketCloseError
+ })
+ }
+}
diff --git a/core/sdk/src/client_wrappers/binary_cluster_client.rs b/core/sdk/src/websocket/websocket_stream.rs
similarity index 57%
copy from core/sdk/src/client_wrappers/binary_cluster_client.rs
copy to core/sdk/src/websocket/websocket_stream.rs
index 964c2a7e3..f51eddc56 100644
--- a/core/sdk/src/client_wrappers/binary_cluster_client.rs
+++ b/core/sdk/src/websocket/websocket_stream.rs
@@ -16,19 +16,22 @@
* under the License.
*/
-use crate::client_wrappers::client_wrapper::ClientWrapper;
use async_trait::async_trait;
-use iggy_binary_protocol::ClusterClient;
-use iggy_common::{ClusterMetadata, IggyError};
+use iggy_common::IggyError;
+/// Connection stream trait for WebSocket connections.
+/// Similar to the TCP ConnectionStream trait but adapted for WebSocket semantics.
#[async_trait]
-impl ClusterClient for ClientWrapper {
- async fn get_cluster_metadata(&self) -> Result<ClusterMetadata, IggyError> {
- match self {
- ClientWrapper::Iggy(client) => client.get_cluster_metadata().await,
- ClientWrapper::Http(client) => client.get_cluster_metadata().await,
- ClientWrapper::Quic(client) => client.get_cluster_metadata().await,
- ClientWrapper::Tcp(client) => client.get_cluster_metadata().await,
- }
- }
+pub trait ConnectionStream: Send + Sync {
+ /// Read exact amount of bytes from the stream.
+ async fn read(&mut self, buf: &mut [u8]) -> Result<usize, IggyError>;
+
+ /// Write all bytes to the stream.
+ async fn write(&mut self, buf: &[u8]) -> Result<(), IggyError>;
+
+ /// Flush any buffered data.
+ async fn flush(&mut self) -> Result<(), IggyError>;
+
+ /// Shutdown the connection.
+ async fn shutdown(&mut self) -> Result<(), IggyError>;
}
diff --git a/core/server/Cargo.toml b/core/server/Cargo.toml
index dc230fe1f..6301615e6 100644
--- a/core/server/Cargo.toml
+++ b/core/server/Cargo.toml
@@ -55,6 +55,9 @@ cyper-axum = { workspace = true }
compio-net = { workspace = true }
compio-quic = { workspace = true }
compio-tls = { workspace = true }
+compio-io = { workspace = true }
+compio-ws = { workspace = true }
+tungstenite = { workspace = true }
console-subscriber = { workspace = true, optional = true }
crossbeam = { workspace = true }
ctrlc = { version = "3.4", features = ["termination"] }
diff --git a/core/server/src/binary/mapper.rs b/core/server/src/binary/mapper.rs
index 6856dea2f..1d6fea71b 100644
--- a/core/server/src/binary/mapper.rs
+++ b/core/server/src/binary/mapper.rs
@@ -298,6 +298,7 @@ fn extend_client(client: &Client, bytes: &mut BytesMut) {
TransportProtocol::Tcp => 1,
TransportProtocol::Quic => 2,
TransportProtocol::Http => 3,
+ TransportProtocol::WebSocket => 4,
};
bytes.put_u8(transport);
let address = client.session.ip_address.to_string();
diff --git a/core/server/src/binary/sender.rs b/core/server/src/binary/sender.rs
index 2980726ed..eac21de20 100644
--- a/core/server/src/binary/sender.rs
+++ b/core/server/src/binary/sender.rs
@@ -19,6 +19,7 @@
use crate::streaming::utils::PooledBuffer;
use crate::tcp::tcp_sender::TcpSender;
use crate::tcp::tcp_tls_sender::TcpTlsSender;
+use crate::websocket::websocket_sender::WebSocketSender;
use crate::{quic::quic_sender::QuicSender, server_error::ServerError};
use compio::buf::IoBufMut;
use compio::net::TcpStream;
@@ -45,6 +46,7 @@ macro_rules! forward_async_methods {
Self::Tcp(d) => d.$method_name$(::<$($generic),+>)?($( $arg ),*).await,
Self::TcpTls(s) => s.$method_name$(::<$($generic),+>)?($( $arg ),*).await,
Self::Quic(s) => s.$method_name$(::<$($generic),+>)?($( $arg ),*).await,
+ Self::WebSocket(s) => s.$method_name$(::<$($generic),+>)?($( $arg ),*).await,
}
}
)*
@@ -72,6 +74,7 @@ pub enum SenderKind {
Tcp(TcpSender),
TcpTls(TcpTlsSender),
Quic(QuicSender),
+ WebSocket(WebSocketSender),
}
impl SenderKind {
@@ -90,6 +93,10 @@ impl SenderKind {
})
}
+ pub fn get_websocket_sender(stream: WebSocketSender) -> Self {
+ Self::WebSocket(stream)
+ }
+
forward_async_methods! {
async fn read<B: IoBufMut>(&mut self, buffer: B) -> (Result<(), IggyError>, B);
async fn send_empty_ok_response(&mut self) -> Result<(), IggyError>;
diff --git a/core/server/src/configs/defaults.rs b/core/server/src/configs/defaults.rs
index 36ae3c2a6..bb14600f9 100644
--- a/core/server/src/configs/defaults.rs
+++ b/core/server/src/configs/defaults.rs
@@ -35,6 +35,7 @@ use crate::configs::system::{
StateConfig, StreamConfig, SystemConfig, TopicConfig,
};
use crate::configs::tcp::{TcpConfig, TcpTlsConfig};
+use crate::configs::websocket::WebSocketConfig;
use iggy_common::IggyByteSize;
use iggy_common::IggyDuration;
use std::sync::Arc;
@@ -55,6 +56,7 @@ impl Default for ServerConfig {
system: Arc::new(SystemConfig::default()),
quic: QuicConfig::default(),
tcp: TcpConfig::default(),
+ websocket: WebSocketConfig::default(),
http: HttpConfig::default(),
telemetry: TelemetryConfig::default(),
cluster: ClusterConfig::default(),
@@ -171,6 +173,21 @@ impl Default for TcpSocketConfig {
}
}
+impl Default for WebSocketConfig {
+ fn default() -> WebSocketConfig {
+ WebSocketConfig {
+ enabled: SERVER_CONFIG.websocket.enabled,
+ address: SERVER_CONFIG.websocket.address.parse().unwrap(),
+ read_buffer_size: None,
+ write_buffer_size: None,
+ max_write_buffer_size: None,
+ max_message_size: None,
+ max_frame_size: None,
+ accept_unmasked_frames: false,
+ }
+ }
+}
+
impl Default for HttpConfig {
fn default() -> HttpConfig {
HttpConfig {
diff --git a/core/server/src/configs/mod.rs b/core/server/src/configs/mod.rs
index a6a65d491..ff8003afa 100644
--- a/core/server/src/configs/mod.rs
+++ b/core/server/src/configs/mod.rs
@@ -28,5 +28,6 @@ pub mod sharding;
pub mod system;
pub mod tcp;
pub mod validators;
+pub mod websocket;
pub const COMPONENT: &str = "CONFIG";
diff --git a/core/server/src/configs/server.rs b/core/server/src/configs/server.rs
index 003afe99d..83608bfcc 100644
--- a/core/server/src/configs/server.rs
+++ b/core/server/src/configs/server.rs
@@ -23,6 +23,7 @@ use crate::configs::http::HttpConfig;
use crate::configs::quic::QuicConfig;
use crate::configs::system::SystemConfig;
use crate::configs::tcp::TcpConfig;
+use crate::configs::websocket::WebSocketConfig;
use crate::server_error::ConfigError;
use derive_more::Display;
use error_set::ErrContext;
@@ -44,6 +45,7 @@ pub struct ServerConfig {
pub quic: QuicConfig,
pub tcp: TcpConfig,
pub http: HttpConfig,
+ pub websocket: WebSocketConfig,
pub telemetry: TelemetryConfig,
pub cluster: ClusterConfig,
}
diff --git a/core/server/src/configs/websocket.rs b/core/server/src/configs/websocket.rs
new file mode 100644
index 000000000..6d8a5b95b
--- /dev/null
+++ b/core/server/src/configs/websocket.rs
@@ -0,0 +1,107 @@
+/* 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.
+ */
+
+use iggy_common::IggyByteSize;
+use serde::{Deserialize, Serialize};
+use std::fmt::{Display, Formatter};
+use tungstenite::protocol::WebSocketConfig as TungsteniteConfig;
+
+#[derive(Debug, Deserialize, Serialize, Clone)]
+pub struct WebSocketConfig {
+ pub enabled: bool,
+ pub address: String,
+ #[serde(default)]
+ pub read_buffer_size: Option<String>,
+ #[serde(default)]
+ pub write_buffer_size: Option<String>,
+ #[serde(default)]
+ pub max_write_buffer_size: Option<String>,
+ #[serde(default)]
+ pub max_message_size: Option<String>,
+ #[serde(default)]
+ pub max_frame_size: Option<String>,
+ #[serde(default)]
+ pub accept_unmasked_frames: bool,
+}
+
+impl WebSocketConfig {
+ pub fn to_tungstenite_config(&self) -> Option<TungsteniteConfig> {
+ if self.read_buffer_size.is_none()
+ && self.write_buffer_size.is_none()
+ && self.max_write_buffer_size.is_none()
+ && self.max_message_size.is_none()
+ && self.max_frame_size.is_none()
+ && !self.accept_unmasked_frames
+ {
+ return None;
+ }
+
+ let mut config = TungsteniteConfig::default();
+
+ if let Some(read_buf_size_str) = &self.read_buffer_size {
+ if let Ok(byte_size) = read_buf_size_str.parse::<IggyByteSize>() {
+ config = config.read_buffer_size(byte_size.as_bytes_u64() as usize);
+ }
+ }
+
+ if let Some(write_buf_size_str) = &self.write_buffer_size {
+ if let Ok(byte_size) = write_buf_size_str.parse::<IggyByteSize>() {
+ config = config.write_buffer_size(byte_size.as_bytes_u64() as usize);
+ }
+ }
+
+ if let Some(max_write_buf_size_str) = &self.max_write_buffer_size {
+ if let Ok(byte_size) = max_write_buf_size_str.parse::<IggyByteSize>() {
+ config = config.max_write_buffer_size(byte_size.as_bytes_u64() as usize);
+ }
+ }
+
+ if let Some(msg_size_str) = &self.max_message_size {
+ if let Ok(byte_size) = msg_size_str.parse::<IggyByteSize>() {
+ config = config.max_message_size(Some(byte_size.as_bytes_u64() as usize));
+ }
+ }
+
+ if let Some(frame_size_str) = &self.max_frame_size {
+ if let Ok(byte_size) = frame_size_str.parse::<IggyByteSize>() {
+ config = config.max_frame_size(Some(byte_size.as_bytes_u64() as usize));
+ }
+ }
+
+ config = config.accept_unmasked_frames(self.accept_unmasked_frames);
+
+ Some(config)
+ }
+}
+
+impl Display for WebSocketConfig {
+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+ write!(
+ f,
+ "{{ enabled: {}, address: {}, read_buffer_size: {:?}, write_buffer_size: {:?}, max_write_buffer_size: {:?}, max_message_size: {:?}, max_frame_size: {:?}, accept_unmasked_frames: {} }}",
+ self.enabled,
+ self.address,
+ self.read_buffer_size,
+ self.write_buffer_size,
+ self.max_write_buffer_size,
+ self.max_message_size,
+ self.max_frame_size,
+ self.accept_unmasked_frames
+ )
+ }
+}
diff --git a/core/server/src/lib.rs b/core/server/src/lib.rs
index e24396ac9..9a9277ca7 100644
--- a/core/server/src/lib.rs
+++ b/core/server/src/lib.rs
@@ -42,6 +42,7 @@ pub mod state;
pub mod streaming;
pub mod tcp;
pub mod versioning;
+pub mod websocket;
pub const VERSION: &str = env!("CARGO_PKG_VERSION");
pub const IGGY_ROOT_USERNAME_ENV: &str = "IGGY_ROOT_USERNAME";
diff --git a/core/server/src/shard/builder.rs b/core/server/src/shard/builder.rs
index 5f25ce630..b3b5d5348 100644
--- a/core/server/src/shard/builder.rs
+++ b/core/server/src/shard/builder.rs
@@ -158,6 +158,7 @@ impl IggyShardBuilder {
is_shutting_down: AtomicBool::new(false),
tcp_bound_address: Cell::new(None),
quic_bound_address: Cell::new(None),
+ websocket_bound_address: Cell::new(None),
http_bound_address: Cell::new(None),
config_writer_notify,
config_writer_receiver,
diff --git a/core/server/src/shard/mod.rs b/core/server/src/shard/mod.rs
index 4bebc36c5..a4aeaffcc 100644
--- a/core/server/src/shard/mod.rs
+++ b/core/server/src/shard/mod.rs
@@ -145,6 +145,7 @@ pub struct IggyShard {
pub(crate) is_shutting_down: AtomicBool,
pub(crate) tcp_bound_address: Cell<Option<SocketAddr>>,
pub(crate) quic_bound_address: Cell<Option<SocketAddr>>,
+ pub websocket_bound_address: Cell<Option<SocketAddr>>,
pub(crate) http_bound_address: Cell<Option<SocketAddr>>,
config_writer_notify: async_channel::Sender<()>,
config_writer_receiver: async_channel::Receiver<()>,
@@ -190,6 +191,10 @@ impl IggyShard {
periodic::spawn_message_saver(self.clone());
}
+ if self.config.websocket.enabled && self.id == 0 {
+ continuous::spawn_websocket_server(self.clone());
+ }
+
if self.config.heartbeat.enabled {
periodic::spawn_heartbeat_verifier(self.clone());
}
@@ -862,6 +867,11 @@ impl IggyShard {
// Notify config writer that a server has bound
let _ = self.config_writer_notify.try_send(());
}
+ TransportProtocol::WebSocket => {
+ self.websocket_bound_address.set(Some(address));
+ // Notify config writer that a server has bound
+ let _ = self.config_writer_notify.try_send(());
+ }
}
Ok(())
}
@@ -1014,6 +1024,10 @@ impl IggyShard {
)?;
Ok(())
}
+ ShardEvent::WebSocketBound { address } => {
+ self.websocket_bound_address.set(Some(address));
+ Ok(())
+ }
ShardEvent::LeftConsumerGroup {
client_id,
stream_id,
diff --git a/core/server/src/shard/tasks/continuous/mod.rs b/core/server/src/shard/tasks/continuous/mod.rs
index f8ae5fbc2..6edc3551a 100644
--- a/core/server/src/shard/tasks/continuous/mod.rs
+++ b/core/server/src/shard/tasks/continuous/mod.rs
@@ -20,8 +20,9 @@ mod http_server;
mod message_pump;
mod quic_server;
mod tcp_server;
-
+mod websocket_server;
pub use http_server::spawn_http_server;
pub use message_pump::spawn_message_pump;
pub use quic_server::spawn_quic_server;
pub use tcp_server::spawn_tcp_server;
+pub use websocket_server::spawn_websocket_server;
diff --git a/core/bench/report/src/types/transport.rs b/core/server/src/shard/tasks/continuous/websocket_server.rs
similarity index 57%
copy from core/bench/report/src/types/transport.rs
copy to core/server/src/shard/tasks/continuous/websocket_server.rs
index 73f036646..1c18325ce 100644
--- a/core/bench/report/src/types/transport.rs
+++ b/core/server/src/shard/tasks/continuous/websocket_server.rs
@@ -16,19 +16,24 @@
* under the License.
*/
-use derive_more::derive::Display;
-use serde::{Deserialize, Serialize};
+use crate::shard::IggyShard;
+use crate::shard::task_registry::ShutdownToken;
+use iggy_common::IggyError;
+use std::rc::Rc;
-#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Display, Default, Hash)]
-pub enum BenchmarkTransport {
- #[default]
- #[display("TCP")]
- #[serde(rename = "tcp")]
- Tcp,
- #[display("HTTP")]
- #[serde(rename = "http")]
- Http,
- #[display("QUIC")]
- #[serde(rename = "quic")]
- Quic,
+pub fn spawn_websocket_server(shard: Rc<IggyShard>) {
+ let shard_clone = shard.clone();
+ shard
+ .task_registry
+ .continuous("websocket_server")
+ .critical(true)
+ .run(move |shutdown| websocket_server_task(shard_clone, shutdown))
+ .spawn();
+}
+
+async fn websocket_server_task(
+ shard: Rc<IggyShard>,
+ shutdown: ShutdownToken,
+) -> Result<(), IggyError> {
+ crate::websocket::websocket_server::spawn_websocket_server(shard, shutdown).await
}
diff --git a/core/server/src/shard/transmission/event.rs b/core/server/src/shard/transmission/event.rs
index 79c30edff..13ee295f7 100644
--- a/core/server/src/shard/transmission/event.rs
+++ b/core/server/src/shard/transmission/event.rs
@@ -150,4 +150,7 @@ pub enum ShardEvent {
protocol: TransportProtocol,
address: SocketAddr,
},
+ WebSocketBound {
+ address: SocketAddr,
+ },
}
diff --git a/core/server/src/websocket/connection_handler.rs b/core/server/src/websocket/connection_handler.rs
new file mode 100644
index 000000000..8f100d8c2
--- /dev/null
+++ b/core/server/src/websocket/connection_handler.rs
@@ -0,0 +1,140 @@
+/* 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.
+ */
+
+use crate::binary::command::ServerCommandHandler;
+use crate::binary::{command, sender::SenderKind};
+use crate::server_error::ConnectionError;
+use crate::shard::IggyShard;
+use crate::streaming::session::Session;
+use crate::websocket::connection_handler::command::ServerCommand;
+use async_channel::Receiver;
+use bytes::BytesMut;
+use futures::FutureExt;
+use iggy_common::IggyError;
+use std::io::ErrorKind;
+use std::rc::Rc;
+use tracing::{debug, error, info};
+
+const INITIAL_BYTES_LENGTH: usize = 4;
+
+pub(crate) async fn handle_connection(
+ session: &Rc<Session>,
+ sender: &mut SenderKind,
+ shard: &Rc<IggyShard>,
+ stop_receiver: Receiver<()>,
+) -> Result<(), ConnectionError> {
+ let mut length_buffer = BytesMut::with_capacity(INITIAL_BYTES_LENGTH);
+ let mut code_buffer = BytesMut::with_capacity(INITIAL_BYTES_LENGTH);
+
+ loop {
+ let read_future = sender.read(length_buffer);
+ let (_, mut initial_buffer) = futures::select! {
+ _ = stop_receiver.recv().fuse() => {
+ info!("Connection stop signal received for session: {}", session);
+ let _ = sender.send_error_response(IggyError::Disconnected).await;
+ return Ok(());
+ }
+ result = read_future.fuse() => {
+ match result {
+ (Ok(read_length), initial_buffer) => (read_length, initial_buffer),
+ (Err(error), initial_buffer) => {
+ length_buffer = initial_buffer;
+ if error.as_code() == IggyError::ConnectionClosed.as_code() {
+ return Err(ConnectionError::from(error));
+ } else {
+ error!("got error: {:?}", error);
+ sender.send_error_response(error).await?;
+ continue;
+ }
+ }
+ }
+ }
+ };
+
+ let length =
+ u32::from_le_bytes(initial_buffer[0..INITIAL_BYTES_LENGTH].try_into().unwrap());
+ let (res, mut code_buffer_out) = sender.read(code_buffer).await;
+ let _ = res?;
+ let code: u32 =
+ u32::from_le_bytes(code_buffer_out[0..INITIAL_BYTES_LENGTH].try_into().unwrap());
+
+ initial_buffer.clear();
+ code_buffer_out.clear();
+ length_buffer = BytesMut::from(initial_buffer);
+ code_buffer = BytesMut::from(code_buffer_out);
+
+ debug!("Received a WebSocket request, length: {length}, code: {code}");
+ let command = ServerCommand::from_code_and_reader(code, sender, length - 4).await?;
+ debug!("Received a WebSocket command: {command}, payload size: {length}");
+
+ match command.handle(sender, length, session, shard).await {
+ Ok(_) => {
+ debug!(
+ "Command was handled successfully, session: {session}. WebSocket response was sent."
+ );
+ }
+ Err(error) => {
+ error!("Command was not handled successfully, session: {session}, error: {error}.");
+ if let IggyError::ClientNotFound(_) = error {
+ sender.send_error_response(error).await?;
+ debug!("WebSocket error response was sent to: {session}.");
+ error!("Session: {session} will be deleted.");
+ return Err(ConnectionError::from(IggyError::ClientNotFound(
+ session.client_id,
+ )));
+ } else {
+ sender.send_error_response(error).await?;
+ debug!("WebSocket error response was sent to: {session}.");
+ }
+ }
+ }
+ }
+}
+
+pub(crate) fn handle_error(error: ConnectionError) {
+ match error {
+ ConnectionError::IoError(error) => match error.kind() {
+ ErrorKind::UnexpectedEof => {
+ info!("WebSocket connection has been closed.");
+ }
+ ErrorKind::ConnectionAborted => {
+ info!("WebSocket connection has been aborted.");
+ }
+ ErrorKind::ConnectionRefused => {
+ info!("WebSocket connection has been refused.");
+ }
+ ErrorKind::ConnectionReset => {
+ info!("WebSocket connection has been reset.");
+ }
+ _ => {
+ error!("WebSocket connection has failed: {error}");
+ }
+ },
+ ConnectionError::SdkError(sdk_error) => match sdk_error {
+ IggyError::ConnectionClosed => {
+ debug!("Client closed WebSocket connection.");
+ }
+ _ => {
+ error!("Failure in internal SDK call: {sdk_error}");
+ }
+ },
+ _ => {
+ error!("WebSocket connection has failed: {error}");
+ }
+ }
+}
diff --git a/core/server/src/websocket/mod.rs b/core/server/src/websocket/mod.rs
new file mode 100644
index 000000000..745b48cf2
--- /dev/null
+++ b/core/server/src/websocket/mod.rs
@@ -0,0 +1,6 @@
+pub mod connection_handler;
+pub mod websocket_listener;
+pub mod websocket_sender;
+pub mod websocket_server;
+
+pub const COMPONENT: &str = "WEBSOCKET";
diff --git a/core/server/src/websocket/websocket_listener.rs b/core/server/src/websocket/websocket_listener.rs
new file mode 100644
index 000000000..760a87c15
--- /dev/null
+++ b/core/server/src/websocket/websocket_listener.rs
@@ -0,0 +1,163 @@
+/* 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.
+ */
+
+use crate::binary::sender::SenderKind;
+use crate::configs::websocket::WebSocketConfig;
+use crate::shard::IggyShard;
+use crate::shard::task_registry::ShutdownToken;
+use crate::shard::transmission::event::ShardEvent;
+use crate::websocket::connection_handler::{handle_connection, handle_error};
+use crate::websocket::websocket_sender::WebSocketSender;
+use crate::{shard_debug, shard_error, shard_info};
+use compio::net::TcpListener;
+use compio_ws::accept_async_with_config;
+use error_set::ErrContext;
+use futures::FutureExt;
+use iggy_common::IggyError;
+use iggy_common::TransportProtocol;
+use std::net::SocketAddr;
+use std::rc::Rc;
+use tracing::{error, info};
+
+pub async fn start(
+ config: WebSocketConfig,
+ shard: Rc<IggyShard>,
+ shutdown: ShutdownToken,
+) -> Result<(), IggyError> {
+ let addr: SocketAddr = config
+ .address
+ .parse()
+ .with_error_context(|error| {
+ format!(
+ "WebSocket (error: {error}) - failed to parse address: {}",
+ config.address
+ )
+ })
+ .map_err(|_| IggyError::InvalidConfiguration)?;
+
+ let listener = TcpListener::bind(addr)
+ .await
+ .with_error_context(|error| {
+ format!("WebSocket (error: {error}) - failed to bind to address: {addr}")
+ })
+ .map_err(|_| IggyError::CannotBindToSocket(addr.to_string()))?;
+
+ let local_addr = listener.local_addr().unwrap();
+ shard_info!(
+ shard.id,
+ "{} has started on: ws://{}",
+ "WebSocket Server",
+ local_addr
+ );
+
+ if shard.id == 0 {
+ let event = ShardEvent::WebSocketBound {
+ address: local_addr,
+ };
+ shard.broadcast_event_to_all_shards(event).await;
+ }
+
+ let ws_config = config.to_tungstenite_config();
+ shard_info!(
+ shard.id,
+ "WebSocket config: max_message_size: {:?}, max_frame_size: {:?}, accept_unmasked_frames: {}",
+ config.max_message_size,
+ config.max_frame_size,
+ config.accept_unmasked_frames
+ );
+
+ accept_loop(listener, ws_config, shard, shutdown).await
+}
+
+async fn accept_loop(
+ listener: TcpListener,
+ ws_config: Option<compio_ws::WebSocketConfig>,
+ shard: Rc<IggyShard>,
+ shutdown: ShutdownToken,
+) -> Result<(), IggyError> {
+ loop {
+ let shard = shard.clone();
+ let ws_config = ws_config.clone();
+ let accept_future = listener.accept();
+
+ futures::select! {
+ _ = shutdown.wait().fuse() => {
+ shard_debug!(shard.id, "WebSocket Server received shutdown signal, no longer accepting connections");
+ break;
+ }
+ result = accept_future.fuse() => {
+ match result {
+ Ok((tcp_stream, remote_addr)) => {
+ if shard.is_shutting_down() {
+ shard_info!(shard.id, "Rejecting new WebSocket connection from {} during shutdown", remote_addr);
+ continue;
+ }
+ shard_info!(shard.id, "Accepted new WebSocket connection from: {}", remote_addr);
+
+ let shard_clone = shard.clone();
+ let ws_config_clone = ws_config.clone();
+ let registry = shard.task_registry.clone();
+ let registry_clone = registry.clone();
+
+ registry.spawn_connection(async move {
+ match accept_async_with_config(tcp_stream, ws_config_clone).await {
+ Ok(websocket) => {
+ info!("WebSocket handshake successful from: {}", remote_addr);
+
+ let session = shard_clone.add_client(&remote_addr, TransportProtocol::WebSocket);
+ let client_id = session.client_id;
+ shard_clone.add_active_session(session.clone());
+
+ let event = ShardEvent::NewSession {
+ address: remote_addr,
+ transport: TransportProtocol::WebSocket,
+ };
+ let _ = shard_clone.broadcast_event_to_all_shards(event).await;
+
+ let sender = WebSocketSender::new(websocket);
+ let mut sender_kind = SenderKind::get_websocket_sender(sender);
+ let client_stop_receiver = registry_clone.add_connection(client_id);
+
+ if let Err(error) = handle_connection(&session, &mut sender_kind, &shard_clone, client_stop_receiver).await {
+ handle_error(error);
+ }
+ registry_clone.remove_connection(&client_id);
+
+ if let Err(error) = sender_kind.shutdown().await {
+ shard_error!(shard_clone.id, "Failed to shutdown WebSocket stream for client: {}, address: {}. {}", client_id, remote_addr, error);
+ } else {
+ shard_info!(shard_clone.id, "Successfully closed WebSocket stream for client: {}, address: {}.", client_id, remote_addr);
+ }
+ }
+ Err(error) => {
+ error!("WebSocket handshake failed from {}: {:?}", remote_addr, error);
+ }
+ }
+ });
+ }
+ Err(error) => {
+ shard_error!(shard.id, "Failed to accept WebSocket connection: {}", error);
+ }
+ }
+ }
+ }
+ }
+
+ shard_info!(shard.id, "WebSocket Server listener has stopped");
+ Ok(())
+}
diff --git a/core/server/src/websocket/websocket_sender.rs b/core/server/src/websocket/websocket_sender.rs
new file mode 100644
index 000000000..d83042bc5
--- /dev/null
+++ b/core/server/src/websocket/websocket_sender.rs
@@ -0,0 +1,176 @@
+/* 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.
+ */
+
+use crate::binary::sender::Sender;
+use crate::server_error::ServerError;
+use crate::streaming::utils::PooledBuffer;
+use bytes::{BufMut, BytesMut};
+use compio::buf::IoBufMut;
+use compio::net::TcpStream;
+use compio_ws::TungsteniteError;
+use compio_ws::{WebSocketMessage as Message, WebSocketStream};
+use iggy_common::IggyError;
+use std::ptr;
+use tracing::debug;
+
+const READ_BUFFER_CAPACITY: usize = 8192;
+const WRITE_BUFFER_CAPACITY: usize = 8192;
+const STATUS_OK: &[u8] = &[0; 4];
+
+pub struct WebSocketSender {
+ pub(crate) stream: WebSocketStream<TcpStream>,
+ pub(crate) read_buffer: BytesMut,
+ pub(crate) write_buffer: BytesMut,
+}
+
+impl WebSocketSender {
+ pub fn new(stream: WebSocketStream<TcpStream>) -> Self {
+ Self {
+ stream,
+ read_buffer: BytesMut::with_capacity(READ_BUFFER_CAPACITY),
+ write_buffer: BytesMut::with_capacity(WRITE_BUFFER_CAPACITY),
+ }
+ }
+
+ async fn flush_write_buffer(&mut self) -> Result<(), IggyError> {
+ if self.write_buffer.is_empty() {
+ return Ok(());
+ }
+ let data = self.write_buffer.split().freeze();
+ self.stream
+ .send(Message::Binary(data))
+ .await
+ .map_err(|_| IggyError::TcpError)
+ }
+}
+
+impl Sender for WebSocketSender {
+ async fn read<B: IoBufMut>(&mut self, mut buffer: B) -> (Result<(), IggyError>, B) {
+ let required_len = buffer.buf_capacity();
+ if required_len == 0 {
+ return (Ok(()), buffer);
+ }
+
+ while self.read_buffer.len() < required_len {
+ match self.stream.read().await {
+ Ok(Message::Binary(data)) => {
+ self.read_buffer.extend_from_slice(&data);
+ }
+ Ok(Message::Close(_)) => {
+ return (Err(IggyError::ConnectionClosed), buffer);
+ }
+ Ok(Message::Ping(data)) => {
+ if self.stream.send(Message::Pong(data)).await.is_err() {
+ return (Err(IggyError::ConnectionClosed), buffer);
+ }
+ }
+ Ok(_) => { /* Ignore other message types */ }
+ Err(_) => {
+ return (Err(IggyError::ConnectionClosed), buffer);
+ }
+ }
+ }
+
+ let data_to_copy = self.read_buffer.split_to(required_len);
+
+ unsafe {
+ ptr::copy_nonoverlapping(data_to_copy.as_ptr(), buffer.as_buf_mut_ptr(), required_len);
+ buffer.set_buf_init(required_len);
+ }
+
+ (Ok(()), buffer)
+ }
+
+ async fn send_empty_ok_response(&mut self) -> Result<(), IggyError> {
+ self.send_ok_response(&[]).await
+ }
+
+ async fn send_ok_response(&mut self, payload: &[u8]) -> Result<(), IggyError> {
+ debug!(
+ "Sending WebSocket response with status: OK, payload length: {}",
+ payload.len()
+ );
+
+ let length = (payload.len() as u32).to_le_bytes();
+ let total_size = STATUS_OK.len() + length.len() + payload.len();
+
+ if self.write_buffer.len() + total_size > self.write_buffer.capacity() {
+ self.flush_write_buffer().await?;
+ }
+
+ self.write_buffer.put_slice(STATUS_OK);
+ self.write_buffer.put_slice(&length);
+ self.write_buffer.put_slice(payload);
+
+ self.flush_write_buffer().await
+ }
+
+ async fn send_error_response(&mut self, error: IggyError) -> Result<(), IggyError> {
+ let status = &error.as_code().to_le_bytes();
+ debug!("Sending WebSocket error response with status: {:?}", status);
+ let length = 0u32.to_le_bytes();
+ let total_size = status.len() + length.len();
+
+ if self.write_buffer.len() + total_size > self.write_buffer.capacity() {
+ self.flush_write_buffer().await?;
+ }
+ self.write_buffer.put_slice(status);
+ self.write_buffer.put_slice(&length);
+ self.flush_write_buffer().await
+ }
+
+ async fn shutdown(&mut self) -> Result<(), ServerError> {
+ self.flush_write_buffer().await.map_err(ServerError::from)?;
+
+ match self.stream.close(None).await {
+ Ok(_) => Ok(()),
+ Err(e) => match e {
+ TungsteniteError::ConnectionClosed | TungsteniteError::AlreadyClosed => {
+ debug!("WebSocket connection already closed: {}", e);
+ Ok(())
+ }
+ _ => Err(ServerError::from(
+ IggyError::CannotCloseWebSocketConnection(format!("{}", e)),
+ )),
+ },
+ }
+ }
+
+ async fn send_ok_response_vectored(
+ &mut self,
+ length: &[u8],
+ slices: Vec<PooledBuffer>,
+ ) -> Result<(), IggyError> {
+ self.flush_write_buffer().await?;
+
+ let total_payload_size = slices.iter().map(|s| s.len()).sum::<usize>();
+ let total_size = STATUS_OK.len() + length.len() + total_payload_size;
+
+ let mut response_bytes = BytesMut::with_capacity(total_size);
+ response_bytes.put_slice(STATUS_OK);
+ response_bytes.put_slice(length);
+ for slice in slices {
+ response_bytes.put_slice(&slice);
+ }
+
+ self.stream
+ .send(Message::Binary(response_bytes.freeze()))
+ .await
+ .map_err(|_| IggyError::TcpError)
+ }
+}
diff --git a/core/server/src/websocket/websocket_server.rs b/core/server/src/websocket/websocket_server.rs
new file mode 100644
index 000000000..6edfe7a13
--- /dev/null
+++ b/core/server/src/websocket/websocket_server.rs
@@ -0,0 +1,53 @@
+/* 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.
+ */
+
+use crate::shard::IggyShard;
+use crate::shard::task_registry::ShutdownToken;
+use crate::websocket::websocket_listener::start;
+use crate::{shard_error, shard_info};
+use iggy_common::IggyError;
+use std::rc::Rc;
+
+pub async fn spawn_websocket_server(
+ shard: Rc<IggyShard>,
+ shutdown: ShutdownToken,
+) -> Result<(), IggyError> {
+ let config = shard.config.websocket.clone();
+
+ if !config.enabled {
+ shard_info!(shard.id, "WebSocket server is disabled.");
+ return Ok(());
+ }
+
+ shard_info!(
+ shard.id,
+ "Starting WebSocket server on: {} for shard: {}...",
+ config.address,
+ shard.id
+ );
+
+ if let Err(error) = start(config, shard.clone(), shutdown).await {
+ shard_error!(
+ shard.id,
+ "WebSocket server has failed to start, error: {error}"
+ );
+ return Err(error);
+ }
+
+ Ok(())
+}
diff --git a/examples/rust/src/shared/args.rs b/examples/rust/src/shared/args.rs
index a13db47de..aa5eb47f6 100644
--- a/examples/rust/src/shared/args.rs
+++ b/examples/rust/src/shared/args.rs
@@ -158,6 +158,24 @@ pub struct Args {
#[arg(long, default_value = "5s")]
pub quic_heartbeat_interval: String,
+
+ #[arg(long, default_value = "127.0.0.1:8092")]
+ pub websocket_server_address: String,
+
+ #[arg(long, default_value = "true")]
+ pub websocket_reconnection_enabled: bool,
+
+ #[arg(long)]
+ pub websocket_reconnection_max_retries: Option<u32>,
+
+ #[arg(long, default_value = "1s")]
+ pub websocket_reconnection_interval: String,
+
+ #[arg(long, default_value = "5s")]
+ pub websocket_reconnection_reestablish_after: String,
+
+ #[arg(long, default_value = "5s")]
+ pub websocket_heartbeat_interval: String,
}
impl Args {
@@ -219,6 +237,12 @@ impl Default for Args {
quic_max_idle_timeout: 10000,
quic_validate_certificate: false,
quic_heartbeat_interval: "5s".to_string(),
+ websocket_server_address: "127.0.0.1:8092".to_string(),
+ websocket_reconnection_enabled: true,
+ websocket_reconnection_max_retries: None,
+ websocket_reconnection_interval: "1s".to_string(),
+ websocket_reconnection_reestablish_after: "5s".to_string(),
+ websocket_heartbeat_interval: "5s".to_string(),
}
}
}
@@ -312,6 +336,14 @@ impl Args {
quic_max_idle_timeout: self.quic_max_idle_timeout,
quic_validate_certificate: self.quic_validate_certificate,
quic_heartbeat_interval: self.quic_heartbeat_interval.clone(),
+ websocket_server_address: self.websocket_server_address.clone(),
+ websocket_reconnection_enabled: self.websocket_reconnection_enabled,
+ websocket_reconnection_max_retries: self.websocket_reconnection_max_retries,
+ websocket_reconnection_interval: self.websocket_reconnection_interval.clone(),
+ websocket_reconnection_reestablish_after: self
+ .websocket_reconnection_reestablish_after
+ .clone(),
+ websocket_heartbeat_interval: self.websocket_heartbeat_interval.clone(),
}
}