You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by hg...@apache.org on 2026/05/22 12:08:47 UTC
(iggy) branch common-deps updated (e23a47e99 -> b85c2c04f)
This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a change to branch common-deps
in repository https://gitbox.apache.org/repos/asf/iggy.git
omit e23a47e99 refactor(server): extract server-only types into new crate `server_common`
new b85c2c04f refactor(server): extract server-only types into new crate `server_common`
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (e23a47e99)
\
N -- N -- N refs/heads/common-deps (b85c2c04f)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
Cargo.lock | 8 +++++---
core/common/Cargo.toml | 3 ---
core/common/src/lib.rs | 5 -----
core/configs/src/server_ng_config/server_ng.rs | 6 ++----
core/configs/src/server_ng_config/validators.rs | 2 +-
core/consensus/src/client_table.rs | 2 +-
core/consensus/src/impls.rs | 2 +-
core/consensus/src/observability.rs | 2 +-
core/message_bus/src/transports/tls/mod.rs | 2 +-
core/partitions/src/iggy_partition.rs | 5 +++--
core/partitions/src/iggy_partitions.rs | 2 +-
core/server-ng/src/bootstrap.rs | 8 ++++----
core/server/src/binary/handlers/messages/send_messages_handler.rs | 2 +-
.../src/binary/handlers/segments/delete_segments_handler.rs | 2 +-
core/server/src/http/segments.rs | 2 +-
core/server/src/main.rs | 2 +-
core/server/src/metadata/reader.rs | 2 +-
core/server/src/quic/quic_server.rs | 2 +-
core/server/src/shard/builder.rs | 2 +-
core/server/src/shard/communication.rs | 2 +-
core/server/src/shard/handlers.rs | 3 ++-
core/server/src/shard/mod.rs | 2 +-
core/server/src/shard/system/consumer_offsets.rs | 5 ++---
core/server/src/shard/system/messages.rs | 2 +-
core/server/src/shard/system/partitions.rs | 4 ++--
core/server/src/shard/system/segments.rs | 2 +-
core/server/src/shard/system/streams.rs | 2 +-
core/server/src/shard/system/topics.rs | 2 +-
core/server/src/shard/tasks/periodic/message_cleaner.rs | 2 +-
core/server/src/shard/transmission/message.rs | 2 +-
core/server/src/streaming/deduplication/mod.rs | 2 +-
core/server/src/streaming/partitions/local_partitions.rs | 2 +-
core/server/src/streaming/partitions/ops.rs | 2 +-
core/server/src/streaming/partitions/ops_tests.rs | 2 +-
core/server/src/tcp/tcp_tls_listener.rs | 2 +-
core/server/src/websocket/websocket_tls_listener.rs | 2 +-
core/server_common/Cargo.toml | 5 +++++
core/{common => server_common}/src/certificates.rs | 0
.../src/deduplication/message_deduplicator.rs | 2 +-
core/{common => server_common}/src/deduplication/mod.rs | 0
core/server_common/src/lib.rs | 5 +++++
core/server_common/src/messages_batch_mut.rs | 5 +++--
core/server_common/src/send_messages2.rs | 3 ++-
core/{common => server_common}/src/sharding/local_idx.rs | 0
core/{common => server_common}/src/sharding/mod.rs | 0
core/{common => server_common}/src/sharding/namespace.rs | 0
core/{common => server_common}/src/sharding/partition_location.rs | 0
core/{common => server_common}/src/sharding/shard_id.rs | 0
core/shard/src/lib.rs | 3 ++-
core/shard/src/router.rs | 2 +-
core/shard/src/shards_table.rs | 2 +-
core/simulator/src/client.rs | 2 +-
core/simulator/src/lib.rs | 4 ++--
core/simulator/src/main.rs | 2 +-
core/simulator/src/replica.rs | 2 +-
55 files changed, 72 insertions(+), 66 deletions(-)
rename core/{common => server_common}/src/certificates.rs (100%)
rename core/{common => server_common}/src/deduplication/message_deduplicator.rs (99%)
rename core/{common => server_common}/src/deduplication/mod.rs (100%)
rename core/{common => server_common}/src/sharding/local_idx.rs (100%)
rename core/{common => server_common}/src/sharding/mod.rs (100%)
rename core/{common => server_common}/src/sharding/namespace.rs (100%)
rename core/{common => server_common}/src/sharding/partition_location.rs (100%)
rename core/{common => server_common}/src/sharding/shard_id.rs (100%)
(iggy) 01/01: refactor(server): extract server-only types into new crate `server_common`
Posted by hg...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch common-deps
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit b85c2c04f28509135260684e910961a11f206d07
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Fri May 22 11:32:52 2026 +0200
refactor(server): extract server-only types into new crate `server_common`
SDK users transitively pulled the compio runtime (io-uring, quic,
tls, ws, fs) plus aligned-vec, smallvec, compio-buf, moka, rcgen
and rustls through iggy_common and iggy_binary_protocol because
both crates housed server-only primitives: PooledBuffer /
MemoryPool, segment_storage, mutable in-flight message types,
TCP/QUIC/WS sender helpers, Message / *Backing / MessageBag /
ConsensusMessage, iobuf (Owned, Prefix, Frozen), MESSAGE_ALIGN,
send_messages2, sharding (IggyNamespace, ShardId, LocalIdx,
PartitionLocation), MessageDeduplicator and the self-signed
certificate helper. The runtime ballooned SDK build time and dep
surface for no SDK gain.
Introduce server_common (publish = false) holding all of the above.
Move sender/* into core/server. Drop compio, compio-buf, aligned-vec,
smallvec, crossbeam, human-repr, err_trail, once_cell, nix,
serial_test, moka, rcgen and rustls from the SDK side.
iggy_binary_protocol becomes a pure sans-io wire codec. Promote
IggyMessageBoundaries and IggyMessageViewIterator::new_with_boundaries
to pub for cross-crate access. Update server, server-ng, simulator,
partitions, configs to depend on the new crate.
---
Cargo.lock | 52 ++++++++++++++-----
Cargo.toml | 2 +
core/binary_protocol/Cargo.toml | 6 +--
core/binary_protocol/src/consensus/mod.rs | 22 --------
core/binary_protocol/src/lib.rs | 13 +++--
core/common/Cargo.toml | 15 ------
core/common/src/alloc/mod.rs | 20 --------
core/common/src/lib.rs | 14 -----
.../common/src/types/message/message_boundaries.rs | 8 +--
core/common/src/types/message/message_view.rs | 2 +-
core/common/src/types/message/mod.rs | 11 +---
core/common/src/types/mod.rs | 2 -
core/configs/Cargo.toml | 1 +
core/configs/src/server_config/server.rs | 3 +-
core/configs/src/server_ng_config/server_ng.rs | 6 +--
core/configs/src/server_ng_config/validators.rs | 2 +-
core/consensus/Cargo.toml | 1 +
core/consensus/src/client_table.rs | 7 ++-
core/consensus/src/impls.rs | 10 ++--
core/consensus/src/lib.rs | 3 +-
core/consensus/src/metadata_helpers.rs | 6 +--
core/consensus/src/observability.rs | 2 +-
core/consensus/src/plane_helpers.rs | 9 ++--
core/journal/Cargo.toml | 1 +
core/journal/src/prepare_journal.rs | 3 +-
core/message_bus/Cargo.toml | 1 +
core/message_bus/src/framing.rs | 5 +-
core/message_bus/src/installer/replica.rs | 3 +-
core/message_bus/src/installer/tcp.rs | 3 +-
core/message_bus/src/lib.rs | 5 +-
.../src/lifecycle/connection_registry.rs | 6 +--
core/message_bus/src/transports/mod.rs | 3 +-
core/message_bus/src/transports/quic.rs | 7 +--
core/message_bus/src/transports/tcp.rs | 11 ++--
core/message_bus/src/transports/tcp_tls.rs | 10 ++--
core/message_bus/src/transports/tls/mod.rs | 2 +-
core/message_bus/src/transports/ws.rs | 12 ++---
core/message_bus/src/transports/wss.rs | 7 +--
core/message_bus/tests/common/mod.rs | 3 +-
core/message_bus/tests/quic_client_roundtrip.rs | 5 +-
core/message_bus/tests/tcp_tls_client_listener.rs | 5 +-
core/message_bus/tests/tcp_tls_client_roundtrip.rs | 5 +-
core/message_bus/tests/ws_client_roundtrip.rs | 5 +-
core/message_bus/tests/wss_client_listener.rs | 5 +-
core/message_bus/tests/wss_client_roundtrip.rs | 5 +-
core/metadata/Cargo.toml | 1 +
core/metadata/src/impls/metadata.rs | 7 ++-
core/metadata/src/impls/recovery.rs | 4 +-
core/metadata/src/stm/mod.rs | 2 +-
core/metadata/src/stm/mux.rs | 6 ++-
core/partitions/Cargo.toml | 1 +
core/partitions/src/iggy_partition.rs | 18 ++++---
core/partitions/src/iggy_partitions.rs | 2 +-
core/partitions/src/journal.rs | 9 ++--
core/partitions/src/lib.rs | 5 +-
core/partitions/src/log.rs | 3 +-
core/partitions/src/messages_writer.rs | 2 +-
core/partitions/src/types.rs | 2 +-
core/server-ng/Cargo.toml | 1 +
core/server-ng/src/bootstrap.rs | 10 ++--
core/server-ng/src/login_register.rs | 3 +-
core/server-ng/src/main.rs | 2 +-
core/server/Cargo.toml | 1 +
core/server/src/binary/dispatch.rs | 5 +-
.../cluster/get_cluster_metadata_handler.rs | 3 +-
.../create_consumer_group_handler.rs | 3 +-
.../delete_consumer_group_handler.rs | 3 +-
.../consumer_groups/get_consumer_group_handler.rs | 2 +-
.../consumer_groups/get_consumer_groups_handler.rs | 3 +-
.../consumer_groups/join_consumer_group_handler.rs | 3 +-
.../leave_consumer_group_handler.rs | 3 +-
.../delete_consumer_offset_handler.rs | 3 +-
.../get_consumer_offset_handler.rs | 2 +-
.../store_consumer_offset_handler.rs | 2 +-
.../messages/flush_unsaved_buffer_handler.rs | 3 +-
.../handlers/messages/poll_messages_handler.rs | 5 +-
.../handlers/messages/send_messages_handler.rs | 6 +--
.../partitions/create_partitions_handler.rs | 3 +-
.../partitions/delete_partitions_handler.rs | 3 +-
.../create_personal_access_token_handler.rs | 3 +-
.../delete_personal_access_token_handler.rs | 3 +-
.../get_personal_access_tokens_handler.rs | 2 +-
.../login_with_personal_access_token_handler.rs | 3 +-
.../handlers/segments/delete_segments_handler.rs | 5 +-
.../handlers/streams/create_stream_handler.rs | 3 +-
.../handlers/streams/delete_stream_handler.rs | 3 +-
.../binary/handlers/streams/get_stream_handler.rs | 2 +-
.../binary/handlers/streams/get_streams_handler.rs | 2 +-
.../handlers/streams/purge_stream_handler.rs | 3 +-
.../handlers/streams/update_stream_handler.rs | 3 +-
.../binary/handlers/system/get_client_handler.rs | 2 +-
.../binary/handlers/system/get_clients_handler.rs | 2 +-
.../src/binary/handlers/system/get_me_handler.rs | 2 +-
.../binary/handlers/system/get_snapshot_handler.rs | 3 +-
.../binary/handlers/system/get_stats_handler.rs | 2 +-
.../src/binary/handlers/system/ping_handler.rs | 2 +-
.../binary/handlers/topics/create_topic_handler.rs | 3 +-
.../binary/handlers/topics/delete_topic_handler.rs | 3 +-
.../binary/handlers/topics/get_topic_handler.rs | 3 +-
.../binary/handlers/topics/get_topics_handler.rs | 3 +-
.../binary/handlers/topics/purge_topic_handler.rs | 3 +-
.../binary/handlers/topics/update_topic_handler.rs | 3 +-
.../handlers/users/change_password_handler.rs | 3 +-
.../binary/handlers/users/create_user_handler.rs | 3 +-
.../binary/handlers/users/delete_user_handler.rs | 3 +-
.../src/binary/handlers/users/get_user_handler.rs | 2 +-
.../src/binary/handlers/users/get_users_handler.rs | 2 +-
.../binary/handlers/users/login_user_handler.rs | 2 +-
.../binary/handlers/users/logout_user_handler.rs | 3 +-
.../handlers/users/update_permissions_handler.rs | 3 +-
.../binary/handlers/users/update_user_handler.rs | 3 +-
core/server/src/http/messages.rs | 5 +-
core/server/src/http/segments.rs | 2 +-
core/server/src/lib.rs | 1 +
core/server/src/main.rs | 5 +-
core/server/src/metadata/reader.rs | 2 +-
core/server/src/quic/listener.rs | 3 +-
core/server/src/quic/quic_server.rs | 2 +-
core/{common => server}/src/sender/mod.rs | 4 +-
core/{common => server}/src/sender/quic_sender.rs | 5 +-
core/{common => server}/src/sender/tcp_sender.rs | 5 +-
.../src/sender/tcp_tls_sender.rs | 5 +-
.../src/sender/websocket_sender.rs | 4 +-
.../src/sender/websocket_tls_sender.rs | 4 +-
core/server/src/shard/builder.rs | 2 +-
core/server/src/shard/communication.rs | 2 +-
core/server/src/shard/handlers.rs | 4 +-
core/server/src/shard/mod.rs | 2 +-
core/server/src/shard/system/consumer_offsets.rs | 5 +-
core/server/src/shard/system/messages.rs | 4 +-
core/server/src/shard/system/partitions.rs | 4 +-
core/server/src/shard/system/segments.rs | 2 +-
core/server/src/shard/system/storage.rs | 2 +-
core/server/src/shard/system/streams.rs | 2 +-
core/server/src/shard/system/topics.rs | 2 +-
.../src/shard/tasks/periodic/message_cleaner.rs | 2 +-
core/server/src/shard/transmission/message.rs | 2 +-
core/server/src/streaming/deduplication/mod.rs | 2 +-
core/server/src/streaming/partitions/in_flight.rs | 2 +-
.../src/streaming/partitions/local_partitions.rs | 2 +-
core/server/src/streaming/partitions/ops.rs | 2 +-
core/server/src/streaming/partitions/ops_tests.rs | 17 +++---
.../src/streaming/segments/indexes/index_reader.rs | 2 +-
.../src/streaming/segments/indexes/index_writer.rs | 2 +-
core/server/src/streaming/segments/indexes/mod.rs | 2 +-
.../streaming/segments/messages/messages_reader.rs | 2 +-
.../streaming/segments/messages/messages_writer.rs | 2 +-
core/server/src/streaming/segments/storage.rs | 2 +-
core/server/src/streaming/segments/types/mod.rs | 5 +-
core/server/src/tcp/connection_handler.rs | 3 +-
core/server/src/tcp/tcp_listener.rs | 3 +-
core/server/src/tcp/tcp_tls_listener.rs | 5 +-
core/server/src/websocket/connection_handler.rs | 3 +-
core/server/src/websocket/websocket_listener.rs | 3 +-
.../server/src/websocket/websocket_tls_listener.rs | 5 +-
core/{partitions => server_common}/Cargo.toml | 37 +++++++------
.../src/alloc => server_common/src}/buffer.rs | 8 ++-
core/{common => server_common}/src/certificates.rs | 0
.../src/consensus_message.rs} | 12 ++---
.../src/deduplication/message_deduplicator.rs | 2 +-
.../src/deduplication/mod.rs | 0
.../message => server_common/src}/in_flight.rs | 2 +-
.../message => server_common/src}/indexes_mut.rs | 2 +-
.../src/consensus => server_common/src}/iobuf.rs | 0
core/server_common/src/lib.rs | 60 ++++++++++++++++++++++
.../src/alloc => server_common/src}/memory_pool.rs | 5 +-
.../src}/messages_batch_mut.rs | 14 ++---
.../src}/messages_batch_set.rs | 9 ++--
.../src}/segment_storage/index_reader.rs | 3 +-
.../src}/segment_storage/index_writer.rs | 7 ++-
.../src}/segment_storage/messages_reader.rs | 3 +-
.../src}/segment_storage/messages_writer.rs | 2 +-
.../src}/segment_storage/mod.rs | 2 +-
.../types => server_common/src}/send_messages2.rs | 11 ++--
.../src/sharding/local_idx.rs | 0
core/{common => server_common}/src/sharding/mod.rs | 0
.../src/sharding/namespace.rs | 0
.../src/sharding/partition_location.rs | 0
.../src/sharding/shard_id.rs | 0
core/shard/Cargo.toml | 1 +
core/shard/src/lib.rs | 10 ++--
core/shard/src/router.rs | 7 ++-
core/shard/src/shards_table.rs | 2 +-
core/simulator/Cargo.toml | 1 +
core/simulator/src/bus.rs | 8 +--
core/simulator/src/client.rs | 11 ++--
core/simulator/src/deps.rs | 4 +-
core/simulator/src/lib.rs | 13 ++---
core/simulator/src/main.rs | 7 +--
core/simulator/src/network.rs | 3 +-
core/simulator/src/packet.rs | 9 ++--
core/simulator/src/replica.rs | 2 +-
192 files changed, 510 insertions(+), 440 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index a19c63e07..5756dc4aa 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -2904,6 +2904,7 @@ dependencies = [
"serde",
"serde_json",
"serde_with",
+ "server_common",
"static-toml",
"strum 0.28.0",
"tracing",
@@ -2934,6 +2935,7 @@ dependencies = [
"message_bus",
"rand 0.10.1",
"rand_xoshiro",
+ "server_common",
"tracing",
]
@@ -6536,10 +6538,8 @@ dependencies = [
"aligned-vec",
"bytemuck",
"bytes",
- "compio-buf",
"enumset",
"secrecy",
- "smallvec",
"thiserror 2.0.18",
]
@@ -6548,38 +6548,26 @@ name = "iggy_common"
version = "0.10.1-edge.1"
dependencies = [
"aes-gcm",
- "aligned-vec",
"async-broadcast",
"async-trait",
"base64",
"blake3",
"bon",
"byte-unit",
- "bytemuck",
"bytes",
"chrono",
"clap",
"comfy-table",
- "compio",
- "crossbeam",
"derive_more",
- "err_trail",
- "human-repr",
"humantime",
"iggy_binary_protocol",
"lending-iterator",
- "moka",
- "nix",
- "once_cell",
"papaya",
- "rcgen",
"ring",
- "rustls",
"secrecy",
"serde",
"serde_json",
"serde_with",
- "serial_test",
"strum 0.28.0",
"thiserror 2.0.18",
"tokio",
@@ -7339,6 +7327,7 @@ dependencies = [
"bytemuck",
"compio",
"iggy_binary_protocol",
+ "server_common",
"tempfile",
]
@@ -7973,6 +7962,7 @@ dependencies = [
"rustls",
"rustls-pemfile",
"scopeguard",
+ "server_common",
"socket2 0.6.3",
"tempfile",
"thiserror 2.0.18",
@@ -7996,6 +7986,7 @@ dependencies = [
"paste",
"rmp-serde",
"serde",
+ "server_common",
"slab",
"tempfile",
"tracing",
@@ -9095,6 +9086,7 @@ dependencies = [
"journal",
"message_bus",
"ringbuffer",
+ "server_common",
"smallvec",
"tokio",
"tracing",
@@ -11355,6 +11347,7 @@ dependencies = [
"send_wrapper",
"serde",
"serde_json",
+ "server_common",
"slab",
"socket2 0.6.3",
"strum 0.28.0",
@@ -11430,6 +11423,7 @@ dependencies = [
"send_wrapper",
"serde",
"server",
+ "server_common",
"shard",
"slab",
"socket2 0.6.3",
@@ -11447,6 +11441,34 @@ dependencies = [
"vergen-git2",
]
+[[package]]
+name = "server_common"
+version = "0.1.0"
+dependencies = [
+ "aligned-vec",
+ "bytemuck",
+ "bytes",
+ "compio",
+ "compio-buf",
+ "crossbeam",
+ "err_trail",
+ "human-repr",
+ "iggy_binary_protocol",
+ "iggy_common",
+ "lending-iterator",
+ "moka",
+ "nix",
+ "once_cell",
+ "rcgen",
+ "rustls",
+ "serial_test",
+ "smallvec",
+ "thiserror 2.0.18",
+ "tokio",
+ "tracing",
+ "twox-hash",
+]
+
[[package]]
name = "sha1"
version = "0.10.6"
@@ -11519,6 +11541,7 @@ dependencies = [
"papaya",
"partitions",
"prometheus-client",
+ "server_common",
"thiserror 2.0.18",
"tracing",
]
@@ -11641,6 +11664,7 @@ dependencies = [
"partitions",
"rand 0.10.1",
"rand_xoshiro",
+ "server_common",
"shard",
"tracing",
]
diff --git a/Cargo.toml b/Cargo.toml
index 900712fbf..63ecce80a 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -56,6 +56,7 @@ members = [
"core/sdk",
"core/server",
"core/server-ng",
+ "core/server_common",
"core/shard",
"core/simulator",
"core/tools",
@@ -263,6 +264,7 @@ serde_yaml_ng = "0.10.0"
serial_test = "3.4.0"
server = { path = "core/server" }
server-ng = { path = "core/server-ng" }
+server_common = { path = "core/server_common" }
shard = { path = "core/shard" }
simd-json = { version = "0.17.0", features = ["serde_impl"] }
slab = "0.4.12"
diff --git a/core/binary_protocol/Cargo.toml b/core/binary_protocol/Cargo.toml
index 2e66bf312..a7e963534 100644
--- a/core/binary_protocol/Cargo.toml
+++ b/core/binary_protocol/Cargo.toml
@@ -29,15 +29,15 @@ repository = "https://github.com/apache/iggy"
readme = "README.md"
[dependencies]
-aligned-vec = { workspace = true }
bytemuck = { workspace = true }
bytes = { workspace = true }
-compio-buf = { workspace = true }
enumset = { workspace = true }
secrecy = { workspace = true }
-smallvec = { workspace = true }
thiserror = { workspace = true }
+[dev-dependencies]
+aligned-vec = { workspace = true }
+
[lints.clippy]
enum_glob_use = "deny"
nursery = "warn"
diff --git a/core/binary_protocol/src/consensus/mod.rs b/core/binary_protocol/src/consensus/mod.rs
index 9602b3492..6132c2da7 100644
--- a/core/binary_protocol/src/consensus/mod.rs
+++ b/core/binary_protocol/src/consensus/mod.rs
@@ -35,28 +35,10 @@
//! - [`StartViewChangeHeader`] - replica suspects primary failure (header-only)
//! - [`DoViewChangeHeader`] - replica -> primary candidate (header-only)
//! - [`StartViewHeader`] - new primary -> all replicas (header-only)
-//!
-//! ## Message wrapper
-//! - [`message::Message`] - typed consensus message wrapper over request or response backing
mod command;
mod error;
mod header;
-// iobuf was moved here verbatim from the former `core/iobuf` crate. Its
-// implementation is intentionally untouched for rc2; the existing lints
-// predate the move and are tracked as tech debt for a 0.9.x cleanup pass.
-#[allow(
- private_interfaces,
- clippy::missing_errors_doc,
- clippy::missing_panics_doc,
- clippy::must_use_candidate,
- clippy::return_self_not_must_use,
- clippy::missing_const_for_fn,
- clippy::non_send_fields_in_send_ty,
- clippy::use_self
-)]
-pub mod iobuf;
-pub mod message;
mod operation;
pub use command::Command2;
@@ -66,8 +48,4 @@ pub use header::{
GenericHeader, HEADER_SIZE, PrepareHeader, PrepareOkHeader, ReplyHeader, RequestHeader,
SIZE_FIELD_OFFSET, StartViewChangeHeader, StartViewHeader, read_size_field,
};
-pub use message::{
- ConsensusMessage, FragmentedBacking, MESSAGE_ALIGN, Message, MessageBacking, MessageBag,
- MutableBacking, RequestBacking, RequestBackingKind, ResponseBacking, ResponseBackingKind,
-};
pub use operation::Operation;
diff --git a/core/binary_protocol/src/lib.rs b/core/binary_protocol/src/lib.rs
index dfd9bdfdb..1552deb28 100644
--- a/core/binary_protocol/src/lib.rs
+++ b/core/binary_protocol/src/lib.rs
@@ -45,8 +45,9 @@
//! # VSR consensus framing
//!
//! All consensus headers are 256 bytes with `#[repr(C)]` layout.
-//! Deserialization is zero-copy via `bytemuck`. The [`Message`] type
-//! wraps request or response backing with typed header access.
+//! Deserialization is zero-copy via `bytemuck`. Runtime message wrappers
+//! (`Message`, `MessageBag`, backings) live in `server_common`; this crate
+//! exposes only the wire-level header types.
//!
//! - Client-facing: [`RequestHeader`], [`ReplyHeader`]
//! - Replication: [`PrepareHeader`], [`PrepareOkHeader`], [`CommitHeader`]
@@ -68,11 +69,9 @@ pub mod responses;
pub use codec::{WireDecode, WireEncode};
pub use consensus::{
- Command2, CommitHeader, ConsensusError, ConsensusHeader, ConsensusMessage, DoViewChangeHeader,
- EvictionHeader, EvictionReason, FragmentedBacking, GenericHeader, HEADER_SIZE, Message,
- MessageBacking, MessageBag, MutableBacking, Operation, PrepareHeader, PrepareOkHeader,
- ReplyHeader, RequestBacking, RequestBackingKind, RequestHeader, ResponseBacking,
- ResponseBackingKind, SIZE_FIELD_OFFSET, StartViewChangeHeader, StartViewHeader,
+ Command2, CommitHeader, ConsensusError, ConsensusHeader, DoViewChangeHeader, EvictionHeader,
+ EvictionReason, GenericHeader, HEADER_SIZE, Operation, PrepareHeader, PrepareOkHeader,
+ ReplyHeader, RequestHeader, SIZE_FIELD_OFFSET, StartViewChangeHeader, StartViewHeader,
read_size_field,
};
pub use dispatch::{COMMAND_TABLE, CommandMeta, lookup_by_operation, lookup_command};
diff --git a/core/common/Cargo.toml b/core/common/Cargo.toml
index b5d7c49d6..8d3a73d40 100644
--- a/core/common/Cargo.toml
+++ b/core/common/Cargo.toml
@@ -29,32 +29,22 @@ readme = "README.md"
[dependencies]
aes-gcm = { workspace = true }
-aligned-vec = { workspace = true }
async-broadcast = { workspace = true }
async-trait = { workspace = true }
base64 = { workspace = true }
blake3 = { workspace = true }
bon = { workspace = true }
byte-unit = { workspace = true }
-bytemuck = { workspace = true }
bytes = { workspace = true }
chrono = { workspace = true }
clap = { workspace = true }
comfy-table = { workspace = true }
-compio = { workspace = true }
-crossbeam = { workspace = true }
derive_more = { workspace = true }
-err_trail = { workspace = true }
-human-repr = { workspace = true }
humantime = { workspace = true }
iggy_binary_protocol = { workspace = true }
lending-iterator = { workspace = true }
-moka = { workspace = true }
-once_cell = { workspace = true }
papaya = { workspace = true }
-rcgen = { workspace = true }
ring = { workspace = true }
-rustls = { workspace = true }
secrecy = { workspace = true }
serde = { workspace = true }
serde_json = { workspace = true }
@@ -68,8 +58,3 @@ twox-hash = { workspace = true }
ulid = { workspace = true }
url = { workspace = true }
uuid = { workspace = true }
-[target.'cfg(unix)'.dependencies]
-nix = { workspace = true }
-
-[dev-dependencies]
-serial_test = { workspace = true }
diff --git a/core/common/src/alloc/mod.rs b/core/common/src/alloc/mod.rs
deleted file mode 100644
index 3d80f4dff..000000000
--- a/core/common/src/alloc/mod.rs
+++ /dev/null
@@ -1,20 +0,0 @@
-/* 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.
- */
-
-pub(crate) mod buffer;
-pub(crate) mod memory_pool;
diff --git a/core/common/src/lib.rs b/core/common/src/lib.rs
index 0ab795b63..7441ca3c3 100644
--- a/core/common/src/lib.rs
+++ b/core/common/src/lib.rs
@@ -17,14 +17,9 @@
* under the License.
*/
-mod alloc;
-mod certificates;
-mod deduplication;
mod error;
pub mod http;
mod macros;
-mod sender;
-pub mod sharding;
mod traits;
mod types;
mod utils;
@@ -34,11 +29,7 @@ pub use error::client_error::ClientError;
pub use error::iggy_error::{IggyError, IggyErrorDiscriminants};
// Locking is feature gated, thus only mod level re-export.
pub mod locking;
-pub use alloc::buffer::PooledBuffer;
-pub use alloc::memory_pool::{MEMORY_POOL, MemoryPool, MemoryPoolConfigOther, memory_pool};
-pub use certificates::generate_self_signed_certificate;
pub use chrono::{DateTime, Duration as ChronoDuration, Utc};
-pub use deduplication::MessageDeduplicator;
pub use http::consumer_groups::*;
pub use http::consumer_offsets::*;
pub use http::messages::*;
@@ -49,9 +40,6 @@ pub use http::streams::*;
pub use http::system::*;
pub use http::topics::*;
pub use http::users::*;
-pub use sender::{
- QuicSender, Sender, SenderKind, TcpSender, TcpTlsSender, WebSocketSender, WebSocketTlsSender,
-};
pub use traits::binary_client::BinaryClient;
pub use traits::binary_transport::BinaryTransport;
pub use traits::client::Client;
@@ -110,8 +98,6 @@ pub use types::permissions::permissions_global::*;
pub use types::permissions::personal_access_token::*;
pub use types::personal_access_tokens::*;
pub use types::segment::Segment;
-pub use types::segment_storage::*;
-pub use types::send_messages2;
pub use types::snapshot::*;
pub use types::stats::*;
pub use types::stream::*;
diff --git a/core/common/src/types/message/message_boundaries.rs b/core/common/src/types/message/message_boundaries.rs
index f1928e6c6..621a07d18 100644
--- a/core/common/src/types/message/message_boundaries.rs
+++ b/core/common/src/types/message/message_boundaries.rs
@@ -17,7 +17,7 @@
*/
#[derive(Clone, Copy)]
-pub(crate) struct IggyMessageBoundaries<'a> {
+pub struct IggyMessageBoundaries<'a> {
indexes: &'a [u8],
messages_len: usize,
base_position: u32,
@@ -25,7 +25,7 @@ pub(crate) struct IggyMessageBoundaries<'a> {
}
impl<'a> IggyMessageBoundaries<'a> {
- pub(crate) fn new(
+ pub fn new(
indexes: &'a [u8],
messages_len: usize,
base_position: u32,
@@ -45,11 +45,11 @@ impl<'a> IggyMessageBoundaries<'a> {
})
}
- pub(crate) fn count(&self) -> usize {
+ pub fn count(&self) -> usize {
self.count
}
- pub(crate) fn boundaries(&self, index: usize) -> Option<(usize, usize)> {
+ pub fn boundaries(&self, index: usize) -> Option<(usize, usize)> {
if index >= self.count {
return None;
}
diff --git a/core/common/src/types/message/message_view.rs b/core/common/src/types/message/message_view.rs
index 0706b047d..92f83ca87 100644
--- a/core/common/src/types/message/message_view.rs
+++ b/core/common/src/types/message/message_view.rs
@@ -154,7 +154,7 @@ impl<'a> IggyMessageViewIterator<'a> {
}
}
- pub(crate) fn new_with_boundaries(
+ pub fn new_with_boundaries(
messages: &'a [u8],
indexes: &'a [u8],
base_position: u32,
diff --git a/core/common/src/types/message/mod.rs b/core/common/src/types/message/mod.rs
index 281cc3273..a3b92f0c5 100644
--- a/core/common/src/types/message/mod.rs
+++ b/core/common/src/types/message/mod.rs
@@ -17,20 +17,16 @@
*/
mod iggy_message;
-mod in_flight;
mod index;
mod index_view;
mod indexes;
-mod indexes_mut;
-mod message_boundaries;
+pub mod message_boundaries;
mod message_header;
mod message_header_view;
mod message_header_view_mut;
mod message_view;
mod message_view_mut;
mod messages_batch;
-mod messages_batch_mut;
-mod messages_batch_set;
pub mod partitioning;
pub mod partitioning_kind;
mod poll_metadata;
@@ -44,11 +40,10 @@ pub const INDEX_SIZE: usize = 16;
pub use crate::http::messages::poll_messages::PollMessages;
pub use crate::http::messages::send_messages::SendMessages;
pub use iggy_message::{IggyMessage, MAX_PAYLOAD_SIZE, MAX_USER_HEADERS_SIZE};
-pub use in_flight::IggyMessagesBatchSetInFlight;
pub use index::IggyIndex;
pub use index_view::IggyIndexView;
pub use indexes::IggyIndexes;
-pub use indexes_mut::IggyIndexesMut;
+pub use message_boundaries::IggyMessageBoundaries;
pub use message_header::{
IGGY_MESSAGE_CHECKSUM_OFFSET_RANGE, IGGY_MESSAGE_HEADER_RANGE, IGGY_MESSAGE_HEADER_SIZE,
IGGY_MESSAGE_HEADERS_LENGTH_OFFSET_RANGE, IGGY_MESSAGE_ID_OFFSET_RANGE,
@@ -61,8 +56,6 @@ pub use message_header_view_mut::IggyMessageHeaderViewMut;
pub use message_view::{IggyMessageView, IggyMessageViewIterator};
pub use message_view_mut::{IggyMessageViewMut, IggyMessageViewMutIterator};
pub use messages_batch::IggyMessagesBatch;
-pub use messages_batch_mut::IggyMessagesBatchMut;
-pub use messages_batch_set::IggyMessagesBatchSet;
pub use partitioning::Partitioning;
pub use partitioning_kind::PartitioningKind;
pub use poll_metadata::IggyPollMetadata;
diff --git a/core/common/src/types/mod.rs b/core/common/src/types/mod.rs
index 3e7321a80..b0bace663 100644
--- a/core/common/src/types/mod.rs
+++ b/core/common/src/types/mod.rs
@@ -30,8 +30,6 @@ pub(crate) mod partition;
pub(crate) mod permissions;
pub(crate) mod personal_access_tokens;
pub(crate) mod segment;
-pub(crate) mod segment_storage;
-pub mod send_messages2;
pub(crate) mod snapshot;
pub(crate) mod stats;
pub(crate) mod stream;
diff --git a/core/configs/Cargo.toml b/core/configs/Cargo.toml
index ed7d6eba3..c0f606b8a 100644
--- a/core/configs/Cargo.toml
+++ b/core/configs/Cargo.toml
@@ -32,6 +32,7 @@ jsonwebtoken = { workspace = true }
serde = { workspace = true }
serde_json = { workspace = true }
serde_with = { workspace = true }
+server_common = { workspace = true }
static-toml = { workspace = true }
strum = { workspace = true }
tracing = { workspace = true }
diff --git a/core/configs/src/server_config/server.rs b/core/configs/src/server_config/server.rs
index fbfb1d458..98d019860 100644
--- a/core/configs/src/server_config/server.rs
+++ b/core/configs/src/server_config/server.rs
@@ -30,10 +30,11 @@ use err_trail::ErrContext;
use figment::providers::{Format, Toml};
use figment::value::Dict;
use figment::{Metadata, Profile, Provider};
-use iggy_common::{IggyByteSize, IggyDuration, MemoryPoolConfigOther, Validatable};
+use iggy_common::{IggyByteSize, IggyDuration, Validatable};
use serde::{Deserialize, Serialize};
use serde_with::DisplayFromStr;
use serde_with::serde_as;
+use server_common::MemoryPoolConfigOther;
use std::env;
use std::str::FromStr;
use std::sync::Arc;
diff --git a/core/configs/src/server_ng_config/server_ng.rs b/core/configs/src/server_ng_config/server_ng.rs
index a99e8d0d9..2567deb6d 100644
--- a/core/configs/src/server_ng_config/server_ng.rs
+++ b/core/configs/src/server_ng_config/server_ng.rs
@@ -35,11 +35,9 @@ use err_trail::ErrContext;
use figment::providers::{Format, Toml};
use figment::value::Dict;
use figment::{Metadata, Profile, Provider};
-use iggy_common::{
- Validatable,
- sharding::{MAX_PARTITIONS, MAX_STREAMS, MAX_TOPICS},
-};
+use iggy_common::Validatable;
use serde::{Deserialize, Serialize};
+use server_common::sharding::{MAX_PARTITIONS, MAX_STREAMS, MAX_TOPICS};
use std::env;
use std::sync::Arc;
diff --git a/core/configs/src/server_ng_config/validators.rs b/core/configs/src/server_ng_config/validators.rs
index 9cb7215c1..65e7bf3b0 100644
--- a/core/configs/src/server_ng_config/validators.rs
+++ b/core/configs/src/server_ng_config/validators.rs
@@ -31,8 +31,8 @@ use super::COMPONENT_NG;
use super::server_ng::{ExtraConfig, NamespaceConfig, ServerNgConfig};
use crate::ConfigurationError;
use err_trail::ErrContext;
-use iggy_common::sharding::IggyNamespace;
use iggy_common::{IggyExpiry, MaxTopicSize, Validatable};
+use server_common::sharding::IggyNamespace;
impl Validatable<ConfigurationError> for ServerNgConfig {
fn validate(&self) -> Result<(), ConfigurationError> {
diff --git a/core/consensus/Cargo.toml b/core/consensus/Cargo.toml
index 0439063e0..2319d8f33 100644
--- a/core/consensus/Cargo.toml
+++ b/core/consensus/Cargo.toml
@@ -38,6 +38,7 @@ iggy_common = { workspace = true }
message_bus = { workspace = true }
rand = { workspace = true }
rand_xoshiro = { workspace = true }
+server_common = { workspace = true }
tracing = { workspace = true }
[dev-dependencies]
diff --git a/core/consensus/src/client_table.rs b/core/consensus/src/client_table.rs
index 2610c2be2..03f9cf6d9 100644
--- a/core/consensus/src/client_table.rs
+++ b/core/consensus/src/client_table.rs
@@ -15,9 +15,8 @@
// specific language governing permissions and limitations
// under the License.
-use iggy_binary_protocol::consensus::MESSAGE_ALIGN;
-use iggy_binary_protocol::consensus::iobuf::Frozen;
-use iggy_binary_protocol::{Message, ReplyHeader};
+use iggy_binary_protocol::ReplyHeader;
+use server_common::{MESSAGE_ALIGN, Message, iobuf::Frozen};
use std::collections::HashMap;
use std::mem::size_of;
use tracing::trace;
@@ -119,7 +118,7 @@ pub enum RequestStatus {
/// Backs Register session, request contiguity, metadata-retry dedup, and
/// `NoSession`/`SessionTooLow` eviction. Partition plane is at-least-once;
/// `SendMessages` retries can re-commit at a new offset and consumers
-/// dedup via message ID (`iggy_common::MessageDeduplicator`).
+/// dedup via message ID (`server_common::MessageDeduplicator`).
///
/// Do not add per-partition `ClientTable` or `(client_id, request)` dedup
/// on the partition side, that flips iggy's contract toward at-most-once.
diff --git a/core/consensus/src/impls.rs b/core/consensus/src/impls.rs
index e5c8f78bc..b00b73dd7 100644
--- a/core/consensus/src/impls.rs
+++ b/core/consensus/src/impls.rs
@@ -25,12 +25,13 @@ use crate::{
};
use bit_set::BitSet;
use iggy_binary_protocol::{
- Command2, ConsensusHeader, DoViewChangeHeader, GenericHeader, Message, PrepareHeader,
- PrepareOkHeader, ReplyHeader, RequestHeader, StartViewChangeHeader, StartViewHeader,
+ Command2, ConsensusHeader, DoViewChangeHeader, GenericHeader, PrepareHeader, PrepareOkHeader,
+ ReplyHeader, RequestHeader, StartViewChangeHeader, StartViewHeader,
};
-use iggy_common::sharding::{IggyNamespace, METADATA_CONSENSUS_NAMESPACE};
use message_bus::IggyMessageBus;
use message_bus::MessageBus;
+use server_common::Message;
+use server_common::sharding::{IggyNamespace, METADATA_CONSENSUS_NAMESPACE};
use std::cell::{Cell, RefCell};
use std::collections::VecDeque;
@@ -2259,7 +2260,8 @@ mod pipeline_entry_tests {
//! subscriber `Canceled` even on happy path. Tests pin both halves.
use super::*;
- use iggy_binary_protocol::{Command2, Message, ReplyHeader};
+ use iggy_binary_protocol::{Command2, ReplyHeader};
+ use server_common::Message;
fn make_reply(client: u128, request: u64) -> Message<ReplyHeader> {
let header_size = std::mem::size_of::<ReplyHeader>();
diff --git a/core/consensus/src/lib.rs b/core/consensus/src/lib.rs
index babb907fa..4ae31c5df 100644
--- a/core/consensus/src/lib.rs
+++ b/core/consensus/src/lib.rs
@@ -15,8 +15,9 @@
// specific language governing permissions and limitations
// under the License.
-use iggy_binary_protocol::{ConsensusHeader, ConsensusMessage};
+use iggy_binary_protocol::ConsensusHeader;
use message_bus::MessageBus;
+use server_common::ConsensusMessage;
pub trait Project<T, C: Consensus> {
type Consensus: Consensus;
diff --git a/core/consensus/src/metadata_helpers.rs b/core/consensus/src/metadata_helpers.rs
index 5904f1049..f48d29208 100644
--- a/core/consensus/src/metadata_helpers.rs
+++ b/core/consensus/src/metadata_helpers.rs
@@ -21,8 +21,9 @@
use crate::client_table::{ClientTable, REGISTER_REQUEST_ID, RequestStatus};
use crate::{Consensus, Pipeline, PipelineEntry, VsrConsensus};
-use iggy_binary_protocol::{EvictionHeader, EvictionReason, HEADER_SIZE, Message};
+use iggy_binary_protocol::{EvictionHeader, EvictionReason, HEADER_SIZE};
use message_bus::MessageBus;
+use server_common::Message;
use std::cell::RefCell;
/// Request preflight (metadata only): session validation, dedup, in-flight check.
@@ -312,10 +313,9 @@ async fn send_eviction_to_client<B, P>(
mod tests {
use super::*;
use crate::{CLIENTS_TABLE_MAX, LocalPipeline};
- use iggy_binary_protocol::consensus::MESSAGE_ALIGN;
- use iggy_binary_protocol::consensus::iobuf::Frozen;
use iggy_binary_protocol::{Command2, Operation, ReplyHeader};
use message_bus::SendError;
+ use server_common::{MESSAGE_ALIGN, iobuf::Frozen};
/// Production-sized `ClientTable`.
fn fresh_client_table() -> RefCell<ClientTable> {
diff --git a/core/consensus/src/observability.rs b/core/consensus/src/observability.rs
index 2ac1d0044..37dd2eeaf 100644
--- a/core/consensus/src/observability.rs
+++ b/core/consensus/src/observability.rs
@@ -17,8 +17,8 @@
use crate::{Pipeline, PipelineEntry, Status, VsrAction, VsrConsensus};
use iggy_binary_protocol::Operation;
-use iggy_common::sharding::IggyNamespace;
use message_bus::MessageBus;
+use server_common::sharding::IggyNamespace;
use std::borrow::Cow;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
diff --git a/core/consensus/src/plane_helpers.rs b/core/consensus/src/plane_helpers.rs
index 7e0a89cf4..35aebd08b 100644
--- a/core/consensus/src/plane_helpers.rs
+++ b/core/consensus/src/plane_helpers.rs
@@ -19,11 +19,9 @@ use crate::{
Consensus, IgnoreReason, Pipeline, PipelineEntry, PlaneKind, PrepareOkOutcome, Sequencer,
Status, VsrConsensus,
};
-use iggy_binary_protocol::consensus::iobuf::Owned;
-use iggy_binary_protocol::{
- Command2, Message, PrepareHeader, PrepareOkHeader, ReplyHeader, RequestHeader,
-};
+use iggy_binary_protocol::{Command2, PrepareHeader, PrepareOkHeader, ReplyHeader, RequestHeader};
use message_bus::{MessageBus, SendError};
+use server_common::{Message, iobuf::Owned};
use std::ops::AsyncFnOnce;
/// Shared pipeline-first request flow (metadata + partitions).
@@ -449,9 +447,8 @@ pub async fn send_prepare_ok<B, P>(
mod tests {
use super::*;
use crate::{Consensus, LocalPipeline};
- use iggy_binary_protocol::consensus::MESSAGE_ALIGN;
- use iggy_binary_protocol::consensus::iobuf::Frozen;
use message_bus::SendError;
+ use server_common::{MESSAGE_ALIGN, iobuf::Frozen};
#[derive(Debug, Default)]
struct NoopBus;
diff --git a/core/journal/Cargo.toml b/core/journal/Cargo.toml
index 66c4853c1..220eca23f 100644
--- a/core/journal/Cargo.toml
+++ b/core/journal/Cargo.toml
@@ -32,6 +32,7 @@ publish = false
bytemuck = { workspace = true }
compio = { workspace = true }
iggy_binary_protocol = { workspace = true }
+server_common = { workspace = true }
[dev-dependencies]
tempfile = { workspace = true }
diff --git a/core/journal/src/prepare_journal.rs b/core/journal/src/prepare_journal.rs
index ea8d2c9bc..196e23ad4 100644
--- a/core/journal/src/prepare_journal.rs
+++ b/core/journal/src/prepare_journal.rs
@@ -18,9 +18,8 @@
use crate::file_storage::FileStorage;
use crate::{Journal, JournalHandle};
use compio::io::AsyncWriteAtExt;
-use iggy_binary_protocol::consensus::iobuf::Owned;
-use iggy_binary_protocol::consensus::message::{MESSAGE_ALIGN, Message};
use iggy_binary_protocol::consensus::{Command2, PrepareHeader};
+use server_common::{MESSAGE_ALIGN, Message, iobuf::Owned};
use std::cell::{Cell, OnceCell, Ref, RefCell};
use std::fmt;
use std::io;
diff --git a/core/message_bus/Cargo.toml b/core/message_bus/Cargo.toml
index d3096e9a3..17265aa7e 100644
--- a/core/message_bus/Cargo.toml
+++ b/core/message_bus/Cargo.toml
@@ -44,6 +44,7 @@ rand = { workspace = true }
rustls = { workspace = true }
rustls-pemfile = { workspace = true }
scopeguard = { workspace = true }
+server_common = { workspace = true }
socket2 = { workspace = true, features = ["all"] }
thiserror = { workspace = true }
tracing = { workspace = true }
diff --git a/core/message_bus/src/framing.rs b/core/message_bus/src/framing.rs
index 312118747..450154b07 100644
--- a/core/message_bus/src/framing.rs
+++ b/core/message_bus/src/framing.rs
@@ -24,10 +24,9 @@
use compio::BufResult;
use compio::buf::{IntoInner, IoBuf, IoBufMut};
use compio::io::{AsyncReadExt, AsyncWriteExt};
-use iggy_binary_protocol::consensus::MESSAGE_ALIGN;
-use iggy_binary_protocol::consensus::iobuf::Owned;
-use iggy_binary_protocol::{GenericHeader, HEADER_SIZE, Message, read_size_field};
+use iggy_binary_protocol::{GenericHeader, HEADER_SIZE, read_size_field};
use iggy_common::IggyError;
+use server_common::{MESSAGE_ALIGN, Message, iobuf::Owned};
use tracing::error;
/// Default hard ceiling on a single wire frame. Frames above this are
diff --git a/core/message_bus/src/installer/replica.rs b/core/message_bus/src/installer/replica.rs
index 9f33d3e81..9bff6f44b 100644
--- a/core/message_bus/src/installer/replica.rs
+++ b/core/message_bus/src/installer/replica.rs
@@ -27,7 +27,8 @@ use crate::transports::{ActorContext, TransportConn};
use async_channel::Receiver;
use compio::net::TcpStream;
use futures::FutureExt;
-use iggy_binary_protocol::{GenericHeader, Message};
+use iggy_binary_protocol::GenericHeader;
+use server_common::Message;
use std::cell::Cell;
use std::rc::Rc;
use tracing::{debug, info, warn};
diff --git a/core/message_bus/src/installer/tcp.rs b/core/message_bus/src/installer/tcp.rs
index 40f0f4dd1..7916f838d 100644
--- a/core/message_bus/src/installer/tcp.rs
+++ b/core/message_bus/src/installer/tcp.rs
@@ -30,7 +30,8 @@ use crate::transports::{ActorContext, TransportConn};
use async_channel::Receiver;
use compio::net::TcpStream;
use futures::FutureExt;
-use iggy_binary_protocol::{Command2, GenericHeader, Message};
+use iggy_binary_protocol::{Command2, GenericHeader};
+use server_common::Message;
use std::cell::Cell;
use std::rc::Rc;
use tracing::{debug, info, warn};
diff --git a/core/message_bus/src/lib.rs b/core/message_bus/src/lib.rs
index c93b133ef..c176edc60 100644
--- a/core/message_bus/src/lib.rs
+++ b/core/message_bus/src/lib.rs
@@ -101,9 +101,8 @@ pub use transports::tls::TlsServerCredentials;
use compio::runtime::JoinHandle;
use configs::server_ng::ServerNgConfig;
-use iggy_binary_protocol::consensus::MESSAGE_ALIGN;
-use iggy_binary_protocol::consensus::iobuf::Frozen;
-use iggy_binary_protocol::{GenericHeader, Message};
+use iggy_binary_protocol::GenericHeader;
+use server_common::{MESSAGE_ALIGN, Message, iobuf::Frozen};
use std::array;
use std::cell::{OnceCell, RefCell};
use std::net::SocketAddr;
diff --git a/core/message_bus/src/lifecycle/connection_registry.rs b/core/message_bus/src/lifecycle/connection_registry.rs
index 4b5d9b039..162d139df 100644
--- a/core/message_bus/src/lifecycle/connection_registry.rs
+++ b/core/message_bus/src/lifecycle/connection_registry.rs
@@ -33,8 +33,7 @@
//! [`ShutdownToken`]: crate::lifecycle::ShutdownToken
use compio::runtime::JoinHandle;
-use iggy_binary_protocol::consensus::MESSAGE_ALIGN;
-use iggy_binary_protocol::consensus::iobuf::Frozen;
+use server_common::{MESSAGE_ALIGN, iobuf::Frozen};
use std::cell::{Cell, RefCell};
use std::collections::HashMap;
use std::collections::hash_map::Entry as HmEntry;
@@ -730,7 +729,8 @@ impl ReplicaRegistry {
mod tests {
use super::*;
use crate::lifecycle::Shutdown;
- use iggy_binary_protocol::{Command2, GenericHeader, HEADER_SIZE, Message};
+ use iggy_binary_protocol::{Command2, GenericHeader, HEADER_SIZE};
+ use server_common::Message;
#[allow(clippy::cast_possible_truncation)]
fn make_bus_msg() -> BusMessage {
diff --git a/core/message_bus/src/transports/mod.rs b/core/message_bus/src/transports/mod.rs
index a141ba04b..121afaa8b 100644
--- a/core/message_bus/src/transports/mod.rs
+++ b/core/message_bus/src/transports/mod.rs
@@ -84,7 +84,8 @@ pub mod wss;
use crate::lifecycle::{BusReceiver, FusedShutdown};
use async_channel::Sender;
-use iggy_binary_protocol::{GenericHeader, Message};
+use iggy_binary_protocol::GenericHeader;
+use server_common::Message;
use std::io;
use std::net::SocketAddr;
diff --git a/core/message_bus/src/transports/quic.rs b/core/message_bus/src/transports/quic.rs
index b4b13ffda..466577479 100644
--- a/core/message_bus/src/transports/quic.rs
+++ b/core/message_bus/src/transports/quic.rs
@@ -63,7 +63,8 @@ use compio_quic::{
crypto::rustls::QuicServerConfig,
};
use futures::FutureExt;
-use iggy_binary_protocol::{GenericHeader, Message};
+use iggy_binary_protocol::GenericHeader;
+use server_common::Message;
use std::io;
use std::net::SocketAddr;
use std::sync::Arc;
@@ -516,10 +517,10 @@ mod tests {
use async_channel::bounded;
use compio::io::AsyncWrite;
use compio_quic::ClientBuilder;
- use iggy_binary_protocol::consensus::MESSAGE_ALIGN;
- use iggy_binary_protocol::consensus::iobuf::Frozen;
use iggy_binary_protocol::{Command2, HEADER_SIZE, SIZE_FIELD_OFFSET};
use rustls::pki_types::{CertificateDer, PrivateKeyDer, PrivatePkcs8KeyDer};
+ use server_common::MESSAGE_ALIGN;
+ use server_common::iobuf::Frozen;
use std::time::Duration;
fn install_crypto_provider() {
diff --git a/core/message_bus/src/transports/tcp.rs b/core/message_bus/src/transports/tcp.rs
index d6db44bd2..882447365 100644
--- a/core/message_bus/src/transports/tcp.rs
+++ b/core/message_bus/src/transports/tcp.rs
@@ -193,9 +193,7 @@ fn spawn_shutdown_watchdog(
#[allow(clippy::future_not_send)]
async fn reader_loop(
mut read_half: OwnedReadHalf<TcpStream>,
- in_tx: async_channel::Sender<
- iggy_binary_protocol::Message<iggy_binary_protocol::GenericHeader>,
- >,
+ in_tx: async_channel::Sender<server_common::Message<iggy_binary_protocol::GenericHeader>>,
max_message_size: usize,
label: &'static str,
peer: String,
@@ -292,9 +290,10 @@ mod tests {
use super::*;
use crate::lifecycle::Shutdown;
use async_channel::bounded;
- use iggy_binary_protocol::consensus::MESSAGE_ALIGN;
- use iggy_binary_protocol::consensus::iobuf::Frozen;
- use iggy_binary_protocol::{Command2, GenericHeader, HEADER_SIZE, Message, SIZE_FIELD_OFFSET};
+ use iggy_binary_protocol::{Command2, GenericHeader, HEADER_SIZE, SIZE_FIELD_OFFSET};
+ use server_common::MESSAGE_ALIGN;
+ use server_common::Message;
+ use server_common::iobuf::Frozen;
use std::time::Duration;
#[allow(clippy::cast_possible_truncation)]
diff --git a/core/message_bus/src/transports/tcp_tls.rs b/core/message_bus/src/transports/tcp_tls.rs
index 6adc34200..20c74e100 100644
--- a/core/message_bus/src/transports/tcp_tls.rs
+++ b/core/message_bus/src/transports/tcp_tls.rs
@@ -82,11 +82,10 @@ use compio::io::{AsyncRead, AsyncWrite, AsyncWriteExt};
use compio::net::TcpStream;
use compio::tls::{TlsAcceptor, TlsConnector, TlsStream};
use futures::FutureExt;
-use iggy_binary_protocol::consensus::MESSAGE_ALIGN;
-use iggy_binary_protocol::consensus::iobuf::Owned;
-use iggy_binary_protocol::{GenericHeader, HEADER_SIZE, Message, read_size_field};
+use iggy_binary_protocol::{GenericHeader, HEADER_SIZE, read_size_field};
use iggy_common::IggyError;
use rustls::pki_types::ServerName;
+use server_common::{MESSAGE_ALIGN, Message, iobuf::Owned};
use std::cell::UnsafeCell;
use std::marker::PhantomData;
use std::mem::MaybeUninit;
@@ -619,10 +618,9 @@ mod tests {
use crate::transports::tls::{install_default_crypto_provider, self_signed_for_loopback};
use async_channel::{Receiver, Sender, bounded};
use compio::net::TcpListener;
- use iggy_binary_protocol::consensus::MESSAGE_ALIGN;
- use iggy_binary_protocol::consensus::iobuf::Frozen;
- use iggy_binary_protocol::{Command2, GenericHeader, HEADER_SIZE, Message};
+ use iggy_binary_protocol::{Command2, GenericHeader, HEADER_SIZE};
use rustls::RootCertStore;
+ use server_common::{MESSAGE_ALIGN, Message, iobuf::Frozen};
use std::net::SocketAddr;
use std::sync::OnceLock;
diff --git a/core/message_bus/src/transports/tls/mod.rs b/core/message_bus/src/transports/tls/mod.rs
index 0b93ac6ea..626a1a130 100644
--- a/core/message_bus/src/transports/tls/mod.rs
+++ b/core/message_bus/src/transports/tls/mod.rs
@@ -135,7 +135,7 @@ pub fn load_pem(cert_path: &Path, key_path: &Path) -> io::Result<TlsServerCreden
/// in any caller this helper is intended for (tests, dev loops).
#[must_use]
pub fn self_signed_for_loopback() -> TlsServerCredentials {
- let (cert_chain, key_der) = iggy_common::generate_self_signed_certificate("localhost")
+ let (cert_chain, key_der) = server_common::generate_self_signed_certificate("localhost")
.expect("rcgen self-signed certificate generation must not fail in test/dev paths");
TlsServerCredentials {
cert_chain,
diff --git a/core/message_bus/src/transports/ws.rs b/core/message_bus/src/transports/ws.rs
index e27c07308..a2a328264 100644
--- a/core/message_bus/src/transports/ws.rs
+++ b/core/message_bus/src/transports/ws.rs
@@ -79,8 +79,8 @@ use compio::net::TcpStream;
use compio::ws::WebSocketStream;
use compio::ws::tungstenite::{self, Message as WsMessage};
use futures::FutureExt;
-use iggy_binary_protocol::consensus::MESSAGE_ALIGN;
-use iggy_binary_protocol::{GenericHeader, Message, read_size_field};
+use iggy_binary_protocol::{GenericHeader, read_size_field};
+use server_common::{MESSAGE_ALIGN, Message};
use std::time::Duration;
use tracing::{debug, warn};
@@ -118,8 +118,7 @@ pub(in crate::transports) fn decode_consensus_frame(
// straight into io_uring O_DIRECT buffers without a bounce copy. WS
// payloads have no alignment guarantee, so the only way to satisfy
// the invariant is one allocate-and-copy here.
- let owned =
- iggy_binary_protocol::consensus::iobuf::Owned::<MESSAGE_ALIGN>::copy_from_slice(body);
+ let owned = server_common::iobuf::Owned::<MESSAGE_ALIGN>::copy_from_slice(body);
Message::<GenericHeader>::try_from(owned).map_err(|_| FrameDecodeError::BadHeader)
}
@@ -314,8 +313,9 @@ mod tests {
use crate::lifecycle::Shutdown;
use async_channel::{Receiver, Sender, bounded};
use compio::net::TcpListener;
- use iggy_binary_protocol::consensus::iobuf::Frozen;
- use iggy_binary_protocol::{Command2, GenericHeader, HEADER_SIZE, Message};
+ use iggy_binary_protocol::{Command2, GenericHeader, HEADER_SIZE};
+ use server_common::Message;
+ use server_common::iobuf::Frozen;
use std::time::Duration;
#[allow(clippy::cast_possible_truncation)]
diff --git a/core/message_bus/src/transports/wss.rs b/core/message_bus/src/transports/wss.rs
index f2eed8251..853e689af 100644
--- a/core/message_bus/src/transports/wss.rs
+++ b/core/message_bus/src/transports/wss.rs
@@ -469,10 +469,11 @@ mod tests {
use crate::transports::tls::{install_default_crypto_provider, self_signed_for_loopback};
use async_channel::{Receiver, Sender, bounded};
use compio::net::TcpListener;
- use iggy_binary_protocol::consensus::MESSAGE_ALIGN;
- use iggy_binary_protocol::consensus::iobuf::Frozen;
- use iggy_binary_protocol::{Command2, GenericHeader, HEADER_SIZE, Message};
+ use iggy_binary_protocol::{Command2, GenericHeader, HEADER_SIZE};
use rustls::RootCertStore;
+ use server_common::MESSAGE_ALIGN;
+ use server_common::Message;
+ use server_common::iobuf::Frozen;
use std::net::SocketAddr;
use std::sync::OnceLock;
diff --git a/core/message_bus/tests/common/mod.rs b/core/message_bus/tests/common/mod.rs
index fa46d1f78..23d418bbb 100644
--- a/core/message_bus/tests/common/mod.rs
+++ b/core/message_bus/tests/common/mod.rs
@@ -26,7 +26,7 @@
#![allow(dead_code)] // each test binary uses a subset
-use iggy_binary_protocol::{Command2, GenericHeader, HEADER_SIZE, Message};
+use iggy_binary_protocol::{Command2, GenericHeader, HEADER_SIZE};
use message_bus::ConnectionInstaller;
use message_bus::client_listener::RequestHandler;
use message_bus::replica::listener::MessageHandler;
@@ -35,6 +35,7 @@ use message_bus::{
AcceptedTlsClientFn, AcceptedWsClientFn, AcceptedWssClientFn, ClientConnMeta,
ClientTransportKind, IggyMessageBus, fd_transfer, installer,
};
+use server_common::Message;
use std::cell::Cell;
use std::net::SocketAddr;
use std::rc::Rc;
diff --git a/core/message_bus/tests/quic_client_roundtrip.rs b/core/message_bus/tests/quic_client_roundtrip.rs
index 2a42ba040..61d46144f 100644
--- a/core/message_bus/tests/quic_client_roundtrip.rs
+++ b/core/message_bus/tests/quic_client_roundtrip.rs
@@ -25,9 +25,7 @@ use async_channel::bounded;
use common::{header_only, install_quic_clients_locally, loopback};
use compio_quic::{ClientBuilder, Endpoint};
use iggy_binary_protocol::Command2;
-use iggy_binary_protocol::consensus::MESSAGE_ALIGN;
-use iggy_binary_protocol::consensus::iobuf::Frozen;
-use iggy_binary_protocol::{GenericHeader, Message};
+use iggy_binary_protocol::GenericHeader;
use message_bus::QuicTuning;
use message_bus::client_listener::RequestHandler;
use message_bus::client_listener::quic::{bind, run};
@@ -36,6 +34,7 @@ use message_bus::transports::quic::{QuicTransportConn, server_config_with_cert};
use message_bus::transports::{ActorContext, TransportConn};
use message_bus::{FusedShutdown, IggyMessageBus, MessageBus, Shutdown};
use rustls::pki_types::{CertificateDer, PrivateKeyDer, PrivatePkcs8KeyDer};
+use server_common::{MESSAGE_ALIGN, Message, iobuf::Frozen};
use std::rc::Rc;
use std::time::Duration;
diff --git a/core/message_bus/tests/tcp_tls_client_listener.rs b/core/message_bus/tests/tcp_tls_client_listener.rs
index 1621e9923..fe01b4358 100644
--- a/core/message_bus/tests/tcp_tls_client_listener.rs
+++ b/core/message_bus/tests/tcp_tls_client_listener.rs
@@ -28,9 +28,7 @@ use async_channel::bounded;
use common::{header_only, install_tls_clients_locally, loopback};
use compio::net::TcpStream;
use iggy_binary_protocol::Command2;
-use iggy_binary_protocol::consensus::MESSAGE_ALIGN;
-use iggy_binary_protocol::consensus::iobuf::Frozen;
-use iggy_binary_protocol::{GenericHeader, Message};
+use iggy_binary_protocol::GenericHeader;
use message_bus::client_listener::RequestHandler;
use message_bus::client_listener::tcp_tls::{bind, run};
use message_bus::transports::tcp_tls::TcpTlsTransportConn;
@@ -39,6 +37,7 @@ use message_bus::transports::{ActorContext, TransportConn};
use message_bus::{FusedShutdown, IggyMessageBus, MessageBus, MessageBusConfig, Shutdown, framing};
use rustls::RootCertStore;
use rustls::pki_types::ServerName;
+use server_common::{MESSAGE_ALIGN, Message, iobuf::Frozen};
use std::rc::Rc;
use std::sync::Arc;
use std::time::{Duration, Instant};
diff --git a/core/message_bus/tests/tcp_tls_client_roundtrip.rs b/core/message_bus/tests/tcp_tls_client_roundtrip.rs
index 39333fc73..9a922dc1e 100644
--- a/core/message_bus/tests/tcp_tls_client_roundtrip.rs
+++ b/core/message_bus/tests/tcp_tls_client_roundtrip.rs
@@ -32,9 +32,7 @@ use common::{
};
use compio::net::TcpStream;
use iggy_binary_protocol::Command2;
-use iggy_binary_protocol::consensus::MESSAGE_ALIGN;
-use iggy_binary_protocol::consensus::iobuf::Frozen;
-use iggy_binary_protocol::{GenericHeader, Message};
+use iggy_binary_protocol::GenericHeader;
use message_bus::client_listener::RequestHandler;
use message_bus::connector::DEFAULT_RECONNECT_PERIOD;
use message_bus::replica::io::start_on_shard_zero;
@@ -45,6 +43,7 @@ use message_bus::transports::{ActorContext, TransportConn};
use message_bus::{FusedShutdown, IggyMessageBus, MessageBus, Shutdown, framing};
use rustls::RootCertStore;
use rustls::pki_types::ServerName;
+use server_common::{MESSAGE_ALIGN, Message, iobuf::Frozen};
use std::rc::Rc;
use std::sync::Arc;
use std::time::Duration;
diff --git a/core/message_bus/tests/ws_client_roundtrip.rs b/core/message_bus/tests/ws_client_roundtrip.rs
index e936e3394..713472ff5 100644
--- a/core/message_bus/tests/ws_client_roundtrip.rs
+++ b/core/message_bus/tests/ws_client_roundtrip.rs
@@ -34,14 +34,13 @@ use async_channel::bounded;
use common::{header_only, install_ws_clients_locally, loopback};
use compio::net::TcpStream;
use iggy_binary_protocol::Command2;
-use iggy_binary_protocol::consensus::MESSAGE_ALIGN;
-use iggy_binary_protocol::consensus::iobuf::Frozen;
-use iggy_binary_protocol::{GenericHeader, Message};
+use iggy_binary_protocol::GenericHeader;
use message_bus::client_listener::RequestHandler;
use message_bus::client_listener::ws::{bind, run};
use message_bus::transports::ws::WsTransportConn;
use message_bus::transports::{ActorContext, TransportConn};
use message_bus::{FusedShutdown, IggyMessageBus, MessageBus, Shutdown, framing};
+use server_common::{MESSAGE_ALIGN, Message, iobuf::Frozen};
use std::rc::Rc;
use std::time::Duration;
diff --git a/core/message_bus/tests/wss_client_listener.rs b/core/message_bus/tests/wss_client_listener.rs
index cfed3e03a..03ce08787 100644
--- a/core/message_bus/tests/wss_client_listener.rs
+++ b/core/message_bus/tests/wss_client_listener.rs
@@ -32,9 +32,7 @@ use async_channel::bounded;
use common::{header_only, install_wss_clients_locally, loopback};
use compio::net::TcpStream;
use iggy_binary_protocol::Command2;
-use iggy_binary_protocol::consensus::MESSAGE_ALIGN;
-use iggy_binary_protocol::consensus::iobuf::Frozen;
-use iggy_binary_protocol::{GenericHeader, Message};
+use iggy_binary_protocol::GenericHeader;
use message_bus::client_listener::RequestHandler;
use message_bus::client_listener::wss::{bind, run};
use message_bus::transports::tls::{install_default_crypto_provider, self_signed_for_loopback};
@@ -43,6 +41,7 @@ use message_bus::transports::{ActorContext, TransportConn};
use message_bus::{FusedShutdown, IggyMessageBus, MessageBus, MessageBusConfig, Shutdown, framing};
use rustls::RootCertStore;
use rustls::pki_types::ServerName;
+use server_common::{MESSAGE_ALIGN, Message, iobuf::Frozen};
use std::rc::Rc;
use std::sync::Arc;
use std::time::{Duration, Instant};
diff --git a/core/message_bus/tests/wss_client_roundtrip.rs b/core/message_bus/tests/wss_client_roundtrip.rs
index 427acd9b5..44a188dfa 100644
--- a/core/message_bus/tests/wss_client_roundtrip.rs
+++ b/core/message_bus/tests/wss_client_roundtrip.rs
@@ -32,9 +32,7 @@ use common::{
};
use compio::net::TcpStream;
use iggy_binary_protocol::Command2;
-use iggy_binary_protocol::consensus::MESSAGE_ALIGN;
-use iggy_binary_protocol::consensus::iobuf::Frozen;
-use iggy_binary_protocol::{GenericHeader, Message};
+use iggy_binary_protocol::GenericHeader;
use message_bus::client_listener::RequestHandler;
use message_bus::connector::DEFAULT_RECONNECT_PERIOD;
use message_bus::replica::io::start_on_shard_zero;
@@ -45,6 +43,7 @@ use message_bus::transports::{ActorContext, TransportConn};
use message_bus::{FusedShutdown, IggyMessageBus, MessageBus, Shutdown, framing};
use rustls::RootCertStore;
use rustls::pki_types::ServerName;
+use server_common::{MESSAGE_ALIGN, Message, iobuf::Frozen};
use std::rc::Rc;
use std::sync::Arc;
use std::time::Duration;
diff --git a/core/metadata/Cargo.toml b/core/metadata/Cargo.toml
index 213303a81..ac68c96d4 100644
--- a/core/metadata/Cargo.toml
+++ b/core/metadata/Cargo.toml
@@ -41,6 +41,7 @@ message_bus = { workspace = true }
paste = { workspace = true }
rmp-serde = { workspace = true }
serde = { workspace = true, features = ["derive"] }
+server_common = { workspace = true }
slab = { workspace = true }
tracing = { workspace = true }
diff --git a/core/metadata/src/impls/metadata.rs b/core/metadata/src/impls/metadata.rs
index d4318889c..ccffc2937 100644
--- a/core/metadata/src/impls/metadata.rs
+++ b/core/metadata/src/impls/metadata.rs
@@ -35,13 +35,14 @@ use iggy_binary_protocol::requests::partitions::CreatePartitionsWithAssignmentsR
use iggy_binary_protocol::requests::topics::CreateTopicRequest as WireCreateTopicRequest;
use iggy_binary_protocol::requests::topics::CreateTopicWithAssignmentsRequest as PersistedCreateTopicRequest;
use iggy_binary_protocol::{
- Command2, ConsensusHeader, GenericHeader, Message, Operation, PrepareHeader, PrepareOkHeader,
+ Command2, ConsensusHeader, GenericHeader, Operation, PrepareHeader, PrepareOkHeader,
RequestHeader, WireDecode, WireEncode,
};
use iggy_common::IggyError;
use iggy_common::variadic;
use journal::{Journal, JournalHandle};
use message_bus::MessageBus;
+use server_common::Message;
use std::cell::RefCell;
use std::mem::size_of;
use std::path::Path;
@@ -49,9 +50,7 @@ use tracing::{debug, error, warn};
fn freeze_client_reply(
message: Message<GenericHeader>,
-) -> iggy_binary_protocol::consensus::iobuf::Frozen<
- { iggy_binary_protocol::consensus::MESSAGE_ALIGN },
-> {
+) -> server_common::iobuf::Frozen<{ server_common::MESSAGE_ALIGN }> {
message.into_frozen()
}
diff --git a/core/metadata/src/impls/recovery.rs b/core/metadata/src/impls/recovery.rs
index 06987f062..a32e558f9 100644
--- a/core/metadata/src/impls/recovery.rs
+++ b/core/metadata/src/impls/recovery.rs
@@ -19,9 +19,9 @@ use crate::impls::metadata::IggySnapshot;
use crate::stm::StateMachine;
use crate::stm::snapshot::{MetadataSnapshot, RestoreSnapshot, Snapshot, SnapshotError};
use iggy_binary_protocol::consensus::PrepareHeader;
-use iggy_binary_protocol::consensus::message::Message;
use iggy_common::IggyError;
use journal::prepare_journal::{JournalError, PrepareJournal};
+use server_common::Message;
use std::fmt;
use std::path::Path;
@@ -164,9 +164,9 @@ where
#[allow(clippy::cast_possible_truncation)]
mod tests {
use super::*;
- use iggy_binary_protocol::consensus::iobuf::Owned;
use iggy_binary_protocol::consensus::{Command2, Operation};
use journal::Journal;
+ use server_common::iobuf::Owned;
use tempfile::tempdir;
use crate::MuxStateMachine;
diff --git a/core/metadata/src/stm/mod.rs b/core/metadata/src/stm/mod.rs
index 90f19c903..89c9b3a4e 100644
--- a/core/metadata/src/stm/mod.rs
+++ b/core/metadata/src/stm/mod.rs
@@ -387,7 +387,7 @@ macro_rules! collect_handlers {
impl $crate::stm::Command for [<$state Inner>] {
type Cmd = [<$state Command>];
- type Input = ::iggy_binary_protocol::Message<::iggy_binary_protocol::PrepareHeader>;
+ type Input = ::server_common::Message<::iggy_binary_protocol::PrepareHeader>;
type Error = ::iggy_common::IggyError;
fn parse(input: Self::Input) -> Result<::iggy_common::Either<Self::Cmd, Self::Input>, Self::Error> {
diff --git a/core/metadata/src/stm/mux.rs b/core/metadata/src/stm/mux.rs
index c9a0d1e6a..aa72f13d0 100644
--- a/core/metadata/src/stm/mux.rs
+++ b/core/metadata/src/stm/mux.rs
@@ -16,9 +16,10 @@
// under the License.
use crate::stm::snapshot::{FillSnapshot, RestoreSnapshot, SnapshotError};
-use iggy_binary_protocol::{Message, PrepareHeader};
+use iggy_binary_protocol::PrepareHeader;
use iggy_common::Either;
use iggy_common::variadic;
+use server_common::Message;
use crate::stm::{State, StateMachine};
@@ -220,7 +221,8 @@ mod tests {
use crate::stm::mux::MuxStateMachine;
use crate::stm::stream::{Streams, StreamsInner};
use crate::stm::user::{Users, UsersInner};
- use iggy_binary_protocol::{Message, PrepareHeader};
+ use iggy_binary_protocol::PrepareHeader;
+ use server_common::Message;
let users: Users = UsersInner::new().into();
let streams: Streams = StreamsInner::new().into();
diff --git a/core/partitions/Cargo.toml b/core/partitions/Cargo.toml
index 158790862..4562e6c2b 100644
--- a/core/partitions/Cargo.toml
+++ b/core/partitions/Cargo.toml
@@ -38,6 +38,7 @@ iggy_common = { workspace = true }
journal = { workspace = true }
message_bus = { workspace = true }
ringbuffer = { workspace = true }
+server_common = { workspace = true }
smallvec = { workspace = true }
tokio = { workspace = true }
tracing = { workspace = true }
diff --git a/core/partitions/src/iggy_partition.rs b/core/partitions/src/iggy_partition.rs
index f57bd790b..055b0ddf0 100644
--- a/core/partitions/src/iggy_partition.rs
+++ b/core/partitions/src/iggy_partition.rs
@@ -36,18 +36,22 @@ use consensus::{
fence_old_prepare_by_commit, replicate_preflight, replicate_to_next_in_chain,
send_prepare_ok as send_prepare_ok_common,
};
-use iggy_binary_protocol::consensus::iobuf::Frozen;
-use iggy_binary_protocol::{AckLevel, Message, Operation, PrepareHeader};
+use iggy_binary_protocol::{AckLevel, Operation, PrepareHeader};
use iggy_binary_protocol::{PrepareOkHeader, RequestHeader};
use iggy_common::{
ConsumerGroupId, ConsumerGroupOffsets, ConsumerKind, ConsumerOffset, ConsumerOffsets,
- IggyByteSize, IggyError, IggyTimestamp, PartitionStats, PollingKind, SegmentStorage,
- send_messages2::stamp_prepare_for_persistence,
- send_messages2::{convert_request_message, decode_prepare_slice},
- sharding::IggyNamespace,
+ IggyByteSize, IggyError, IggyTimestamp, PartitionStats, PollingKind,
};
use journal::Journal as _;
use message_bus::{IggyMessageBus, MessageBus};
+use server_common::{
+ Message, SegmentStorage,
+ iobuf::Frozen,
+ send_messages2::{
+ convert_request_message, decode_prepare_slice, stamp_prepare_for_persistence,
+ },
+ sharding::IggyNamespace,
+};
use std::collections::HashMap;
use std::rc::Rc;
use std::sync::Arc;
@@ -59,7 +63,7 @@ use tracing::{debug, warn};
//
// Note: there is no per-client write dedup at the partition plane.
// `SendMessages` retries are at-least-once and may commit multiple times.
-// Consumers handle duplicate messages via `iggy_common::MessageDeduplicator`
+// Consumers handle duplicate messages via `server_common::MessageDeduplicator`
// (message-id based) if they care.
#[derive(Debug)]
pub struct IggyPartition<B = IggyMessageBus>
diff --git a/core/partitions/src/iggy_partitions.rs b/core/partitions/src/iggy_partitions.rs
index e61e9f72a..6846abf03 100644
--- a/core/partitions/src/iggy_partitions.rs
+++ b/core/partitions/src/iggy_partitions.rs
@@ -23,8 +23,8 @@ use consensus::{Consensus, Plane, PlaneIdentity, VsrConsensus};
use iggy_binary_protocol::{
Command2, ConsensusHeader, PrepareHeader, PrepareOkHeader, RequestHeader,
};
-use iggy_common::sharding::{IggyNamespace, LocalIdx, ShardId};
use message_bus::MessageBus;
+use server_common::sharding::{IggyNamespace, LocalIdx, ShardId};
use std::cell::UnsafeCell;
use std::collections::HashMap;
use tracing::warn;
diff --git a/core/partitions/src/journal.rs b/core/partitions/src/journal.rs
index 262d86ea0..ab8867eda 100644
--- a/core/partitions/src/journal.rs
+++ b/core/partitions/src/journal.rs
@@ -15,10 +15,12 @@
// specific language governing permissions and limitations
// under the License.
-use iggy_binary_protocol::consensus::iobuf::{Frozen, Owned};
use iggy_binary_protocol::{Operation, PrepareHeader};
-use iggy_common::send_messages2::{COMMAND_HEADER_SIZE, SendMessages2Ref, decode_prepare_slice};
use journal::{Journal, Storage};
+use server_common::{
+ iobuf::{Frozen, Owned},
+ send_messages2::{COMMAND_HEADER_SIZE, SendMessages2Ref, decode_prepare_slice},
+};
use std::io;
use std::{
cell::UnsafeCell,
@@ -569,8 +571,9 @@ fn push_selected_batch_fragments(
#[cfg(test)]
mod tests {
use super::*;
- use iggy_binary_protocol::{Command2, HEADER_SIZE, Message};
+ use iggy_binary_protocol::{Command2, HEADER_SIZE};
use journal::Journal;
+ use server_common::Message;
fn build_prepare(op: u64, size: usize) -> Message<PrepareHeader> {
Message::<PrepareHeader>::new(size).transmute_header(|_, h: &mut PrepareHeader| {
diff --git a/core/partitions/src/lib.rs b/core/partitions/src/lib.rs
index 3dd4c3f2e..fa0ad1558 100644
--- a/core/partitions/src/lib.rs
+++ b/core/partitions/src/lib.rs
@@ -28,14 +28,15 @@ mod offset_storage;
mod segment;
mod types;
-use iggy_binary_protocol::{Message, PrepareHeader};
+use iggy_binary_protocol::PrepareHeader;
use iggy_common::IggyError;
-pub use iggy_common::send_messages2::{IggyMessage2, IggyMessage2Header, IggyMessages2};
pub use iggy_index_writer::IggyIndexWriter;
pub use iggy_partition::IggyPartition;
pub use iggy_partitions::IggyPartitions;
pub use messages_writer::MessagesWriter;
pub use segment::Segment;
+use server_common::Message;
+pub use server_common::send_messages2::{IggyMessage2, IggyMessage2Header, IggyMessages2};
pub use types::{
AppendResult, Fragment, PartitionOffsets, PartitionsConfig, PollFragments, PollQueryResult,
PollingArgs, PollingConsumer, SendMessagesResult,
diff --git a/core/partitions/src/log.rs b/core/partitions/src/log.rs
index 79ffd6790..8908601c4 100644
--- a/core/partitions/src/log.rs
+++ b/core/partitions/src/log.rs
@@ -19,9 +19,10 @@ use crate::iggy_index::{IGGY_INDEX_SIZE, IggyIndexCache};
use crate::iggy_index_writer::IggyIndexWriter;
use crate::messages_writer::MessagesWriter;
use crate::segment::Segment;
-use iggy_common::{IggyByteSize, IggyMessagesBatch, IggyMessagesBatchSetInFlight, SegmentStorage};
+use iggy_common::{IggyByteSize, IggyMessagesBatch};
use journal::{Journal, Storage};
use ringbuffer::AllocRingBuffer;
+use server_common::{IggyMessagesBatchSetInFlight, SegmentStorage};
use std::fmt::Debug;
use std::rc::Rc;
diff --git a/core/partitions/src/messages_writer.rs b/core/partitions/src/messages_writer.rs
index d45f31146..6f330aadb 100644
--- a/core/partitions/src/messages_writer.rs
+++ b/core/partitions/src/messages_writer.rs
@@ -19,8 +19,8 @@ use compio::{
fs::{File, OpenOptions},
io::AsyncWriteAtExt,
};
-use iggy_binary_protocol::consensus::iobuf::Frozen;
use iggy_common::{IggyByteSize, IggyError};
+use server_common::iobuf::Frozen;
use std::{
rc::Rc,
sync::atomic::{AtomicU64, Ordering},
diff --git a/core/partitions/src/types.rs b/core/partitions/src/types.rs
index 1f81d9900..9d599488d 100644
--- a/core/partitions/src/types.rs
+++ b/core/partitions/src/types.rs
@@ -15,8 +15,8 @@
// specific language governing permissions and limitations
// under the License.
-use iggy_binary_protocol::consensus::iobuf::Frozen;
use iggy_common::{IggyByteSize, PollingStrategy};
+use server_common::iobuf::Frozen;
use smallvec::SmallVec;
#[derive(Debug, Clone)]
diff --git a/core/server-ng/Cargo.toml b/core/server-ng/Cargo.toml
index 3b130fdb4..972625d7b 100644
--- a/core/server-ng/Cargo.toml
+++ b/core/server-ng/Cargo.toml
@@ -145,6 +145,7 @@ secrecy = { workspace = true }
send_wrapper = { workspace = true }
serde = { workspace = true }
server = { workspace = true }
+server_common = { workspace = true }
shard = { workspace = true }
slab = { workspace = true }
socket2 = { workspace = true }
diff --git a/core/server-ng/src/bootstrap.rs b/core/server-ng/src/bootstrap.rs
index 79ecf1748..6380bdae0 100644
--- a/core/server-ng/src/bootstrap.rs
+++ b/core/server-ng/src/bootstrap.rs
@@ -29,10 +29,9 @@ use consensus::{LocalPipeline, PartitionsHandle, Sequencer, VsrConsensus};
// non-blocking variants for cancel-safe shutdown polling.
use crossfire::{AsyncRxTrait, AsyncTxTrait};
use iggy_binary_protocol::RequestHeader;
-use iggy_common::sharding::{IggyNamespace, PartitionLocation, ShardId};
use iggy_common::{
ConsumerGroupOffsets, ConsumerOffsets, IggyByteSize, IggyError, PartitionStats, TopicStats,
- sharding::LocalIdx, variadic,
+ variadic,
};
use journal::Journal;
use journal::prepare_journal::PrepareJournal;
@@ -61,6 +60,7 @@ use metadata::stm::user::Users;
use partitions::{
IggyIndexWriter, IggyPartition, IggyPartitions, MessagesWriter, PartitionsConfig, Segment,
};
+use server_common::sharding::{IggyNamespace, LocalIdx, PartitionLocation, ShardId};
// TODO: decouple bootstrap/storage helpers and logging from the `server` crate.
use server::bootstrap::{create_directories, create_shard_executor};
use server::log::logger::Logging;
@@ -1103,7 +1103,7 @@ fn restore_metadata_consensus(
CLUSTER_ID,
self_replica_id,
replica_count,
- iggy_common::sharding::METADATA_CONSENSUS_NAMESPACE,
+ server_common::sharding::METADATA_CONSENSUS_NAMESPACE,
bus,
LocalPipeline::new(),
);
@@ -1373,7 +1373,7 @@ fn indexes_max_timestamp(indexes: &server::streaming::segments::IggyIndexesMut)
}
async fn load_segment_max_timestamp(
- storage: &iggy_common::SegmentStorage,
+ storage: &server_common::SegmentStorage,
stream_id: usize,
topic_id: usize,
partition_id: usize,
@@ -2262,7 +2262,7 @@ fn load_quic_server_credentials(
) -> Result<replica_io::QuicServerCredentials, ServerNgError> {
let certificate = &config.quic.certificate;
if certificate.self_signed {
- let (cert_chain, key_der) = iggy_common::generate_self_signed_certificate("localhost")
+ let (cert_chain, key_der) = server_common::generate_self_signed_certificate("localhost")
.map_err(|error| ServerNgError::ListenerCredentials {
transport: "quic",
source: std::io::Error::other(error.to_string()),
diff --git a/core/server-ng/src/login_register.rs b/core/server-ng/src/login_register.rs
index f152421ff..cea603559 100644
--- a/core/server-ng/src/login_register.rs
+++ b/core/server-ng/src/login_register.rs
@@ -28,7 +28,7 @@
use crate::session_manager::{SessionError, SessionManager};
use consensus::VsrConsensus;
use iggy_binary_protocol::{
- EvictionReason, Message, PrepareHeader,
+ EvictionReason, PrepareHeader,
requests::users::{LoginRegisterRequest, LoginRegisterWithPatRequest},
responses::users::LoginRegisterResponse,
};
@@ -38,6 +38,7 @@ use metadata::{
IggyMetadata, RegisterSubmitError, impls::metadata::StreamsFrontend, stm::StateMachine,
};
use secrecy::ExposeSecret;
+use server_common::Message;
/// Credential verifier. Real impl: metadata user store + Argon2.
pub trait CredentialVerifier {
diff --git a/core/server-ng/src/main.rs b/core/server-ng/src/main.rs
index 0561b3001..6ac5ca2b1 100644
--- a/core/server-ng/src/main.rs
+++ b/core/server-ng/src/main.rs
@@ -69,7 +69,7 @@ fn main() -> Result<(), ServerNgError> {
logging.early_init();
let config = load_config(&mut logging).await?;
- iggy_common::MemoryPool::init_pool(&config.system.memory_pool.into_other());
+ server_common::MemoryPool::init_pool(&config.system.memory_pool.into_other());
Ok((config, args.replica_id))
});
diff --git a/core/server/Cargo.toml b/core/server/Cargo.toml
index 21d00ea9f..9e1b2dd56 100644
--- a/core/server/Cargo.toml
+++ b/core/server/Cargo.toml
@@ -92,6 +92,7 @@ secrecy = { workspace = true }
send_wrapper = { workspace = true }
serde = { workspace = true }
serde_json = { workspace = true }
+server_common = { workspace = true }
slab = { workspace = true }
socket2 = { workspace = true }
strum = { workspace = true }
diff --git a/core/server/src/binary/dispatch.rs b/core/server/src/binary/dispatch.rs
index 9138483e1..a2a4b0b82 100644
--- a/core/server/src/binary/dispatch.rs
+++ b/core/server/src/binary/dispatch.rs
@@ -17,6 +17,7 @@
*/
use crate::binary::handlers;
+use crate::sender::SenderKind;
use crate::shard::IggyShard;
use crate::streaming::session::Session;
use bytes::BytesMut;
@@ -33,9 +34,7 @@ use iggy_binary_protocol::requests::streams::*;
use iggy_binary_protocol::requests::system::*;
use iggy_binary_protocol::requests::topics::*;
use iggy_binary_protocol::requests::users::*;
-use iggy_common::{
- Consumer, ConsumerKind, Identifier, IggyError, PollingKind, PollingStrategy, SenderKind,
-};
+use iggy_common::{Consumer, ConsumerKind, Identifier, IggyError, PollingKind, PollingStrategy};
use std::rc::Rc;
use tracing::{error, warn};
diff --git a/core/server/src/binary/handlers/cluster/get_cluster_metadata_handler.rs b/core/server/src/binary/handlers/cluster/get_cluster_metadata_handler.rs
index 466945f39..3a22efbaf 100644
--- a/core/server/src/binary/handlers/cluster/get_cluster_metadata_handler.rs
+++ b/core/server/src/binary/handlers/cluster/get_cluster_metadata_handler.rs
@@ -17,13 +17,14 @@
*/
use crate::binary::dispatch::HandlerResult;
+use crate::sender::SenderKind;
use crate::shard::IggyShard;
use crate::streaming::session::Session;
use iggy_binary_protocol::codec::WireEncode;
use iggy_binary_protocol::responses::system::get_cluster_metadata::{
ClusterMetadataResponse, ClusterNodeResponse,
};
-use iggy_common::{IggyError, SenderKind};
+use iggy_common::IggyError;
use std::rc::Rc;
use tracing::{debug, instrument};
diff --git a/core/server/src/binary/handlers/consumer_groups/create_consumer_group_handler.rs b/core/server/src/binary/handlers/consumer_groups/create_consumer_group_handler.rs
index 82f3aa049..87b2bc58a 100644
--- a/core/server/src/binary/handlers/consumer_groups/create_consumer_group_handler.rs
+++ b/core/server/src/binary/handlers/consumer_groups/create_consumer_group_handler.rs
@@ -17,6 +17,7 @@
*/
use crate::binary::dispatch::HandlerResult;
+use crate::sender::SenderKind;
use crate::shard::IggyShard;
use crate::shard::transmission::frame::ShardResponse;
use crate::shard::transmission::message::{ShardRequest, ShardRequestPayload};
@@ -25,7 +26,7 @@ use iggy_binary_protocol::WireName;
use iggy_binary_protocol::codec::WireEncode;
use iggy_binary_protocol::requests::consumer_groups::CreateConsumerGroupRequest;
use iggy_binary_protocol::responses::consumer_groups::ConsumerGroupResponse;
-use iggy_common::{IggyError, SenderKind};
+use iggy_common::IggyError;
use std::rc::Rc;
use tracing::{debug, instrument};
diff --git a/core/server/src/binary/handlers/consumer_groups/delete_consumer_group_handler.rs b/core/server/src/binary/handlers/consumer_groups/delete_consumer_group_handler.rs
index f51ebd00e..23f97d629 100644
--- a/core/server/src/binary/handlers/consumer_groups/delete_consumer_group_handler.rs
+++ b/core/server/src/binary/handlers/consumer_groups/delete_consumer_group_handler.rs
@@ -17,12 +17,13 @@
*/
use crate::binary::dispatch::HandlerResult;
+use crate::sender::SenderKind;
use crate::shard::IggyShard;
use crate::shard::transmission::frame::ShardResponse;
use crate::shard::transmission::message::{ShardRequest, ShardRequestPayload};
use crate::streaming::session::Session;
use iggy_binary_protocol::requests::consumer_groups::DeleteConsumerGroupRequest;
-use iggy_common::{IggyError, SenderKind};
+use iggy_common::IggyError;
use std::rc::Rc;
use tracing::{debug, instrument};
diff --git a/core/server/src/binary/handlers/consumer_groups/get_consumer_group_handler.rs b/core/server/src/binary/handlers/consumer_groups/get_consumer_group_handler.rs
index 3a54bd5ab..73693209b 100644
--- a/core/server/src/binary/handlers/consumer_groups/get_consumer_group_handler.rs
+++ b/core/server/src/binary/handlers/consumer_groups/get_consumer_group_handler.rs
@@ -17,6 +17,7 @@
*/
use crate::binary::dispatch::{HandlerResult, wire_id_to_identifier};
+use crate::sender::SenderKind;
use crate::shard::IggyShard;
use crate::streaming::session::Session;
use iggy_binary_protocol::WireName;
@@ -26,7 +27,6 @@ use iggy_binary_protocol::responses::consumer_groups::{
ConsumerGroupDetailsResponse, ConsumerGroupMemberResponse, ConsumerGroupResponse,
};
use iggy_common::IggyError;
-use iggy_common::SenderKind;
use std::rc::Rc;
use tracing::debug;
diff --git a/core/server/src/binary/handlers/consumer_groups/get_consumer_groups_handler.rs b/core/server/src/binary/handlers/consumer_groups/get_consumer_groups_handler.rs
index 2d64c8d00..86c2c07d9 100644
--- a/core/server/src/binary/handlers/consumer_groups/get_consumer_groups_handler.rs
+++ b/core/server/src/binary/handlers/consumer_groups/get_consumer_groups_handler.rs
@@ -17,6 +17,7 @@
*/
use crate::binary::dispatch::{HandlerResult, wire_id_to_identifier};
+use crate::sender::SenderKind;
use crate::shard::IggyShard;
use crate::streaming::session::Session;
use bytes::Bytes;
@@ -26,7 +27,7 @@ use iggy_binary_protocol::requests::consumer_groups::GetConsumerGroupsRequest;
use iggy_binary_protocol::responses::consumer_groups::{
ConsumerGroupResponse, GetConsumerGroupsResponse,
};
-use iggy_common::{IggyError, SenderKind};
+use iggy_common::IggyError;
use std::rc::Rc;
use tracing::debug;
diff --git a/core/server/src/binary/handlers/consumer_groups/join_consumer_group_handler.rs b/core/server/src/binary/handlers/consumer_groups/join_consumer_group_handler.rs
index 32d3947ca..15e9a68fc 100644
--- a/core/server/src/binary/handlers/consumer_groups/join_consumer_group_handler.rs
+++ b/core/server/src/binary/handlers/consumer_groups/join_consumer_group_handler.rs
@@ -17,12 +17,13 @@
*/
use crate::binary::dispatch::HandlerResult;
+use crate::sender::SenderKind;
use crate::shard::IggyShard;
use crate::shard::transmission::frame::ShardResponse;
use crate::shard::transmission::message::{ShardRequest, ShardRequestPayload};
use crate::streaming::session::Session;
use iggy_binary_protocol::requests::consumer_groups::JoinConsumerGroupRequest;
-use iggy_common::{IggyError, SenderKind};
+use iggy_common::IggyError;
use std::rc::Rc;
use tracing::{debug, instrument};
diff --git a/core/server/src/binary/handlers/consumer_groups/leave_consumer_group_handler.rs b/core/server/src/binary/handlers/consumer_groups/leave_consumer_group_handler.rs
index f4c976166..f12a99e9e 100644
--- a/core/server/src/binary/handlers/consumer_groups/leave_consumer_group_handler.rs
+++ b/core/server/src/binary/handlers/consumer_groups/leave_consumer_group_handler.rs
@@ -17,12 +17,13 @@
*/
use crate::binary::dispatch::HandlerResult;
+use crate::sender::SenderKind;
use crate::shard::IggyShard;
use crate::shard::transmission::frame::ShardResponse;
use crate::shard::transmission::message::{ShardRequest, ShardRequestPayload};
use crate::streaming::session::Session;
use iggy_binary_protocol::requests::consumer_groups::LeaveConsumerGroupRequest;
-use iggy_common::{IggyError, SenderKind};
+use iggy_common::IggyError;
use std::rc::Rc;
use tracing::{debug, instrument};
diff --git a/core/server/src/binary/handlers/consumer_offsets/delete_consumer_offset_handler.rs b/core/server/src/binary/handlers/consumer_offsets/delete_consumer_offset_handler.rs
index 459198a06..002adca98 100644
--- a/core/server/src/binary/handlers/consumer_offsets/delete_consumer_offset_handler.rs
+++ b/core/server/src/binary/handlers/consumer_offsets/delete_consumer_offset_handler.rs
@@ -18,11 +18,12 @@
use crate::binary::dispatch::{HandlerResult, wire_consumer_to_consumer, wire_id_to_identifier};
use crate::binary::handlers::consumer_offsets::COMPONENT;
+use crate::sender::SenderKind;
use crate::shard::IggyShard;
use crate::streaming::session::Session;
use err_trail::ErrContext;
use iggy_binary_protocol::requests::consumer_offsets::DeleteConsumerOffsetRequest;
-use iggy_common::{IggyError, SenderKind};
+use iggy_common::IggyError;
use std::rc::Rc;
use tracing::debug;
diff --git a/core/server/src/binary/handlers/consumer_offsets/get_consumer_offset_handler.rs b/core/server/src/binary/handlers/consumer_offsets/get_consumer_offset_handler.rs
index 97eefd75b..cbe83a72c 100644
--- a/core/server/src/binary/handlers/consumer_offsets/get_consumer_offset_handler.rs
+++ b/core/server/src/binary/handlers/consumer_offsets/get_consumer_offset_handler.rs
@@ -17,6 +17,7 @@
*/
use crate::binary::dispatch::{HandlerResult, wire_consumer_to_consumer, wire_id_to_identifier};
+use crate::sender::SenderKind;
use crate::shard::IggyShard;
use crate::shard::transmission::message::ResolvedTopic;
use crate::streaming::session::Session;
@@ -24,7 +25,6 @@ use iggy_binary_protocol::codec::WireEncode;
use iggy_binary_protocol::requests::consumer_offsets::GetConsumerOffsetRequest;
use iggy_binary_protocol::responses::consumer_offsets::ConsumerOffsetResponse;
use iggy_common::IggyError;
-use iggy_common::SenderKind;
use std::rc::Rc;
use tracing::debug;
diff --git a/core/server/src/binary/handlers/consumer_offsets/store_consumer_offset_handler.rs b/core/server/src/binary/handlers/consumer_offsets/store_consumer_offset_handler.rs
index 87eb4c5e5..5f1540d5f 100644
--- a/core/server/src/binary/handlers/consumer_offsets/store_consumer_offset_handler.rs
+++ b/core/server/src/binary/handlers/consumer_offsets/store_consumer_offset_handler.rs
@@ -20,12 +20,12 @@ use std::rc::Rc;
use crate::binary::dispatch::{HandlerResult, wire_consumer_to_consumer, wire_id_to_identifier};
use crate::binary::handlers::consumer_offsets::COMPONENT;
+use crate::sender::SenderKind;
use crate::shard::IggyShard;
use crate::streaming::session::Session;
use err_trail::ErrContext;
use iggy_binary_protocol::requests::consumer_offsets::StoreConsumerOffsetRequest;
use iggy_common::IggyError;
-use iggy_common::SenderKind;
use tracing::debug;
pub async fn handle_store_consumer_offset(
diff --git a/core/server/src/binary/handlers/messages/flush_unsaved_buffer_handler.rs b/core/server/src/binary/handlers/messages/flush_unsaved_buffer_handler.rs
index 79f8dbd4e..91927a4b3 100644
--- a/core/server/src/binary/handlers/messages/flush_unsaved_buffer_handler.rs
+++ b/core/server/src/binary/handlers/messages/flush_unsaved_buffer_handler.rs
@@ -18,12 +18,13 @@
use crate::binary::dispatch::{HandlerResult, wire_id_to_identifier};
use crate::binary::handlers::messages::COMPONENT;
+use crate::sender::SenderKind;
use crate::shard::IggyShard;
use crate::shard::transmission::message::ResolvedPartition;
use crate::streaming::session::Session;
use err_trail::ErrContext;
use iggy_binary_protocol::requests::messages::FlushUnsavedBufferRequest;
-use iggy_common::{IggyError, SenderKind};
+use iggy_common::IggyError;
use std::rc::Rc;
use tracing::{debug, instrument};
diff --git a/core/server/src/binary/handlers/messages/poll_messages_handler.rs b/core/server/src/binary/handlers/messages/poll_messages_handler.rs
index ca1b11c14..305e293f1 100644
--- a/core/server/src/binary/handlers/messages/poll_messages_handler.rs
+++ b/core/server/src/binary/handlers/messages/poll_messages_handler.rs
@@ -19,12 +19,13 @@
use crate::binary::dispatch::{
HandlerResult, wire_consumer_to_consumer, wire_id_to_identifier, wire_polling_to_strategy,
};
+use crate::sender::SenderKind;
use crate::shard::IggyShard;
use crate::shard::system::messages::PollingArgs;
use crate::streaming::session::Session;
use iggy_binary_protocol::requests::messages::PollMessagesRequest;
-use iggy_common::SenderKind;
-use iggy_common::{IggyError, PooledBuffer};
+use iggy_common::IggyError;
+use server_common::PooledBuffer;
use std::rc::Rc;
use tracing::{debug, trace};
diff --git a/core/server/src/binary/handlers/messages/send_messages_handler.rs b/core/server/src/binary/handlers/messages/send_messages_handler.rs
index 2964dcfce..b1757f5be 100644
--- a/core/server/src/binary/handlers/messages/send_messages_handler.rs
+++ b/core/server/src/binary/handlers/messages/send_messages_handler.rs
@@ -17,6 +17,7 @@
*/
use crate::binary::dispatch::HandlerResult;
+use crate::sender::SenderKind;
use crate::shard::IggyShard;
use crate::shard::transmission::message::{ResolvedPartition, ShardRequest, ShardRequestPayload};
use crate::streaming::segments::{IggyIndexesMut, IggyMessagesBatchMut};
@@ -24,12 +25,11 @@ use crate::streaming::session::Session;
use crate::streaming::topics;
use compio::buf::{IntoInner as _, IoBuf};
use iggy_common::Identifier;
-use iggy_common::PooledBuffer;
-use iggy_common::SenderKind;
use iggy_common::Sizeable;
-use iggy_common::sharding::IggyNamespace;
use iggy_common::{INDEX_SIZE, PartitioningKind};
use iggy_common::{IggyError, Partitioning, Validatable};
+use server_common::PooledBuffer;
+use server_common::sharding::IggyNamespace;
use std::rc::Rc;
use tracing::{debug, error, info, instrument};
diff --git a/core/server/src/binary/handlers/partitions/create_partitions_handler.rs b/core/server/src/binary/handlers/partitions/create_partitions_handler.rs
index a3548460a..fbe4c8bbd 100644
--- a/core/server/src/binary/handlers/partitions/create_partitions_handler.rs
+++ b/core/server/src/binary/handlers/partitions/create_partitions_handler.rs
@@ -17,13 +17,14 @@
*/
use crate::binary::dispatch::HandlerResult;
+use crate::sender::SenderKind;
use crate::shard::IggyShard;
use crate::shard::transmission::frame::ShardResponse;
use crate::shard::transmission::message::{ShardRequest, ShardRequestPayload};
use crate::streaming::session::Session;
use iggy_binary_protocol::MAX_PARTITIONS_PER_REQUEST;
use iggy_binary_protocol::requests::partitions::CreatePartitionsRequest;
-use iggy_common::{IggyError, SenderKind};
+use iggy_common::IggyError;
use std::rc::Rc;
use tracing::{debug, instrument};
diff --git a/core/server/src/binary/handlers/partitions/delete_partitions_handler.rs b/core/server/src/binary/handlers/partitions/delete_partitions_handler.rs
index 14fcd577e..aae224534 100644
--- a/core/server/src/binary/handlers/partitions/delete_partitions_handler.rs
+++ b/core/server/src/binary/handlers/partitions/delete_partitions_handler.rs
@@ -17,12 +17,13 @@
*/
use crate::binary::dispatch::HandlerResult;
+use crate::sender::SenderKind;
use crate::shard::IggyShard;
use crate::shard::transmission::frame::ShardResponse;
use crate::shard::transmission::message::{ShardRequest, ShardRequestPayload};
use crate::streaming::session::Session;
use iggy_binary_protocol::requests::partitions::DeletePartitionsRequest;
-use iggy_common::{IggyError, SenderKind};
+use iggy_common::IggyError;
use std::rc::Rc;
use tracing::{debug, instrument};
diff --git a/core/server/src/binary/handlers/personal_access_tokens/create_personal_access_token_handler.rs b/core/server/src/binary/handlers/personal_access_tokens/create_personal_access_token_handler.rs
index 6bfc77fa0..45a7f2aaa 100644
--- a/core/server/src/binary/handlers/personal_access_tokens/create_personal_access_token_handler.rs
+++ b/core/server/src/binary/handlers/personal_access_tokens/create_personal_access_token_handler.rs
@@ -17,6 +17,7 @@
*/
use crate::binary::dispatch::HandlerResult;
+use crate::sender::SenderKind;
use crate::shard::IggyShard;
use crate::shard::transmission::frame::ShardResponse;
use crate::shard::transmission::message::{ShardRequest, ShardRequestPayload};
@@ -25,10 +26,10 @@ use iggy_binary_protocol::WireName;
use iggy_binary_protocol::codec::WireEncode;
use iggy_binary_protocol::requests::personal_access_tokens::CreatePersonalAccessTokenRequest;
use iggy_binary_protocol::responses::personal_access_tokens::RawPersonalAccessTokenResponse;
+use iggy_common::IggyError;
use iggy_common::defaults::{
MAX_PERSONAL_ACCESS_TOKEN_NAME_LENGTH, MIN_PERSONAL_ACCESS_TOKEN_NAME_LENGTH,
};
-use iggy_common::{IggyError, SenderKind};
use std::rc::Rc;
use tracing::{debug, instrument};
diff --git a/core/server/src/binary/handlers/personal_access_tokens/delete_personal_access_token_handler.rs b/core/server/src/binary/handlers/personal_access_tokens/delete_personal_access_token_handler.rs
index 06d29fb98..ed1f66582 100644
--- a/core/server/src/binary/handlers/personal_access_tokens/delete_personal_access_token_handler.rs
+++ b/core/server/src/binary/handlers/personal_access_tokens/delete_personal_access_token_handler.rs
@@ -17,12 +17,13 @@
*/
use crate::binary::dispatch::HandlerResult;
+use crate::sender::SenderKind;
use crate::shard::IggyShard;
use crate::shard::transmission::frame::ShardResponse;
use crate::shard::transmission::message::{ShardRequest, ShardRequestPayload};
use crate::streaming::session::Session;
use iggy_binary_protocol::requests::personal_access_tokens::DeletePersonalAccessTokenRequest;
-use iggy_common::{IggyError, SenderKind};
+use iggy_common::IggyError;
use std::rc::Rc;
use tracing::{debug, instrument};
diff --git a/core/server/src/binary/handlers/personal_access_tokens/get_personal_access_tokens_handler.rs b/core/server/src/binary/handlers/personal_access_tokens/get_personal_access_tokens_handler.rs
index b04e1490c..2adcb4f55 100644
--- a/core/server/src/binary/handlers/personal_access_tokens/get_personal_access_tokens_handler.rs
+++ b/core/server/src/binary/handlers/personal_access_tokens/get_personal_access_tokens_handler.rs
@@ -18,6 +18,7 @@
use crate::binary::dispatch::HandlerResult;
use crate::binary::handlers::personal_access_tokens::COMPONENT;
+use crate::sender::SenderKind;
use crate::shard::IggyShard;
use crate::streaming::session::Session;
use err_trail::ErrContext;
@@ -27,7 +28,6 @@ use iggy_binary_protocol::responses::personal_access_tokens::{
GetPersonalAccessTokensResponse, PersonalAccessTokenResponse,
};
use iggy_common::IggyError;
-use iggy_common::SenderKind;
use std::rc::Rc;
use tracing::debug;
diff --git a/core/server/src/binary/handlers/personal_access_tokens/login_with_personal_access_token_handler.rs b/core/server/src/binary/handlers/personal_access_tokens/login_with_personal_access_token_handler.rs
index 81fa44eb3..9da028b90 100644
--- a/core/server/src/binary/handlers/personal_access_tokens/login_with_personal_access_token_handler.rs
+++ b/core/server/src/binary/handlers/personal_access_tokens/login_with_personal_access_token_handler.rs
@@ -18,13 +18,14 @@
use crate::binary::dispatch::HandlerResult;
use crate::binary::handlers::personal_access_tokens::COMPONENT;
+use crate::sender::SenderKind;
use crate::shard::IggyShard;
use crate::streaming::session::Session;
use err_trail::ErrContext;
use iggy_binary_protocol::codec::WireEncode;
use iggy_binary_protocol::requests::personal_access_tokens::LoginWithPersonalAccessTokenRequest;
use iggy_binary_protocol::responses::users::IdentityResponse;
-use iggy_common::{IggyError, SenderKind};
+use iggy_common::IggyError;
use std::rc::Rc;
use tracing::{debug, instrument};
diff --git a/core/server/src/binary/handlers/segments/delete_segments_handler.rs b/core/server/src/binary/handlers/segments/delete_segments_handler.rs
index cf55d875c..548a7b4c0 100644
--- a/core/server/src/binary/handlers/segments/delete_segments_handler.rs
+++ b/core/server/src/binary/handlers/segments/delete_segments_handler.rs
@@ -17,13 +17,14 @@
*/
use crate::binary::dispatch::{HandlerResult, wire_id_to_identifier};
+use crate::sender::SenderKind;
use crate::shard::IggyShard;
use crate::shard::transmission::frame::ShardResponse;
use crate::shard::transmission::message::{ShardRequest, ShardRequestPayload};
use crate::streaming::session::Session;
use iggy_binary_protocol::requests::segments::DeleteSegmentsRequest;
-use iggy_common::sharding::IggyNamespace;
-use iggy_common::{IggyError, SenderKind};
+use iggy_common::IggyError;
+use server_common::sharding::IggyNamespace;
use std::rc::Rc;
use tracing::{debug, instrument};
diff --git a/core/server/src/binary/handlers/streams/create_stream_handler.rs b/core/server/src/binary/handlers/streams/create_stream_handler.rs
index e3544edd8..c7fca3045 100644
--- a/core/server/src/binary/handlers/streams/create_stream_handler.rs
+++ b/core/server/src/binary/handlers/streams/create_stream_handler.rs
@@ -17,6 +17,7 @@
*/
use crate::binary::dispatch::HandlerResult;
+use crate::sender::SenderKind;
use crate::shard::IggyShard;
use crate::shard::transmission::frame::ShardResponse;
use crate::shard::transmission::message::{ShardRequest, ShardRequestPayload};
@@ -25,7 +26,7 @@ use iggy_binary_protocol::WireName;
use iggy_binary_protocol::codec::WireEncode;
use iggy_binary_protocol::requests::streams::CreateStreamRequest;
use iggy_binary_protocol::responses::streams::StreamResponse;
-use iggy_common::{IggyError, SenderKind};
+use iggy_common::IggyError;
use std::rc::Rc;
use tracing::{debug, instrument};
diff --git a/core/server/src/binary/handlers/streams/delete_stream_handler.rs b/core/server/src/binary/handlers/streams/delete_stream_handler.rs
index 59ffd1e15..5c68939a1 100644
--- a/core/server/src/binary/handlers/streams/delete_stream_handler.rs
+++ b/core/server/src/binary/handlers/streams/delete_stream_handler.rs
@@ -17,12 +17,13 @@
*/
use crate::binary::dispatch::HandlerResult;
+use crate::sender::SenderKind;
use crate::shard::IggyShard;
use crate::shard::transmission::frame::ShardResponse;
use crate::shard::transmission::message::{ShardRequest, ShardRequestPayload};
use crate::streaming::session::Session;
use iggy_binary_protocol::requests::streams::DeleteStreamRequest;
-use iggy_common::{IggyError, SenderKind};
+use iggy_common::IggyError;
use std::rc::Rc;
use tracing::{debug, instrument};
diff --git a/core/server/src/binary/handlers/streams/get_stream_handler.rs b/core/server/src/binary/handlers/streams/get_stream_handler.rs
index fd451a71d..319bbf787 100644
--- a/core/server/src/binary/handlers/streams/get_stream_handler.rs
+++ b/core/server/src/binary/handlers/streams/get_stream_handler.rs
@@ -18,6 +18,7 @@
use crate::binary::dispatch::{HandlerResult, wire_id_to_identifier};
use crate::metadata::{StreamMeta, TopicMeta};
+use crate::sender::SenderKind;
use crate::shard::IggyShard;
use crate::streaming::session::Session;
use iggy_binary_protocol::WireName;
@@ -26,7 +27,6 @@ use iggy_binary_protocol::requests::streams::GetStreamRequest;
use iggy_binary_protocol::responses::streams::StreamResponse;
use iggy_binary_protocol::responses::streams::get_stream::{GetStreamResponse, TopicHeader};
use iggy_common::IggyError;
-use iggy_common::SenderKind;
use std::rc::Rc;
use tracing::debug;
diff --git a/core/server/src/binary/handlers/streams/get_streams_handler.rs b/core/server/src/binary/handlers/streams/get_streams_handler.rs
index ca5524378..5c843036c 100644
--- a/core/server/src/binary/handlers/streams/get_streams_handler.rs
+++ b/core/server/src/binary/handlers/streams/get_streams_handler.rs
@@ -18,6 +18,7 @@
use super::get_stream_handler::compute_stream_stats;
use crate::binary::dispatch::HandlerResult;
+use crate::sender::SenderKind;
use crate::shard::IggyShard;
use crate::streaming::session::Session;
use iggy_binary_protocol::WireName;
@@ -25,7 +26,6 @@ use iggy_binary_protocol::codec::WireEncode;
use iggy_binary_protocol::responses::streams::StreamResponse;
use iggy_binary_protocol::responses::streams::get_streams::GetStreamsResponse;
use iggy_common::IggyError;
-use iggy_common::SenderKind;
use std::rc::Rc;
use tracing::debug;
diff --git a/core/server/src/binary/handlers/streams/purge_stream_handler.rs b/core/server/src/binary/handlers/streams/purge_stream_handler.rs
index a2695d7a6..e9c093b24 100644
--- a/core/server/src/binary/handlers/streams/purge_stream_handler.rs
+++ b/core/server/src/binary/handlers/streams/purge_stream_handler.rs
@@ -17,12 +17,13 @@
*/
use crate::binary::dispatch::HandlerResult;
+use crate::sender::SenderKind;
use crate::shard::IggyShard;
use crate::shard::transmission::frame::ShardResponse;
use crate::shard::transmission::message::{ShardRequest, ShardRequestPayload};
use crate::streaming::session::Session;
use iggy_binary_protocol::requests::streams::PurgeStreamRequest;
-use iggy_common::{IggyError, SenderKind};
+use iggy_common::IggyError;
use std::rc::Rc;
use tracing::{debug, instrument};
diff --git a/core/server/src/binary/handlers/streams/update_stream_handler.rs b/core/server/src/binary/handlers/streams/update_stream_handler.rs
index e702f862f..ec7718e92 100644
--- a/core/server/src/binary/handlers/streams/update_stream_handler.rs
+++ b/core/server/src/binary/handlers/streams/update_stream_handler.rs
@@ -17,12 +17,13 @@
*/
use crate::binary::dispatch::HandlerResult;
+use crate::sender::SenderKind;
use crate::shard::IggyShard;
use crate::shard::transmission::frame::ShardResponse;
use crate::shard::transmission::message::{ShardRequest, ShardRequestPayload};
use crate::streaming::session::Session;
use iggy_binary_protocol::requests::streams::UpdateStreamRequest;
-use iggy_common::{IggyError, SenderKind};
+use iggy_common::IggyError;
use std::rc::Rc;
use tracing::{debug, instrument};
diff --git a/core/server/src/binary/handlers/system/get_client_handler.rs b/core/server/src/binary/handlers/system/get_client_handler.rs
index 594cf8373..d01a1268a 100644
--- a/core/server/src/binary/handlers/system/get_client_handler.rs
+++ b/core/server/src/binary/handlers/system/get_client_handler.rs
@@ -18,12 +18,12 @@
use super::get_me_handler::build_client_details_response;
use crate::binary::dispatch::HandlerResult;
+use crate::sender::SenderKind;
use crate::shard::IggyShard;
use crate::streaming::session::Session;
use iggy_binary_protocol::codec::WireEncode;
use iggy_binary_protocol::requests::system::GetClientRequest;
use iggy_common::IggyError;
-use iggy_common::SenderKind;
use std::rc::Rc;
use tracing::debug;
diff --git a/core/server/src/binary/handlers/system/get_clients_handler.rs b/core/server/src/binary/handlers/system/get_clients_handler.rs
index cceda4622..dd9aa7967 100644
--- a/core/server/src/binary/handlers/system/get_clients_handler.rs
+++ b/core/server/src/binary/handlers/system/get_clients_handler.rs
@@ -18,12 +18,12 @@
use super::get_me_handler::build_client_response;
use crate::binary::dispatch::HandlerResult;
+use crate::sender::SenderKind;
use crate::shard::IggyShard;
use crate::streaming::session::Session;
use iggy_binary_protocol::codec::WireEncode;
use iggy_binary_protocol::responses::clients::GetClientsResponse;
use iggy_common::IggyError;
-use iggy_common::SenderKind;
use std::rc::Rc;
use tracing::debug;
diff --git a/core/server/src/binary/handlers/system/get_me_handler.rs b/core/server/src/binary/handlers/system/get_me_handler.rs
index 0d635bf6a..ffa657568 100644
--- a/core/server/src/binary/handlers/system/get_me_handler.rs
+++ b/core/server/src/binary/handlers/system/get_me_handler.rs
@@ -17,6 +17,7 @@
*/
use crate::binary::dispatch::HandlerResult;
+use crate::sender::SenderKind;
use crate::shard::IggyShard;
use crate::streaming::clients::client_manager::Client;
use crate::streaming::session::Session;
@@ -25,7 +26,6 @@ use iggy_binary_protocol::responses::clients::{
ClientDetailsResponse, ClientResponse, ConsumerGroupInfoResponse,
};
use iggy_common::IggyError;
-use iggy_common::SenderKind;
use std::rc::Rc;
pub async fn handle_get_me(
diff --git a/core/server/src/binary/handlers/system/get_snapshot_handler.rs b/core/server/src/binary/handlers/system/get_snapshot_handler.rs
index 4eabf63ac..a67083528 100644
--- a/core/server/src/binary/handlers/system/get_snapshot_handler.rs
+++ b/core/server/src/binary/handlers/system/get_snapshot_handler.rs
@@ -17,11 +17,12 @@
*/
use crate::binary::dispatch::HandlerResult;
+use crate::sender::SenderKind;
use crate::shard::IggyShard;
use crate::streaming::session::Session;
use bytes::Bytes;
use iggy_binary_protocol::requests::system::GetSnapshotRequest;
-use iggy_common::{IggyError, SenderKind, SnapshotCompression, SystemSnapshotType};
+use iggy_common::{IggyError, SnapshotCompression, SystemSnapshotType};
use std::rc::Rc;
use tracing::debug;
diff --git a/core/server/src/binary/handlers/system/get_stats_handler.rs b/core/server/src/binary/handlers/system/get_stats_handler.rs
index 26aa2ec23..778019ae3 100644
--- a/core/server/src/binary/handlers/system/get_stats_handler.rs
+++ b/core/server/src/binary/handlers/system/get_stats_handler.rs
@@ -17,6 +17,7 @@
*/
use crate::binary::dispatch::HandlerResult;
+use crate::sender::SenderKind;
use crate::shard::IggyShard;
use crate::shard::transmission::frame::ShardResponse;
use crate::shard::transmission::message::{ShardRequest, ShardRequestPayload};
@@ -24,7 +25,6 @@ use crate::streaming::session::Session;
use iggy_binary_protocol::codec::WireEncode;
use iggy_binary_protocol::responses::system::get_stats::{CacheMetricEntry, StatsResponse};
use iggy_common::IggyError;
-use iggy_common::SenderKind;
use std::rc::Rc;
use tracing::debug;
diff --git a/core/server/src/binary/handlers/system/ping_handler.rs b/core/server/src/binary/handlers/system/ping_handler.rs
index 6633cd0b0..8a2d22668 100644
--- a/core/server/src/binary/handlers/system/ping_handler.rs
+++ b/core/server/src/binary/handlers/system/ping_handler.rs
@@ -17,11 +17,11 @@
*/
use crate::binary::dispatch::HandlerResult;
+use crate::sender::SenderKind;
use crate::shard::IggyShard;
use crate::streaming::session::Session;
use iggy_common::IggyError;
use iggy_common::IggyTimestamp;
-use iggy_common::SenderKind;
use std::rc::Rc;
use tracing::debug;
diff --git a/core/server/src/binary/handlers/topics/create_topic_handler.rs b/core/server/src/binary/handlers/topics/create_topic_handler.rs
index eb5da9757..900bf6968 100644
--- a/core/server/src/binary/handlers/topics/create_topic_handler.rs
+++ b/core/server/src/binary/handlers/topics/create_topic_handler.rs
@@ -17,6 +17,7 @@
*/
use crate::binary::dispatch::HandlerResult;
+use crate::sender::SenderKind;
use crate::shard::IggyShard;
use crate::shard::transmission::frame::ShardResponse;
use crate::shard::transmission::message::{ShardRequest, ShardRequestPayload};
@@ -28,7 +29,7 @@ use iggy_binary_protocol::codec::WireEncode;
use iggy_binary_protocol::requests::topics::CreateTopicRequest;
use iggy_binary_protocol::responses::streams::get_stream::TopicHeader;
use iggy_binary_protocol::responses::topics::get_topic::PartitionResponse;
-use iggy_common::{IggyError, SenderKind};
+use iggy_common::IggyError;
use std::rc::Rc;
use tracing::{debug, instrument};
diff --git a/core/server/src/binary/handlers/topics/delete_topic_handler.rs b/core/server/src/binary/handlers/topics/delete_topic_handler.rs
index 8f4c93f60..1cb33dae6 100644
--- a/core/server/src/binary/handlers/topics/delete_topic_handler.rs
+++ b/core/server/src/binary/handlers/topics/delete_topic_handler.rs
@@ -17,12 +17,13 @@
*/
use crate::binary::dispatch::HandlerResult;
+use crate::sender::SenderKind;
use crate::shard::IggyShard;
use crate::shard::transmission::frame::ShardResponse;
use crate::shard::transmission::message::{ShardRequest, ShardRequestPayload};
use crate::streaming::session::Session;
use iggy_binary_protocol::requests::topics::DeleteTopicRequest;
-use iggy_common::{IggyError, SenderKind};
+use iggy_common::IggyError;
use std::rc::Rc;
use tracing::{debug, instrument};
diff --git a/core/server/src/binary/handlers/topics/get_topic_handler.rs b/core/server/src/binary/handlers/topics/get_topic_handler.rs
index 5e44a4b4a..973c328da 100644
--- a/core/server/src/binary/handlers/topics/get_topic_handler.rs
+++ b/core/server/src/binary/handlers/topics/get_topic_handler.rs
@@ -19,12 +19,13 @@
use crate::binary::dispatch::{HandlerResult, wire_id_to_identifier};
use crate::binary::handlers::streams::get_stream_handler::build_topic_header;
use crate::metadata::TopicMeta;
+use crate::sender::SenderKind;
use crate::shard::IggyShard;
use crate::streaming::session::Session;
use iggy_binary_protocol::codec::WireEncode;
use iggy_binary_protocol::requests::topics::GetTopicRequest;
use iggy_binary_protocol::responses::topics::get_topic::{GetTopicResponse, PartitionResponse};
-use iggy_common::{IggyError, SenderKind};
+use iggy_common::IggyError;
use std::rc::Rc;
use tracing::debug;
diff --git a/core/server/src/binary/handlers/topics/get_topics_handler.rs b/core/server/src/binary/handlers/topics/get_topics_handler.rs
index e8d0a3d30..b46a93c2f 100644
--- a/core/server/src/binary/handlers/topics/get_topics_handler.rs
+++ b/core/server/src/binary/handlers/topics/get_topics_handler.rs
@@ -18,12 +18,13 @@
use crate::binary::dispatch::{HandlerResult, wire_id_to_identifier};
use crate::binary::handlers::streams::get_stream_handler::build_topic_header;
+use crate::sender::SenderKind;
use crate::shard::IggyShard;
use crate::streaming::session::Session;
use iggy_binary_protocol::codec::WireEncode;
use iggy_binary_protocol::requests::topics::GetTopicsRequest;
use iggy_binary_protocol::responses::topics::get_topics::GetTopicsResponse;
-use iggy_common::{IggyError, SenderKind};
+use iggy_common::IggyError;
use std::rc::Rc;
use tracing::debug;
diff --git a/core/server/src/binary/handlers/topics/purge_topic_handler.rs b/core/server/src/binary/handlers/topics/purge_topic_handler.rs
index 6b8430b84..9b07e3631 100644
--- a/core/server/src/binary/handlers/topics/purge_topic_handler.rs
+++ b/core/server/src/binary/handlers/topics/purge_topic_handler.rs
@@ -17,12 +17,13 @@
*/
use crate::binary::dispatch::HandlerResult;
+use crate::sender::SenderKind;
use crate::shard::IggyShard;
use crate::shard::transmission::frame::ShardResponse;
use crate::shard::transmission::message::{ShardRequest, ShardRequestPayload};
use crate::streaming::session::Session;
use iggy_binary_protocol::requests::topics::PurgeTopicRequest;
-use iggy_common::{IggyError, SenderKind};
+use iggy_common::IggyError;
use std::rc::Rc;
use tracing::{debug, instrument};
diff --git a/core/server/src/binary/handlers/topics/update_topic_handler.rs b/core/server/src/binary/handlers/topics/update_topic_handler.rs
index 31796dfc3..a78a1e73d 100644
--- a/core/server/src/binary/handlers/topics/update_topic_handler.rs
+++ b/core/server/src/binary/handlers/topics/update_topic_handler.rs
@@ -17,12 +17,13 @@
*/
use crate::binary::dispatch::HandlerResult;
+use crate::sender::SenderKind;
use crate::shard::IggyShard;
use crate::shard::transmission::frame::ShardResponse;
use crate::shard::transmission::message::{ShardRequest, ShardRequestPayload};
use crate::streaming::session::Session;
use iggy_binary_protocol::requests::topics::UpdateTopicRequest;
-use iggy_common::{IggyError, SenderKind};
+use iggy_common::IggyError;
use std::rc::Rc;
use tracing::{debug, instrument};
diff --git a/core/server/src/binary/handlers/users/change_password_handler.rs b/core/server/src/binary/handlers/users/change_password_handler.rs
index 860b0d82e..eceae9236 100644
--- a/core/server/src/binary/handlers/users/change_password_handler.rs
+++ b/core/server/src/binary/handlers/users/change_password_handler.rs
@@ -17,13 +17,14 @@
*/
use crate::binary::dispatch::HandlerResult;
+use crate::sender::SenderKind;
use crate::shard::IggyShard;
use crate::shard::transmission::frame::ShardResponse;
use crate::shard::transmission::message::{ShardRequest, ShardRequestPayload};
use crate::streaming::session::Session;
use iggy_binary_protocol::requests::users::ChangePasswordRequest;
+use iggy_common::IggyError;
use iggy_common::defaults::{MAX_PASSWORD_LENGTH, MIN_PASSWORD_LENGTH};
-use iggy_common::{IggyError, SenderKind};
use std::rc::Rc;
use tracing::{debug, instrument};
diff --git a/core/server/src/binary/handlers/users/create_user_handler.rs b/core/server/src/binary/handlers/users/create_user_handler.rs
index 7532523a4..c3d9670e6 100644
--- a/core/server/src/binary/handlers/users/create_user_handler.rs
+++ b/core/server/src/binary/handlers/users/create_user_handler.rs
@@ -17,6 +17,7 @@
*/
use crate::binary::dispatch::HandlerResult;
+use crate::sender::SenderKind;
use crate::shard::IggyShard;
use crate::shard::transmission::frame::ShardResponse;
use crate::shard::transmission::message::{ShardRequest, ShardRequestPayload};
@@ -25,11 +26,11 @@ use iggy_binary_protocol::WireName;
use iggy_binary_protocol::codec::WireEncode;
use iggy_binary_protocol::requests::users::CreateUserRequest;
use iggy_binary_protocol::responses::users::{UserDetailsResponse, UserResponse};
+use iggy_common::IggyError;
use iggy_common::defaults::{
MAX_PASSWORD_LENGTH, MAX_USERNAME_LENGTH, MIN_PASSWORD_LENGTH, MIN_USERNAME_LENGTH,
};
use iggy_common::wire_conversions::permissions_to_wire;
-use iggy_common::{IggyError, SenderKind};
use std::rc::Rc;
use tracing::{debug, instrument};
diff --git a/core/server/src/binary/handlers/users/delete_user_handler.rs b/core/server/src/binary/handlers/users/delete_user_handler.rs
index c92ddfba1..48497d193 100644
--- a/core/server/src/binary/handlers/users/delete_user_handler.rs
+++ b/core/server/src/binary/handlers/users/delete_user_handler.rs
@@ -17,12 +17,13 @@
*/
use crate::binary::dispatch::HandlerResult;
+use crate::sender::SenderKind;
use crate::shard::IggyShard;
use crate::shard::transmission::frame::ShardResponse;
use crate::shard::transmission::message::{ShardRequest, ShardRequestPayload};
use crate::streaming::session::Session;
use iggy_binary_protocol::requests::users::DeleteUserRequest;
-use iggy_common::{IggyError, SenderKind};
+use iggy_common::IggyError;
use std::rc::Rc;
use tracing::{debug, instrument};
diff --git a/core/server/src/binary/handlers/users/get_user_handler.rs b/core/server/src/binary/handlers/users/get_user_handler.rs
index f7415c8f5..832715607 100644
--- a/core/server/src/binary/handlers/users/get_user_handler.rs
+++ b/core/server/src/binary/handlers/users/get_user_handler.rs
@@ -17,6 +17,7 @@
*/
use crate::binary::dispatch::{HandlerResult, wire_id_to_identifier};
+use crate::sender::SenderKind;
use crate::shard::IggyShard;
use crate::streaming::session::Session;
use iggy_binary_protocol::WireName;
@@ -24,7 +25,6 @@ use iggy_binary_protocol::codec::WireEncode;
use iggy_binary_protocol::requests::users::GetUserRequest;
use iggy_binary_protocol::responses::users::{UserDetailsResponse, UserResponse};
use iggy_common::IggyError;
-use iggy_common::SenderKind;
use iggy_common::wire_conversions::permissions_to_wire;
use std::rc::Rc;
use tracing::debug;
diff --git a/core/server/src/binary/handlers/users/get_users_handler.rs b/core/server/src/binary/handlers/users/get_users_handler.rs
index aaccdeaa5..11671b367 100644
--- a/core/server/src/binary/handlers/users/get_users_handler.rs
+++ b/core/server/src/binary/handlers/users/get_users_handler.rs
@@ -17,13 +17,13 @@
*/
use crate::binary::dispatch::HandlerResult;
+use crate::sender::SenderKind;
use crate::shard::IggyShard;
use crate::streaming::session::Session;
use iggy_binary_protocol::WireName;
use iggy_binary_protocol::codec::WireEncode;
use iggy_binary_protocol::responses::users::{GetUsersResponse, UserResponse};
use iggy_common::IggyError;
-use iggy_common::SenderKind;
use std::rc::Rc;
use tracing::debug;
diff --git a/core/server/src/binary/handlers/users/login_user_handler.rs b/core/server/src/binary/handlers/users/login_user_handler.rs
index aa0924427..718a03a1b 100644
--- a/core/server/src/binary/handlers/users/login_user_handler.rs
+++ b/core/server/src/binary/handlers/users/login_user_handler.rs
@@ -18,6 +18,7 @@
use crate::binary::dispatch::HandlerResult;
use crate::binary::handlers::users::COMPONENT;
+use crate::sender::SenderKind;
use crate::shard::IggyShard;
use crate::streaming::session::Session;
use err_trail::ErrContext;
@@ -25,7 +26,6 @@ use iggy_binary_protocol::codec::WireEncode;
use iggy_binary_protocol::requests::users::LoginUserRequest;
use iggy_binary_protocol::responses::users::IdentityResponse;
use iggy_common::IggyError;
-use iggy_common::SenderKind;
use iggy_common::defaults::{
MAX_PASSWORD_LENGTH, MAX_USERNAME_LENGTH, MIN_PASSWORD_LENGTH, MIN_USERNAME_LENGTH,
};
diff --git a/core/server/src/binary/handlers/users/logout_user_handler.rs b/core/server/src/binary/handlers/users/logout_user_handler.rs
index d3f5111f7..20c71327b 100644
--- a/core/server/src/binary/handlers/users/logout_user_handler.rs
+++ b/core/server/src/binary/handlers/users/logout_user_handler.rs
@@ -18,10 +18,11 @@
use crate::binary::dispatch::HandlerResult;
use crate::binary::handlers::users::COMPONENT;
+use crate::sender::SenderKind;
use crate::shard::IggyShard;
use crate::streaming::session::Session;
use err_trail::ErrContext;
-use iggy_common::{IggyError, SenderKind};
+use iggy_common::IggyError;
use std::rc::Rc;
use tracing::{debug, info, instrument};
diff --git a/core/server/src/binary/handlers/users/update_permissions_handler.rs b/core/server/src/binary/handlers/users/update_permissions_handler.rs
index 8c9df44a1..8d0efd054 100644
--- a/core/server/src/binary/handlers/users/update_permissions_handler.rs
+++ b/core/server/src/binary/handlers/users/update_permissions_handler.rs
@@ -17,12 +17,13 @@
*/
use crate::binary::dispatch::HandlerResult;
+use crate::sender::SenderKind;
use crate::shard::IggyShard;
use crate::shard::transmission::frame::ShardResponse;
use crate::shard::transmission::message::{ShardRequest, ShardRequestPayload};
use crate::streaming::session::Session;
use iggy_binary_protocol::requests::users::UpdatePermissionsRequest;
-use iggy_common::{IggyError, SenderKind};
+use iggy_common::IggyError;
use std::rc::Rc;
use tracing::{debug, instrument};
diff --git a/core/server/src/binary/handlers/users/update_user_handler.rs b/core/server/src/binary/handlers/users/update_user_handler.rs
index 23b2e51cd..e33ade822 100644
--- a/core/server/src/binary/handlers/users/update_user_handler.rs
+++ b/core/server/src/binary/handlers/users/update_user_handler.rs
@@ -17,13 +17,14 @@
*/
use crate::binary::dispatch::HandlerResult;
+use crate::sender::SenderKind;
use crate::shard::IggyShard;
use crate::shard::transmission::frame::ShardResponse;
use crate::shard::transmission::message::{ShardRequest, ShardRequestPayload};
use crate::streaming::session::Session;
use iggy_binary_protocol::requests::users::UpdateUserRequest;
+use iggy_common::IggyError;
use iggy_common::defaults::{MAX_USERNAME_LENGTH, MIN_USERNAME_LENGTH};
-use iggy_common::{IggyError, SenderKind};
use std::rc::Rc;
use tracing::{debug, instrument};
diff --git a/core/server/src/http/messages.rs b/core/server/src/http/messages.rs
index f12e36869..9c719d5b9 100644
--- a/core/server/src/http/messages.rs
+++ b/core/server/src/http/messages.rs
@@ -29,11 +29,12 @@ use axum::http::StatusCode;
use axum::routing::get;
use axum::{Extension, Json, Router, debug_handler};
use err_trail::ErrContext;
+use iggy_common::Identifier;
+use iggy_common::Validatable;
use iggy_common::{Consumer, PollMessages, SendMessages};
-use iggy_common::{Identifier, PooledBuffer};
use iggy_common::{IggyError, IggyMessagesBatch, PolledMessages};
-use iggy_common::{IggyIndexesMut, Validatable};
use send_wrapper::SendWrapper;
+use server_common::{IggyIndexesMut, PooledBuffer};
use std::sync::Arc;
use tracing::instrument;
diff --git a/core/server/src/http/segments.rs b/core/server/src/http/segments.rs
index c06f119eb..bf2beeb48 100644
--- a/core/server/src/http/segments.rs
+++ b/core/server/src/http/segments.rs
@@ -31,9 +31,9 @@ use iggy_binary_protocol::requests::segments::DeleteSegmentsRequest;
use iggy_common::Identifier;
use iggy_common::Validatable;
use iggy_common::delete_segments::DeleteSegments;
-use iggy_common::sharding::IggyNamespace;
use iggy_common::wire_conversions::identifier_to_wire;
use send_wrapper::SendWrapper;
+use server_common::sharding::IggyNamespace;
use std::sync::Arc;
use tracing::instrument;
diff --git a/core/server/src/lib.rs b/core/server/src/lib.rs
index b4bac970d..9177e27af 100644
--- a/core/server/src/lib.rs
+++ b/core/server/src/lib.rs
@@ -40,6 +40,7 @@ pub mod io;
pub mod log;
pub mod metadata;
pub mod quic;
+pub mod sender;
pub mod server_error;
pub mod shard;
pub mod shard_allocator;
diff --git a/core/server/src/main.rs b/core/server/src/main.rs
index eba3d3536..cba421b01 100644
--- a/core/server/src/main.rs
+++ b/core/server/src/main.rs
@@ -24,8 +24,7 @@ use dotenvy::dotenv;
use err_trail::ErrContext;
use figlet_rs::FIGlet;
use iggy_common::SemanticVersion;
-use iggy_common::sharding::{IggyNamespace, LocalIdx, PartitionLocation, ShardId};
-use iggy_common::{Aes256GcmEncryptor, EncryptorKind, IggyError, MemoryPool};
+use iggy_common::{Aes256GcmEncryptor, EncryptorKind, IggyError};
use server::SEMANTIC_VERSION;
use server::args::Args;
use server::bootstrap::{
@@ -49,6 +48,8 @@ use server::streaming::clients::client_manager::{Client, ClientManager};
use server::streaming::diagnostics::metrics::Metrics;
use server::streaming::storage::SystemStorage;
use server::streaming::utils::ptr::EternalPtr;
+use server_common::MemoryPool;
+use server_common::sharding::{IggyNamespace, LocalIdx, PartitionLocation, ShardId};
use std::panic::AssertUnwindSafe;
use std::rc::Rc;
use std::str::FromStr;
diff --git a/core/server/src/metadata/reader.rs b/core/server/src/metadata/reader.rs
index ba60e177f..59d6fcd0b 100644
--- a/core/server/src/metadata/reader.rs
+++ b/core/server/src/metadata/reader.rs
@@ -24,11 +24,11 @@ use crate::streaming::partitions::consumer_group_offsets::ConsumerGroupOffsets;
use crate::streaming::partitions::consumer_offsets::ConsumerOffsets;
use crate::streaming::polling_consumer::PollingConsumer;
use crate::streaming::stats::{PartitionStats, StreamStats, TopicStats};
-use iggy_common::sharding::IggyNamespace;
use iggy_common::{
IdKind, Identifier, IggyError, IggyExpiry, IggyTimestamp, MaxTopicSize, PersonalAccessToken,
};
use left_right::ReadGuard;
+use server_common::sharding::IggyNamespace;
use std::sync::Arc;
use std::sync::atomic::Ordering;
diff --git a/core/server/src/quic/listener.rs b/core/server/src/quic/listener.rs
index 91287f198..f8dde57fc 100644
--- a/core/server/src/quic/listener.rs
+++ b/core/server/src/quic/listener.rs
@@ -17,6 +17,7 @@
*/
use crate::binary::dispatch::{self, HandlerResult, MAX_CONTROL_FRAME_PAYLOAD};
+use crate::sender::SenderKind;
use crate::server_error::ConnectionError;
use crate::shard::IggyShard;
use crate::shard::task_registry::ShutdownToken;
@@ -27,7 +28,7 @@ use compio::quic::{Connection, Endpoint, RecvStream, SendStream};
use futures::FutureExt;
use iggy_binary_protocol::RequestFrame;
use iggy_binary_protocol::codes::{GET_CLUSTER_METADATA_CODE, SEND_MESSAGES_CODE, command_name};
-use iggy_common::{IggyError, SenderKind, TransportProtocol};
+use iggy_common::{IggyError, TransportProtocol};
use std::rc::Rc;
use tracing::{debug, error, info, trace, warn};
diff --git a/core/server/src/quic/quic_server.rs b/core/server/src/quic/quic_server.rs
index 51c946299..ca94b0676 100644
--- a/core/server/src/quic/quic_server.rs
+++ b/core/server/src/quic/quic_server.rs
@@ -185,7 +185,7 @@ fn configure_quic(config: &QuicConfig) -> Result<ServerConfig, QuicError> {
fn generate_self_signed_cert<'a>() -> Result<(Vec<CertificateDer<'a>>, PrivateKeyDer<'a>), QuicError>
{
- iggy_common::generate_self_signed_certificate("localhost").map_err(|e| {
+ server_common::generate_self_signed_certificate("localhost").map_err(|e| {
error!("{COMPONENT} (error: {e}) - failed to generate self-signed certificate");
QuicError::CertGenerationError
})
diff --git a/core/common/src/sender/mod.rs b/core/server/src/sender/mod.rs
similarity index 99%
rename from core/common/src/sender/mod.rs
rename to core/server/src/sender/mod.rs
index 1da8eb0de..292606747 100644
--- a/core/common/src/sender/mod.rs
+++ b/core/server/src/sender/mod.rs
@@ -28,14 +28,14 @@ pub use tcp_tls_sender::TcpTlsSender;
pub use websocket_sender::WebSocketSender;
pub use websocket_tls_sender::WebSocketTlsSender;
-use crate::IggyError;
-use crate::alloc::buffer::PooledBuffer;
use compio::BufResult;
use compio::buf::IoBufMut;
use compio::io::{AsyncReadExt, AsyncWriteExt};
use compio::net::TcpStream;
use compio::quic::{RecvStream, SendStream};
use compio::tls::TlsStream;
+use iggy_common::IggyError;
+use server_common::PooledBuffer;
use std::future::Future;
#[cfg(unix)]
use std::os::fd::{AsFd, OwnedFd};
diff --git a/core/common/src/sender/quic_sender.rs b/core/server/src/sender/quic_sender.rs
similarity index 98%
rename from core/common/src/sender/quic_sender.rs
rename to core/server/src/sender/quic_sender.rs
index a8b86c378..f0dcc610a 100644
--- a/core/common/src/sender/quic_sender.rs
+++ b/core/server/src/sender/quic_sender.rs
@@ -16,13 +16,14 @@
* under the License.
*/
-use super::{PooledBuffer, Sender};
-use crate::IggyError;
+use super::Sender;
use compio::BufResult;
use compio::buf::IoBufMut;
use compio::io::{AsyncReadExt, AsyncWriteExt};
use compio::quic::{ClosedStream, RecvStream, SendStream};
use err_trail::ErrContext;
+use iggy_common::IggyError;
+use server_common::PooledBuffer;
use tracing::{debug, error};
const COMPONENT: &str = "QUIC";
diff --git a/core/common/src/sender/tcp_sender.rs b/core/server/src/sender/tcp_sender.rs
similarity index 97%
rename from core/common/src/sender/tcp_sender.rs
rename to core/server/src/sender/tcp_sender.rs
index 2011650e1..45cc63614 100644
--- a/core/common/src/sender/tcp_sender.rs
+++ b/core/server/src/sender/tcp_sender.rs
@@ -16,12 +16,13 @@
* under the License.
*/
-use super::{PooledBuffer, Sender};
-use crate::IggyError;
+use super::Sender;
use compio::buf::IoBufMut;
use compio::io::AsyncWrite;
use compio::net::TcpStream;
use err_trail::ErrContext;
+use iggy_common::IggyError;
+use server_common::PooledBuffer;
const COMPONENT: &str = "TCP";
diff --git a/core/common/src/sender/tcp_tls_sender.rs b/core/server/src/sender/tcp_tls_sender.rs
similarity index 97%
rename from core/common/src/sender/tcp_tls_sender.rs
rename to core/server/src/sender/tcp_tls_sender.rs
index 81c7b287e..73b513f16 100644
--- a/core/common/src/sender/tcp_tls_sender.rs
+++ b/core/server/src/sender/tcp_tls_sender.rs
@@ -16,13 +16,14 @@
* under the License.
*/
-use super::{PooledBuffer, Sender};
-use crate::IggyError;
+use super::Sender;
use compio::buf::IoBufMut;
use compio::io::AsyncWrite;
use compio::net::TcpStream;
use compio::tls::TlsStream;
use err_trail::ErrContext;
+use iggy_common::IggyError;
+use server_common::PooledBuffer;
const COMPONENT: &str = "TCP";
diff --git a/core/common/src/sender/websocket_sender.rs b/core/server/src/sender/websocket_sender.rs
similarity index 99%
rename from core/common/src/sender/websocket_sender.rs
rename to core/server/src/sender/websocket_sender.rs
index 3af20df1b..b432158fc 100644
--- a/core/common/src/sender/websocket_sender.rs
+++ b/core/server/src/sender/websocket_sender.rs
@@ -17,13 +17,13 @@
*/
use super::Sender;
-use crate::IggyError;
-use crate::alloc::buffer::PooledBuffer;
use bytes::{BufMut, BytesMut};
use compio::buf::IoBufMut;
use compio::net::TcpStream;
use compio::ws::WebSocketStream;
use compio::ws::tungstenite::{Error as TungsteniteError, Message};
+use iggy_common::IggyError;
+use server_common::PooledBuffer;
use std::ptr;
use tracing::{debug, warn};
diff --git a/core/common/src/sender/websocket_tls_sender.rs b/core/server/src/sender/websocket_tls_sender.rs
similarity index 99%
rename from core/common/src/sender/websocket_tls_sender.rs
rename to core/server/src/sender/websocket_tls_sender.rs
index 0946a27c8..82f94b3ae 100644
--- a/core/common/src/sender/websocket_tls_sender.rs
+++ b/core/server/src/sender/websocket_tls_sender.rs
@@ -17,14 +17,14 @@
*/
use super::Sender;
-use crate::IggyError;
-use crate::alloc::buffer::PooledBuffer;
use bytes::{BufMut, BytesMut};
use compio::buf::IoBufMut;
use compio::net::TcpStream;
use compio::tls::TlsStream;
use compio::ws::WebSocketStream;
use compio::ws::tungstenite::{Error as TungsteniteError, Message};
+use iggy_common::IggyError;
+use server_common::PooledBuffer;
use std::ptr;
use tracing::debug;
diff --git a/core/server/src/shard/builder.rs b/core/server/src/shard/builder.rs
index a19643a82..95652377e 100644
--- a/core/server/src/shard/builder.rs
+++ b/core/server/src/shard/builder.rs
@@ -35,7 +35,7 @@ use ahash::AHashSet;
use dashmap::DashMap;
use iggy_common::EncryptorKind;
use iggy_common::SemanticVersion;
-use iggy_common::sharding::{IggyNamespace, PartitionLocation};
+use server_common::sharding::{IggyNamespace, PartitionLocation};
use std::{
cell::{Cell, RefCell},
rc::Rc,
diff --git a/core/server/src/shard/communication.rs b/core/server/src/shard/communication.rs
index de4df5794..a67014110 100644
--- a/core/server/src/shard/communication.rs
+++ b/core/server/src/shard/communication.rs
@@ -26,8 +26,8 @@ use crate::shard::{
};
use futures::future::join_all;
use hash32::{Hasher, Murmur3Hasher};
-use iggy_common::sharding::{IggyNamespace, PartitionLocation};
use iggy_common::{Identifier, IggyError};
+use server_common::sharding::{IggyNamespace, PartitionLocation};
use std::hash::Hasher as _;
use tracing::{error, info, warn};
diff --git a/core/server/src/shard/handlers.rs b/core/server/src/shard/handlers.rs
index 5d4de789c..44cc1c0cc 100644
--- a/core/server/src/shard/handlers.rs
+++ b/core/server/src/shard/handlers.rs
@@ -16,6 +16,7 @@
// under the License.
use super::*;
+use crate::sender::SenderKind;
use crate::{
shard::{
IggyShard, execution,
@@ -31,8 +32,9 @@ use crate::{
},
};
use compio::net::TcpStream;
-use iggy_common::{IggyError, SenderKind, TransportProtocol, sharding::IggyNamespace};
+use iggy_common::{IggyError, TransportProtocol};
use nix::sys::stat::SFlag;
+use server_common::sharding::IggyNamespace;
use std::os::fd::{FromRawFd, IntoRawFd};
use tracing::info;
diff --git a/core/server/src/shard/mod.rs b/core/server/src/shard/mod.rs
index 95f766a57..cf8149a37 100644
--- a/core/server/src/shard/mod.rs
+++ b/core/server/src/shard/mod.rs
@@ -36,8 +36,8 @@ use ahash::AHashSet;
use builder::IggyShardBuilder;
use dashmap::DashMap;
use iggy_common::SemanticVersion;
-use iggy_common::sharding::{IggyNamespace, PartitionLocation};
use iggy_common::{EncryptorKind, IggyByteSize, IggyError};
+use server_common::sharding::{IggyNamespace, PartitionLocation};
use std::{
cell::{Cell, RefCell},
net::SocketAddr,
diff --git a/core/server/src/shard/system/consumer_offsets.rs b/core/server/src/shard/system/consumer_offsets.rs
index ab287a834..5efb263c9 100644
--- a/core/server/src/shard/system/consumer_offsets.rs
+++ b/core/server/src/shard/system/consumer_offsets.rs
@@ -26,9 +26,8 @@ use crate::{
},
};
use err_trail::ErrContext;
-use iggy_common::{
- Consumer, ConsumerKind, ConsumerOffsetInfo, Identifier, IggyError, sharding::IggyNamespace,
-};
+use iggy_common::{Consumer, ConsumerKind, ConsumerOffsetInfo, Identifier, IggyError};
+use server_common::sharding::IggyNamespace;
use std::sync::atomic::Ordering;
impl IggyShard {
diff --git a/core/server/src/shard/system/messages.rs b/core/server/src/shard/system/messages.rs
index 82a5d25b0..2d4f368ee 100644
--- a/core/server/src/shard/system/messages.rs
+++ b/core/server/src/shard/system/messages.rs
@@ -27,11 +27,11 @@ use crate::streaming::polling_consumer::PollingConsumer;
use crate::streaming::segments::{IggyIndexesMut, IggyMessagesBatchMut, IggyMessagesBatchSet};
use err_trail::ErrContext;
use iggy_common::IggyPollMetadata;
-use iggy_common::PooledBuffer;
-use iggy_common::sharding::IggyNamespace;
use iggy_common::{
Consumer, EncryptorKind, IGGY_MESSAGE_HEADER_SIZE, Identifier, IggyError, PollingStrategy,
};
+use server_common::PooledBuffer;
+use server_common::sharding::IggyNamespace;
use std::sync::atomic::Ordering;
use tracing::error;
diff --git a/core/server/src/shard/system/partitions.rs b/core/server/src/shard/system/partitions.rs
index f0831828a..309b90c54 100644
--- a/core/server/src/shard/system/partitions.rs
+++ b/core/server/src/shard/system/partitions.rs
@@ -32,8 +32,8 @@ use crate::streaming::stats::PartitionStats;
use iggy_common::Identifier;
use iggy_common::IggyError;
use iggy_common::IggyTimestamp;
-use iggy_common::sharding::IggyNamespace;
-use iggy_common::sharding::{LocalIdx, PartitionLocation, ShardId};
+use server_common::sharding::IggyNamespace;
+use server_common::sharding::{LocalIdx, PartitionLocation, ShardId};
use std::sync::Arc;
use std::time::Duration;
use tracing::{info, warn};
diff --git a/core/server/src/shard/system/segments.rs b/core/server/src/shard/system/segments.rs
index 8eb7473e3..0d6546e15 100644
--- a/core/server/src/shard/system/segments.rs
+++ b/core/server/src/shard/system/segments.rs
@@ -18,8 +18,8 @@
use crate::configs::cache_indexes::CacheIndexesConfig;
use crate::shard::IggyShard;
use crate::streaming::segments::Segment;
-use iggy_common::sharding::IggyNamespace;
use iggy_common::{ConsumerKind, IggyError, IggyExpiry, IggyTimestamp, MaxTopicSize};
+use server_common::sharding::IggyNamespace;
impl IggyShard {
/// Performs all cleanup for a topic's partitions: time-based expiry then size-based trimming.
diff --git a/core/server/src/shard/system/storage.rs b/core/server/src/shard/system/storage.rs
index 3c05b1c29..5d86aec41 100644
--- a/core/server/src/shard/system/storage.rs
+++ b/core/server/src/shard/system/storage.rs
@@ -25,7 +25,7 @@ use compio::buf::IoBuf;
use compio::io::AsyncReadAtExt;
use err_trail::ErrContext;
use iggy_common::IggyError;
-use iggy_common::PooledBuffer;
+use server_common::PooledBuffer;
use std::sync::Arc;
use tracing::info;
diff --git a/core/server/src/shard/system/streams.rs b/core/server/src/shard/system/streams.rs
index 408e8d2d4..dc22869a1 100644
--- a/core/server/src/shard/system/streams.rs
+++ b/core/server/src/shard/system/streams.rs
@@ -20,8 +20,8 @@ use crate::metadata::StreamMeta;
use crate::shard::IggyShard;
use crate::shard::transmission::message::{ResolvedStream, ResolvedTopic};
use crate::streaming::streams::storage::{create_stream_file_hierarchy, delete_stream_directory};
-use iggy_common::sharding::IggyNamespace;
use iggy_common::{IggyError, IggyTimestamp};
+use server_common::sharding::IggyNamespace;
use std::sync::Arc;
/// Info returned when a stream is deleted - contains what callers need for logging/events.
diff --git a/core/server/src/shard/system/topics.rs b/core/server/src/shard/system/topics.rs
index c31da7f39..0a9012904 100644
--- a/core/server/src/shard/system/topics.rs
+++ b/core/server/src/shard/system/topics.rs
@@ -20,10 +20,10 @@ use crate::metadata::TopicMeta;
use crate::shard::IggyShard;
use crate::shard::transmission::message::{ResolvedStream, ResolvedTopic};
use crate::streaming::topics::storage::{create_topic_file_hierarchy, delete_topic_directory};
-use iggy_common::sharding::IggyNamespace;
use iggy_common::{
CompressionAlgorithm, Identifier, IggyError, IggyExpiry, IggyTimestamp, MaxTopicSize,
};
+use server_common::sharding::IggyNamespace;
use std::sync::Arc;
/// Info returned when a topic is deleted - contains what callers need for logging/events.
diff --git a/core/server/src/shard/tasks/periodic/message_cleaner.rs b/core/server/src/shard/tasks/periodic/message_cleaner.rs
index c0443da6e..c8d1b103c 100644
--- a/core/server/src/shard/tasks/periodic/message_cleaner.rs
+++ b/core/server/src/shard/tasks/periodic/message_cleaner.rs
@@ -20,7 +20,7 @@ use crate::shard::IggyShard;
use crate::shard::transmission::frame::ShardResponse;
use crate::shard::transmission::message::{ShardRequest, ShardRequestPayload};
use iggy_common::IggyError;
-use iggy_common::sharding::IggyNamespace;
+use server_common::sharding::IggyNamespace;
use std::rc::Rc;
use tracing::{error, info, trace};
diff --git a/core/server/src/shard/transmission/message.rs b/core/server/src/shard/transmission/message.rs
index 1238f1fed..eb07aecd2 100644
--- a/core/server/src/shard/transmission/message.rs
+++ b/core/server/src/shard/transmission/message.rs
@@ -22,7 +22,7 @@ use crate::{
use iggy_binary_protocol::requests::{
consumer_groups::*, partitions::*, personal_access_tokens::*, streams::*, topics::*, users::*,
};
-use iggy_common::sharding::IggyNamespace;
+use server_common::sharding::IggyNamespace;
use std::{net::SocketAddr, os::fd::OwnedFd};
diff --git a/core/server/src/streaming/deduplication/mod.rs b/core/server/src/streaming/deduplication/mod.rs
index fee04f314..762cd8a02 100644
--- a/core/server/src/streaming/deduplication/mod.rs
+++ b/core/server/src/streaming/deduplication/mod.rs
@@ -16,4 +16,4 @@
* under the License.
*/
-pub use iggy_common::MessageDeduplicator;
+pub use server_common::MessageDeduplicator;
diff --git a/core/server/src/streaming/partitions/in_flight.rs b/core/server/src/streaming/partitions/in_flight.rs
index 01f5b4f58..ca726df8f 100644
--- a/core/server/src/streaming/partitions/in_flight.rs
+++ b/core/server/src/streaming/partitions/in_flight.rs
@@ -15,4 +15,4 @@
// specific language governing permissions and limitations
// under the License.
-pub use iggy_common::IggyMessagesBatchSetInFlight;
+pub use server_common::IggyMessagesBatchSetInFlight;
diff --git a/core/server/src/streaming/partitions/local_partitions.rs b/core/server/src/streaming/partitions/local_partitions.rs
index 6d31a3230..1b3073fcb 100644
--- a/core/server/src/streaming/partitions/local_partitions.rs
+++ b/core/server/src/streaming/partitions/local_partitions.rs
@@ -20,7 +20,7 @@
//! Single-threaded (compio runtime) - NO synchronization needed!
use super::local_partition::LocalPartition;
-use iggy_common::sharding::IggyNamespace;
+use server_common::sharding::IggyNamespace;
use std::collections::HashMap;
/// Per-shard partition storage.
diff --git a/core/server/src/streaming/partitions/ops.rs b/core/server/src/streaming/partitions/ops.rs
index 489e35b8a..8b9ba7205 100644
--- a/core/server/src/streaming/partitions/ops.rs
+++ b/core/server/src/streaming/partitions/ops.rs
@@ -41,8 +41,8 @@ use crate::shard::system::messages::PollingArgs;
use crate::streaming::polling_consumer::PollingConsumer;
use crate::streaming::segments::IggyMessagesBatchSet;
use iggy_common::IggyPollMetadata;
-use iggy_common::sharding::IggyNamespace;
use iggy_common::{IggyError, PollingKind};
+use server_common::sharding::IggyNamespace;
use std::cell::RefCell;
use std::sync::atomic::Ordering;
diff --git a/core/server/src/streaming/partitions/ops_tests.rs b/core/server/src/streaming/partitions/ops_tests.rs
index fa290acbf..7769666f4 100644
--- a/core/server/src/streaming/partitions/ops_tests.rs
+++ b/core/server/src/streaming/partitions/ops_tests.rs
@@ -36,10 +36,9 @@ mod tests {
use crate::streaming::partitions::ops;
use crate::streaming::polling_consumer::PollingConsumer;
use crate::streaming::stats::{PartitionStats, StreamStats, TopicStats};
- use iggy_common::sharding::IggyNamespace;
- use iggy_common::{
- IggyByteSize, IggyMessage, MemoryPool, MemoryPoolConfigOther, PollingStrategy, Sizeable,
- };
+ use iggy_common::{IggyByteSize, IggyMessage, PollingStrategy, Sizeable};
+ use server_common::sharding::IggyNamespace;
+ use server_common::{MemoryPool, MemoryPoolConfigOther};
use std::cell::RefCell;
use std::sync::Arc;
use std::sync::atomic::AtomicU64;
@@ -73,7 +72,7 @@ mod tests {
)
}
- fn create_batch(count: u32) -> iggy_common::IggyMessagesBatchMut {
+ fn create_batch(count: u32) -> server_common::IggyMessagesBatchMut {
let messages: Vec<IggyMessage> = (0..count)
.map(|_| {
IggyMessage::builder()
@@ -87,7 +86,7 @@ mod tests {
.iter()
.map(|m| m.get_size_bytes().as_bytes_u32())
.sum();
- iggy_common::IggyMessagesBatchMut::from_messages(&messages, messages_size)
+ server_common::IggyMessagesBatchMut::from_messages(&messages, messages_size)
}
/// Sets up "State C": in-flight holds committed data, journal holds new
@@ -112,7 +111,7 @@ mod tests {
let mut partition = create_test_partition(journal_end);
let segment = iggy_common::Segment::new(0, IggyByteSize::from(1_073_741_824u64));
- let storage = iggy_common::SegmentStorage::default();
+ let storage = server_common::SegmentStorage::default();
partition.log.add_persisted_segment(segment, storage);
let seg = &mut partition.log.segments_mut()[0];
@@ -216,7 +215,7 @@ mod tests {
let segment = iggy_common::Segment::new(0, IggyByteSize::from(1_073_741_824u64));
partition
.log
- .add_persisted_segment(segment, iggy_common::SegmentStorage::default());
+ .add_persisted_segment(segment, server_common::SegmentStorage::default());
let seg = &mut partition.log.segments_mut()[0];
seg.end_offset = 9;
seg.start_timestamp = 1;
@@ -245,7 +244,7 @@ mod tests {
let segment = iggy_common::Segment::new(0, IggyByteSize::from(1_073_741_824u64));
partition
.log
- .add_persisted_segment(segment, iggy_common::SegmentStorage::default());
+ .add_persisted_segment(segment, server_common::SegmentStorage::default());
let seg = &mut partition.log.segments_mut()[0];
seg.end_offset = 9;
seg.start_timestamp = 1;
diff --git a/core/server/src/streaming/segments/indexes/index_reader.rs b/core/server/src/streaming/segments/indexes/index_reader.rs
index 2ad9f0ed1..3decdb46e 100644
--- a/core/server/src/streaming/segments/indexes/index_reader.rs
+++ b/core/server/src/streaming/segments/indexes/index_reader.rs
@@ -16,4 +16,4 @@
// under the License.
#[allow(unused_imports)]
-pub use iggy_common::IndexReader;
+pub use server_common::IndexReader;
diff --git a/core/server/src/streaming/segments/indexes/index_writer.rs b/core/server/src/streaming/segments/indexes/index_writer.rs
index e1ca3b94b..c77bada32 100644
--- a/core/server/src/streaming/segments/indexes/index_writer.rs
+++ b/core/server/src/streaming/segments/indexes/index_writer.rs
@@ -15,4 +15,4 @@
// specific language governing permissions and limitations
// under the License.
-pub use iggy_common::IndexWriter;
+pub use server_common::IndexWriter;
diff --git a/core/server/src/streaming/segments/indexes/mod.rs b/core/server/src/streaming/segments/indexes/mod.rs
index 1103b7fd9..3b31e8a0f 100644
--- a/core/server/src/streaming/segments/indexes/mod.rs
+++ b/core/server/src/streaming/segments/indexes/mod.rs
@@ -19,5 +19,5 @@
mod index_reader;
mod index_writer;
-pub use iggy_common::IggyIndexesMut;
pub use index_writer::IndexWriter;
+pub use server_common::IggyIndexesMut;
diff --git a/core/server/src/streaming/segments/messages/messages_reader.rs b/core/server/src/streaming/segments/messages/messages_reader.rs
index cb45a7750..3398351b8 100644
--- a/core/server/src/streaming/segments/messages/messages_reader.rs
+++ b/core/server/src/streaming/segments/messages/messages_reader.rs
@@ -16,4 +16,4 @@
// under the License.
#[allow(unused_imports)]
-pub use iggy_common::MessagesReader;
+pub use server_common::MessagesReader;
diff --git a/core/server/src/streaming/segments/messages/messages_writer.rs b/core/server/src/streaming/segments/messages/messages_writer.rs
index 3639474fa..b9911b201 100644
--- a/core/server/src/streaming/segments/messages/messages_writer.rs
+++ b/core/server/src/streaming/segments/messages/messages_writer.rs
@@ -15,4 +15,4 @@
// specific language governing permissions and limitations
// under the License.
-pub use iggy_common::MessagesWriter;
+pub use server_common::MessagesWriter;
diff --git a/core/server/src/streaming/segments/storage.rs b/core/server/src/streaming/segments/storage.rs
index 7e62daecf..8d0212ce3 100644
--- a/core/server/src/streaming/segments/storage.rs
+++ b/core/server/src/streaming/segments/storage.rs
@@ -15,7 +15,7 @@
// specific language governing permissions and limitations
// under the License.
-pub use iggy_common::SegmentStorage as Storage;
+pub use server_common::SegmentStorage as Storage;
use crate::configs::system::SystemConfig;
use iggy_common::IggyError;
diff --git a/core/server/src/streaming/segments/types/mod.rs b/core/server/src/streaming/segments/types/mod.rs
index fe00aca5f..cec3328f9 100644
--- a/core/server/src/streaming/segments/types/mod.rs
+++ b/core/server/src/streaming/segments/types/mod.rs
@@ -16,6 +16,5 @@
* under the License.
*/
-pub use iggy_common::{
- IggyMessageHeaderViewMut, IggyMessageViewMut, IggyMessagesBatchMut, IggyMessagesBatchSet,
-};
+pub use iggy_common::{IggyMessageHeaderViewMut, IggyMessageViewMut};
+pub use server_common::{IggyMessagesBatchMut, IggyMessagesBatchSet};
diff --git a/core/server/src/tcp/connection_handler.rs b/core/server/src/tcp/connection_handler.rs
index 3225ea273..cbd060a87 100644
--- a/core/server/src/tcp/connection_handler.rs
+++ b/core/server/src/tcp/connection_handler.rs
@@ -17,6 +17,7 @@
*/
use crate::binary::dispatch::{self, HandlerResult, MAX_CONTROL_FRAME_PAYLOAD};
+use crate::sender::SenderKind;
use crate::server_error::ConnectionError;
use crate::shard::IggyShard;
use crate::streaming::session::Session;
@@ -25,7 +26,7 @@ use bytes::BytesMut;
use futures::FutureExt;
use iggy_binary_protocol::RequestFrame;
use iggy_binary_protocol::codes::{GET_CLUSTER_METADATA_CODE, SEND_MESSAGES_CODE, command_name};
-use iggy_common::{IggyError, SenderKind};
+use iggy_common::IggyError;
use std::io::ErrorKind;
use std::rc::Rc;
use tracing::{debug, error, info};
diff --git a/core/server/src/tcp/tcp_listener.rs b/core/server/src/tcp/tcp_listener.rs
index 9a514aa20..5c70c920f 100644
--- a/core/server/src/tcp/tcp_listener.rs
+++ b/core/server/src/tcp/tcp_listener.rs
@@ -18,6 +18,7 @@
use crate::configs::tcp::TcpSocketConfig;
+use crate::sender::SenderKind;
use crate::shard::IggyShard;
use crate::shard::task_registry::{ShutdownToken, TaskRegistry};
use crate::shard::transmission::event::ShardEvent;
@@ -25,7 +26,7 @@ use crate::tcp::connection_handler::{ConnectionAction, handle_connection, handle
use compio::net::{SocketOpts, TcpListener};
use err_trail::ErrContext;
use futures::FutureExt;
-use iggy_common::{IggyError, SenderKind, TransportProtocol};
+use iggy_common::{IggyError, TransportProtocol};
use std::net::SocketAddr;
use std::rc::Rc;
use std::time::Duration;
diff --git a/core/server/src/tcp/tcp_tls_listener.rs b/core/server/src/tcp/tcp_tls_listener.rs
index c0230774e..41781bdbf 100644
--- a/core/server/src/tcp/tcp_tls_listener.rs
+++ b/core/server/src/tcp/tcp_tls_listener.rs
@@ -17,6 +17,7 @@
*/
use crate::configs::tcp::TcpSocketConfig;
+use crate::sender::SenderKind;
use crate::shard::IggyShard;
use crate::shard::task_registry::ShutdownToken;
use crate::shard::transmission::event::ShardEvent;
@@ -25,7 +26,7 @@ use compio::net::{SocketOpts, TcpListener};
use compio::tls::TlsAcceptor;
use err_trail::ErrContext;
use futures::FutureExt;
-use iggy_common::{IggyError, SenderKind, TransportProtocol};
+use iggy_common::{IggyError, TransportProtocol};
use rustls::ServerConfig;
use rustls::pki_types::{CertificateDer, PrivateKeyDer};
use rustls_pemfile::{certs, private_key};
@@ -225,7 +226,7 @@ async fn accept_loop(
fn generate_self_signed_cert()
-> Result<(Vec<CertificateDer<'static>>, PrivateKeyDer<'static>), Box<dyn std::error::Error>> {
- iggy_common::generate_self_signed_certificate("localhost")
+ server_common::generate_self_signed_certificate("localhost")
}
fn load_certificates(
diff --git a/core/server/src/websocket/connection_handler.rs b/core/server/src/websocket/connection_handler.rs
index e68bcc902..1cdb8a271 100644
--- a/core/server/src/websocket/connection_handler.rs
+++ b/core/server/src/websocket/connection_handler.rs
@@ -17,6 +17,7 @@
*/
use crate::binary::dispatch::{self, HandlerResult, MAX_CONTROL_FRAME_PAYLOAD};
+use crate::sender::SenderKind;
use crate::server_error::ConnectionError;
use crate::shard::IggyShard;
use crate::streaming::session::Session;
@@ -25,7 +26,7 @@ use bytes::BytesMut;
use futures::FutureExt;
use iggy_binary_protocol::RequestFrame;
use iggy_binary_protocol::codes::{SEND_MESSAGES_CODE, command_name};
-use iggy_common::{IggyError, SenderKind};
+use iggy_common::IggyError;
use std::io::ErrorKind;
use std::rc::Rc;
use tracing::{debug, error, info, warn};
diff --git a/core/server/src/websocket/websocket_listener.rs b/core/server/src/websocket/websocket_listener.rs
index 56a6225c8..56e6b9aac 100644
--- a/core/server/src/websocket/websocket_listener.rs
+++ b/core/server/src/websocket/websocket_listener.rs
@@ -17,6 +17,7 @@
*/
use crate::configs::websocket::WebSocketConfig;
+use crate::sender::{SenderKind, WebSocketSender};
use crate::shard::IggyShard;
use crate::shard::task_registry::ShutdownToken;
use crate::shard::transmission::event::ShardEvent;
@@ -25,8 +26,8 @@ use compio::net::{SocketOpts, TcpListener};
use compio::ws::accept_async_with_config;
use err_trail::ErrContext;
use futures::FutureExt;
+use iggy_common::IggyError;
use iggy_common::TransportProtocol;
-use iggy_common::{IggyError, SenderKind, WebSocketSender};
use std::net::SocketAddr;
use std::rc::Rc;
use tracing::{debug, error, info};
diff --git a/core/server/src/websocket/websocket_tls_listener.rs b/core/server/src/websocket/websocket_tls_listener.rs
index bc75d9e50..4c17dc237 100644
--- a/core/server/src/websocket/websocket_tls_listener.rs
+++ b/core/server/src/websocket/websocket_tls_listener.rs
@@ -17,6 +17,7 @@
*/
use crate::configs::websocket::WebSocketConfig;
+use crate::sender::{SenderKind, WebSocketTlsSender};
use crate::shard::IggyShard;
use crate::shard::task_registry::ShutdownToken;
use crate::shard::transmission::event::ShardEvent;
@@ -26,7 +27,7 @@ use compio::tls::TlsAcceptor;
use compio::ws::accept_async_with_config;
use err_trail::ErrContext;
use futures::FutureExt;
-use iggy_common::{IggyError, SenderKind, TransportProtocol, WebSocketTlsSender};
+use iggy_common::{IggyError, TransportProtocol};
use rustls::ServerConfig;
use rustls::pki_types::{CertificateDer, PrivateKeyDer};
use rustls_pemfile::{certs, private_key};
@@ -247,7 +248,7 @@ async fn accept_loop(
fn generate_self_signed_cert()
-> Result<(Vec<CertificateDer<'static>>, PrivateKeyDer<'static>), Box<dyn std::error::Error>> {
- iggy_common::generate_self_signed_certificate("localhost")
+ server_common::generate_self_signed_certificate("localhost")
}
fn load_certificates(
diff --git a/core/partitions/Cargo.toml b/core/server_common/Cargo.toml
similarity index 64%
copy from core/partitions/Cargo.toml
copy to core/server_common/Cargo.toml
index 158790862..706089b6f 100644
--- a/core/partitions/Cargo.toml
+++ b/core/server_common/Cargo.toml
@@ -14,35 +14,38 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
-
[package]
-name = "partitions"
+name = "server_common"
version = "0.1.0"
-description = "Iggy partitions abstraction for clustering support"
+description = "Server-only primitives shared between iggy server crates. Not part of the public SDK surface."
edition = "2024"
license = "Apache-2.0"
-keywords = ["iggy", "messaging", "streaming", "partitions"]
-homepage = "https://iggy.apache.org"
-documentation = "https://iggy.apache.org/docs"
-repository = "https://github.com/apache/iggy"
-readme = "../../README.md"
publish = false
[dependencies]
+aligned-vec = { workspace = true }
bytemuck = { workspace = true }
bytes = { workspace = true }
compio = { workspace = true }
-consensus = { workspace = true }
+compio-buf = { workspace = true }
+crossbeam = { workspace = true }
+err_trail = { workspace = true }
+human-repr = { workspace = true }
iggy_binary_protocol = { workspace = true }
iggy_common = { workspace = true }
-journal = { workspace = true }
-message_bus = { workspace = true }
-ringbuffer = { workspace = true }
+lending-iterator = { workspace = true }
+moka = { workspace = true }
+once_cell = { workspace = true }
+rcgen = { workspace = true }
+rustls = { workspace = true }
smallvec = { workspace = true }
-tokio = { workspace = true }
+thiserror = { workspace = true }
tracing = { workspace = true }
+twox-hash = { workspace = true }
-[lints.clippy]
-enum_glob_use = "deny"
-pedantic = "deny"
-nursery = "warn"
+[target.'cfg(unix)'.dependencies]
+nix = { workspace = true }
+
+[dev-dependencies]
+serial_test = { workspace = true }
+tokio = { workspace = true }
diff --git a/core/common/src/alloc/buffer.rs b/core/server_common/src/buffer.rs
similarity index 98%
rename from core/common/src/alloc/buffer.rs
rename to core/server_common/src/buffer.rs
index 0d81b9cc9..5d8366cf1 100644
--- a/core/common/src/alloc/buffer.rs
+++ b/core/server_common/src/buffer.rs
@@ -16,9 +16,7 @@
* under the License.
*/
-use crate::alloc::memory_pool::{ALIGNMENT, AlignedBuffer};
-
-use super::memory_pool::{AlignedBufferExt, memory_pool};
+use crate::memory_pool::{ALIGNMENT, AlignedBuffer, AlignedBufferExt, memory_pool};
use bytes::Bytes;
use compio::buf::{IoBuf, IoBufMut, SetLen};
use std::{
@@ -437,8 +435,8 @@ mod miri_tests {
#[cfg(not(miri))]
mod split_to {
use super::*;
- use crate::IggyByteSize;
- use crate::alloc::memory_pool::{MemoryPool, MemoryPoolConfigOther};
+ use crate::memory_pool::{MemoryPool, MemoryPoolConfigOther};
+ use iggy_common::IggyByteSize;
use serial_test::serial;
use std::str::FromStr;
use std::sync::Once;
diff --git a/core/common/src/certificates.rs b/core/server_common/src/certificates.rs
similarity index 100%
rename from core/common/src/certificates.rs
rename to core/server_common/src/certificates.rs
diff --git a/core/binary_protocol/src/consensus/message.rs b/core/server_common/src/consensus_message.rs
similarity index 98%
rename from core/binary_protocol/src/consensus/message.rs
rename to core/server_common/src/consensus_message.rs
index 76bed247b..b086d1bc3 100644
--- a/core/binary_protocol/src/consensus/message.rs
+++ b/core/server_common/src/consensus_message.rs
@@ -15,10 +15,10 @@
// specific language governing permissions and limitations
// under the License.
-use super::iobuf::{Frozen, Owned};
-use crate::consensus::{
- self, Command2, CommitHeader, ConsensusError, ConsensusHeader, DoViewChangeHeader,
- GenericHeader, PrepareHeader, PrepareOkHeader, RequestHeader, StartViewChangeHeader,
+use crate::iobuf::{Frozen, Owned};
+use iggy_binary_protocol::{
+ Command2, CommitHeader, ConsensusError, ConsensusHeader, DoViewChangeHeader, GenericHeader,
+ Operation, PrepareHeader, PrepareOkHeader, RequestHeader, StartViewChangeHeader,
StartViewHeader,
};
use smallvec::SmallVec;
@@ -510,7 +510,7 @@ impl MessageBag {
}
#[must_use]
- pub fn operation(&self) -> consensus::Operation {
+ pub fn operation(&self) -> Operation {
match self {
Self::Request(message) => message.header().operation,
Self::Prepare(message) => message.header().operation,
@@ -564,7 +564,7 @@ where
#[cfg(test)]
mod tests {
use super::*;
- use crate::consensus::{Operation, ReplyHeader};
+ use iggy_binary_protocol::{Operation, ReplyHeader};
use smallvec::smallvec;
// Field offsets via `offset_of!`: a field reorder fails to compile here
diff --git a/core/common/src/deduplication/message_deduplicator.rs b/core/server_common/src/deduplication/message_deduplicator.rs
similarity index 99%
rename from core/common/src/deduplication/message_deduplicator.rs
rename to core/server_common/src/deduplication/message_deduplicator.rs
index 3a98ad925..9ce77e02c 100644
--- a/core/common/src/deduplication/message_deduplicator.rs
+++ b/core/server_common/src/deduplication/message_deduplicator.rs
@@ -16,7 +16,7 @@
* under the License.
*/
-use crate::IggyDuration;
+use iggy_common::IggyDuration;
use moka::future::{Cache, CacheBuilder};
#[derive(Debug)]
diff --git a/core/common/src/deduplication/mod.rs b/core/server_common/src/deduplication/mod.rs
similarity index 100%
rename from core/common/src/deduplication/mod.rs
rename to core/server_common/src/deduplication/mod.rs
diff --git a/core/common/src/types/message/in_flight.rs b/core/server_common/src/in_flight.rs
similarity index 98%
rename from core/common/src/types/message/in_flight.rs
rename to core/server_common/src/in_flight.rs
index 47654520d..f3eec7995 100644
--- a/core/common/src/types/message/in_flight.rs
+++ b/core/server_common/src/in_flight.rs
@@ -16,7 +16,7 @@
* under the License.
*/
-use crate::IggyMessagesBatch;
+use iggy_common::IggyMessagesBatch;
/// Holds batches that are being written to disk.
///
diff --git a/core/common/src/types/message/indexes_mut.rs b/core/server_common/src/indexes_mut.rs
similarity index 99%
rename from core/common/src/types/message/indexes_mut.rs
rename to core/server_common/src/indexes_mut.rs
index 51673bd52..240b96646 100644
--- a/core/common/src/types/message/indexes_mut.rs
+++ b/core/server_common/src/indexes_mut.rs
@@ -17,7 +17,7 @@
*/
use crate::PooledBuffer;
-use crate::{INDEX_SIZE, IggyIndexView, IggyIndexes};
+use iggy_common::{INDEX_SIZE, IggyIndexView, IggyIndexes};
use std::fmt;
use std::ops::{Deref, Index as StdIndex};
diff --git a/core/binary_protocol/src/consensus/iobuf.rs b/core/server_common/src/iobuf.rs
similarity index 100%
rename from core/binary_protocol/src/consensus/iobuf.rs
rename to core/server_common/src/iobuf.rs
diff --git a/core/server_common/src/lib.rs b/core/server_common/src/lib.rs
new file mode 100644
index 000000000..b11447e59
--- /dev/null
+++ b/core/server_common/src/lib.rs
@@ -0,0 +1,60 @@
+/* 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.
+ */
+
+mod buffer;
+mod certificates;
+mod consensus_message;
+mod deduplication;
+mod in_flight;
+mod indexes_mut;
+// TODO(hubcio): iobuf was relocated verbatim from `core/binary_protocol/src/consensus/iobuf.rs`
+// during the sans-io split. Its implementation is intentionally untouched; the
+// lints below predate the move and are tracked as tech debt for a 0.9.x cleanup pass.
+#[allow(
+ private_interfaces,
+ clippy::missing_errors_doc,
+ clippy::missing_panics_doc,
+ clippy::must_use_candidate,
+ clippy::return_self_not_must_use,
+ clippy::missing_const_for_fn,
+ clippy::non_send_fields_in_send_ty,
+ clippy::use_self
+)]
+pub mod iobuf;
+mod memory_pool;
+mod messages_batch_mut;
+mod messages_batch_set;
+mod segment_storage;
+pub mod send_messages2;
+pub mod sharding;
+
+pub use buffer::PooledBuffer;
+pub use certificates::generate_self_signed_certificate;
+pub use consensus_message::{
+ ConsensusMessage, FragmentedBacking, MESSAGE_ALIGN, Message, MessageBacking, MessageBag,
+ MutableBacking, RequestBacking, RequestBackingKind, ResponseBacking, ResponseBackingKind,
+};
+pub use deduplication::MessageDeduplicator;
+pub use in_flight::IggyMessagesBatchSetInFlight;
+pub use indexes_mut::IggyIndexesMut;
+pub use memory_pool::{MEMORY_POOL, MemoryPool, MemoryPoolConfigOther, memory_pool};
+pub use messages_batch_mut::IggyMessagesBatchMut;
+pub use messages_batch_set::IggyMessagesBatchSet;
+pub use segment_storage::{
+ IndexReader, IndexWriter, MessagesReader, MessagesWriter, SegmentStorage,
+};
diff --git a/core/common/src/alloc/memory_pool.rs b/core/server_common/src/memory_pool.rs
similarity index 99%
rename from core/common/src/alloc/memory_pool.rs
rename to core/server_common/src/memory_pool.rs
index 0dd2782ec..e50dcba68 100644
--- a/core/common/src/alloc/memory_pool.rs
+++ b/core/server_common/src/memory_pool.rs
@@ -81,7 +81,7 @@ pub struct MemoryPoolConfigOther {
/// Whether the pool is enabled.
pub enabled: bool,
/// Maximum size of the pool.
- pub size: crate::IggyByteSize,
+ pub size: iggy_common::IggyByteSize,
/// Maximum number of buffers per bucket.
pub bucket_capacity: u32,
}
@@ -493,7 +493,8 @@ fn size_str(size: usize) -> String {
#[cfg(test)]
mod tests {
- use crate::{IggyByteSize, alloc::buffer::PooledBuffer};
+ use crate::buffer::PooledBuffer;
+ use iggy_common::IggyByteSize;
use super::*;
use serial_test::serial;
diff --git a/core/common/src/types/message/messages_batch_mut.rs b/core/server_common/src/messages_batch_mut.rs
similarity index 98%
rename from core/common/src/types/message/messages_batch_mut.rs
rename to core/server_common/src/messages_batch_mut.rs
index 15174376a..8dc3b70bc 100644
--- a/core/common/src/types/message/messages_batch_mut.rs
+++ b/core/server_common/src/messages_batch_mut.rs
@@ -16,15 +16,15 @@
* under the License.
*/
-use super::indexes_mut::IggyIndexesMut;
-use super::message_boundaries::IggyMessageBoundaries;
-use super::message_view_mut::IggyMessageViewMutIterator;
-use crate::{
+use crate::MessageDeduplicator;
+use crate::PooledBuffer;
+use crate::indexes_mut::IggyIndexesMut;
+use iggy_common::{
IGGY_MESSAGE_HEADER_SIZE, INDEX_SIZE, IggyByteSize, IggyError, IggyIndexView, IggyMessage,
- IggyMessageView, IggyMessageViewIterator, IggyMessagesBatch, IggyTimestamp, MAX_PAYLOAD_SIZE,
- MAX_USER_HEADERS_SIZE, Sizeable, Validatable,
+ IggyMessageBoundaries, IggyMessageView, IggyMessageViewIterator, IggyMessageViewMutIterator,
+ IggyMessagesBatch, IggyTimestamp, MAX_PAYLOAD_SIZE, MAX_USER_HEADERS_SIZE, Sizeable,
+ Validatable, random_id,
};
-use crate::{MessageDeduplicator, PooledBuffer, random_id};
use lending_iterator::prelude::*;
use std::ops::Index;
use std::sync::Arc;
diff --git a/core/common/src/types/message/messages_batch_set.rs b/core/server_common/src/messages_batch_set.rs
similarity index 98%
rename from core/common/src/types/message/messages_batch_set.rs
rename to core/server_common/src/messages_batch_set.rs
index 6751b3580..4c488a17c 100644
--- a/core/common/src/types/message/messages_batch_set.rs
+++ b/core/server_common/src/messages_batch_set.rs
@@ -16,11 +16,12 @@
* under the License.
*/
-use crate::{
- IggyByteSize, IggyIndexesMut, IggyMessage, IggyMessageView, IggyMessagesBatch,
- IggyMessagesBatchMut, IggyPollMetadata, PolledMessages, PooledBuffer, Sizeable,
-};
+use crate::{IggyIndexesMut, IggyMessagesBatchMut, PooledBuffer};
use bytes::Bytes;
+use iggy_common::{
+ IggyByteSize, IggyMessage, IggyMessageView, IggyMessagesBatch, IggyPollMetadata,
+ PolledMessages, Sizeable,
+};
use std::ops::Index;
use tracing::trace;
diff --git a/core/common/src/types/segment_storage/index_reader.rs b/core/server_common/src/segment_storage/index_reader.rs
similarity index 99%
rename from core/common/src/types/segment_storage/index_reader.rs
rename to core/server_common/src/segment_storage/index_reader.rs
index e62233ca6..7f45c840b 100644
--- a/core/common/src/types/segment_storage/index_reader.rs
+++ b/core/server_common/src/segment_storage/index_reader.rs
@@ -15,13 +15,14 @@
// specific language governing permissions and limitations
// under the License.
-use crate::{INDEX_SIZE, IggyError, IggyIndex, IggyIndexView, IggyIndexesMut, PooledBuffer};
+use crate::{IggyIndexesMut, PooledBuffer};
use compio::{
buf::{IntoInner, IoBuf},
fs::{File, OpenOptions},
io::AsyncReadAtExt,
};
use err_trail::ErrContext;
+use iggy_common::{INDEX_SIZE, IggyError, IggyIndex, IggyIndexView};
use std::{
io::ErrorKind,
rc::Rc,
diff --git a/core/common/src/types/segment_storage/index_writer.rs b/core/server_common/src/segment_storage/index_writer.rs
similarity index 98%
rename from core/common/src/types/segment_storage/index_writer.rs
rename to core/server_common/src/segment_storage/index_writer.rs
index e1bee3ef2..c8a1c06a6 100644
--- a/core/common/src/types/segment_storage/index_writer.rs
+++ b/core/server_common/src/segment_storage/index_writer.rs
@@ -15,18 +15,17 @@
// specific language governing permissions and limitations
// under the License.
-use crate::INDEX_SIZE;
-use crate::IggyError;
+use crate::PooledBuffer;
use compio::fs::File;
use compio::fs::OpenOptions;
use compio::io::AsyncWriteAtExt;
use err_trail::ErrContext;
+use iggy_common::INDEX_SIZE;
+use iggy_common::IggyError;
use std::rc::Rc;
use std::sync::atomic::{AtomicU64, Ordering};
use tracing::trace;
-use crate::PooledBuffer;
-
/// A dedicated struct for writing to the index file.
#[derive(Debug)]
pub struct IndexWriter {
diff --git a/core/common/src/types/segment_storage/messages_reader.rs b/core/server_common/src/segment_storage/messages_reader.rs
similarity index 98%
rename from core/common/src/types/segment_storage/messages_reader.rs
rename to core/server_common/src/segment_storage/messages_reader.rs
index 824261b0a..ecfe7e255 100644
--- a/core/common/src/types/segment_storage/messages_reader.rs
+++ b/core/server_common/src/segment_storage/messages_reader.rs
@@ -15,11 +15,12 @@
// specific language governing permissions and limitations
// under the License.
-use crate::{IggyError, IggyIndexesMut, IggyMessagesBatchMut, PooledBuffer};
+use crate::{IggyIndexesMut, IggyMessagesBatchMut, PooledBuffer};
use compio::buf::{IntoInner, IoBuf};
use compio::fs::{File, OpenOptions};
use compio::io::AsyncReadAtExt;
use err_trail::ErrContext;
+use iggy_common::IggyError;
use std::rc::Rc;
use std::{
io::ErrorKind,
diff --git a/core/common/src/types/segment_storage/messages_writer.rs b/core/server_common/src/segment_storage/messages_writer.rs
similarity index 98%
rename from core/common/src/types/segment_storage/messages_writer.rs
rename to core/server_common/src/segment_storage/messages_writer.rs
index 051fc3f1b..3749d99d4 100644
--- a/core/common/src/types/segment_storage/messages_writer.rs
+++ b/core/server_common/src/segment_storage/messages_writer.rs
@@ -16,9 +16,9 @@
// under the License.
use super::write_batch_frozen;
-use crate::{IggyByteSize, IggyError, IggyMessagesBatch};
use compio::fs::{File, OpenOptions};
use err_trail::ErrContext;
+use iggy_common::{IggyByteSize, IggyError, IggyMessagesBatch};
use std::{
rc::Rc,
sync::atomic::{AtomicU64, Ordering},
diff --git a/core/common/src/types/segment_storage/mod.rs b/core/server_common/src/segment_storage/mod.rs
similarity index 98%
rename from core/common/src/types/segment_storage/mod.rs
rename to core/server_common/src/segment_storage/mod.rs
index a9c12b12b..953b35908 100644
--- a/core/common/src/types/segment_storage/mod.rs
+++ b/core/server_common/src/segment_storage/mod.rs
@@ -20,9 +20,9 @@ mod index_writer;
mod messages_reader;
mod messages_writer;
-use crate::{IggyError, IggyMessagesBatch};
use bytes::Bytes;
use compio::{fs::File, io::AsyncWriteAtExt};
+use iggy_common::{IggyError, IggyMessagesBatch};
use std::rc::Rc;
use tracing::error;
diff --git a/core/common/src/types/send_messages2.rs b/core/server_common/src/send_messages2.rs
similarity index 98%
rename from core/common/src/types/send_messages2.rs
rename to core/server_common/src/send_messages2.rs
index 73d07126f..1508b0254 100644
--- a/core/common/src/types/send_messages2.rs
+++ b/core/server_common/src/send_messages2.rs
@@ -15,14 +15,15 @@
// specific language governing permissions and limitations
// under the License.
-use crate::{INDEX_SIZE, IggyError, random_id, sharding::IggyNamespace};
+use crate::consensus_message::{MESSAGE_ALIGN, Message};
+use crate::iobuf::Owned;
+use crate::sharding::IggyNamespace;
use bytes::{Bytes, BytesMut};
-use iggy_binary_protocol::consensus::iobuf::Owned;
-use iggy_binary_protocol::{Message, PrepareHeader, RequestHeader};
+use iggy_binary_protocol::{PrepareHeader, RequestHeader};
+use iggy_common::{INDEX_SIZE, IggyError, random_id};
use std::hash::Hasher;
use twox_hash::XxHash3_64;
-const MESSAGE_ALIGN: usize = 4096;
pub const COMMAND_HEADER_SIZE: usize = 256;
pub const PREPARE_SPLIT_POINT: usize = 512;
const MESSAGE_HEADER_SIZE: usize = 48;
@@ -544,7 +545,7 @@ impl<'a> Iterator for SendMessages2IteratorWithOffsets<'a> {
}
}
-pub(crate) type FrozenBatchHeader = iggy_binary_protocol::consensus::iobuf::Frozen<MESSAGE_ALIGN>;
+pub(crate) type FrozenBatchHeader = crate::iobuf::Frozen<MESSAGE_ALIGN>;
pub fn convert_request_message(
namespace: IggyNamespace,
diff --git a/core/common/src/sharding/local_idx.rs b/core/server_common/src/sharding/local_idx.rs
similarity index 100%
rename from core/common/src/sharding/local_idx.rs
rename to core/server_common/src/sharding/local_idx.rs
diff --git a/core/common/src/sharding/mod.rs b/core/server_common/src/sharding/mod.rs
similarity index 100%
rename from core/common/src/sharding/mod.rs
rename to core/server_common/src/sharding/mod.rs
diff --git a/core/common/src/sharding/namespace.rs b/core/server_common/src/sharding/namespace.rs
similarity index 100%
rename from core/common/src/sharding/namespace.rs
rename to core/server_common/src/sharding/namespace.rs
diff --git a/core/common/src/sharding/partition_location.rs b/core/server_common/src/sharding/partition_location.rs
similarity index 100%
rename from core/common/src/sharding/partition_location.rs
rename to core/server_common/src/sharding/partition_location.rs
diff --git a/core/common/src/sharding/shard_id.rs b/core/server_common/src/sharding/shard_id.rs
similarity index 100%
rename from core/common/src/sharding/shard_id.rs
rename to core/server_common/src/sharding/shard_id.rs
diff --git a/core/shard/Cargo.toml b/core/shard/Cargo.toml
index 9a9a8b9f9..9c315dfcf 100644
--- a/core/shard/Cargo.toml
+++ b/core/shard/Cargo.toml
@@ -37,6 +37,7 @@ metadata = { path = "../metadata" }
papaya = { workspace = true }
partitions = { path = "../partitions" }
prometheus-client = { workspace = true }
+server_common = { path = "../server_common" }
thiserror = { workspace = true }
tracing = { workspace = true }
diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs
index d199b3f33..8b10833ec 100644
--- a/core/shard/src/lib.rs
+++ b/core/shard/src/lib.rs
@@ -28,14 +28,12 @@ use consensus::{
LocalPipeline, MetadataHandle, MuxPlane, PartitionsHandle, Pipeline, Plane, PlaneKind,
Sequencer, VsrAction, VsrConsensus,
};
-use iggy_binary_protocol::consensus::MESSAGE_ALIGN;
-use iggy_binary_protocol::consensus::iobuf::Frozen;
use iggy_binary_protocol::{
- Command2, CommitHeader, DoViewChangeHeader, GenericHeader, Message, MessageBag, PrepareHeader,
- PrepareOkHeader, RequestHeader, StartViewChangeHeader, StartViewHeader,
+ Command2, CommitHeader, DoViewChangeHeader, GenericHeader, PrepareHeader, PrepareOkHeader,
+ RequestHeader, StartViewChangeHeader, StartViewHeader,
};
+use iggy_common::PartitionStats;
use iggy_common::variadic;
-use iggy_common::{PartitionStats, sharding::IggyNamespace};
use journal::{Journal, JournalHandle};
use message_bus::MessageBus;
use message_bus::client_listener::RequestHandler;
@@ -46,6 +44,8 @@ use metadata::IggyMetadata;
use metadata::impls::metadata::StreamsFrontend;
use metadata::stm::StateMachine;
use partitions::{IggyPartition, IggyPartitions};
+use server_common::sharding::IggyNamespace;
+use server_common::{MESSAGE_ALIGN, Message, MessageBag, iobuf::Frozen};
use shards_table::ShardsTable;
use std::rc::Rc;
use std::sync::Arc;
diff --git a/core/shard/src/router.rs b/core/shard/src/router.rs
index bf188841e..5060f713c 100644
--- a/core/shard/src/router.rs
+++ b/core/shard/src/router.rs
@@ -20,14 +20,13 @@ use crate::shards_table::{ShardsTable, calculate_shard_from_consensus_ns};
use crate::{IggyShard, LifecycleFrame, Receiver, ShardFrame};
use crossfire::TrySendError;
use futures::FutureExt;
-use iggy_binary_protocol::{
- ConsensusHeader, GenericHeader, Message, MessageBag, Operation, PrepareHeader,
-};
-use iggy_common::sharding::{IggyNamespace, METADATA_CONSENSUS_NAMESPACE};
+use iggy_binary_protocol::{ConsensusHeader, GenericHeader, Operation, PrepareHeader};
use journal::{Journal, JournalHandle};
use message_bus::{ConnectionInstaller, MessageBus};
use metadata::impls::metadata::StreamsFrontend;
use metadata::stm::StateMachine;
+use server_common::sharding::{IggyNamespace, METADATA_CONSENSUS_NAMESPACE};
+use server_common::{Message, MessageBag};
/// Decompose a [`MessageBag`] into the routing-relevant tuple
/// `(operation, namespace, generic_message)`.
diff --git a/core/shard/src/shards_table.rs b/core/shard/src/shards_table.rs
index 6829f28cd..b7c7204e3 100644
--- a/core/shard/src/shards_table.rs
+++ b/core/shard/src/shards_table.rs
@@ -16,7 +16,7 @@
// under the License.
use hash32::{Hasher, Murmur3Hasher};
-use iggy_common::sharding::{IggyNamespace, PartitionLocation};
+use server_common::sharding::{IggyNamespace, PartitionLocation};
use std::hash::Hasher as _;
/// Lookup table that maps partition namespaces to their owning shard.
diff --git a/core/simulator/Cargo.toml b/core/simulator/Cargo.toml
index ab32a58cd..8b3c87255 100644
--- a/core/simulator/Cargo.toml
+++ b/core/simulator/Cargo.toml
@@ -36,6 +36,7 @@ metadata = { path = "../metadata" }
partitions = { path = "../partitions" }
rand = { workspace = true }
rand_xoshiro = { workspace = true }
+server_common = { path = "../server_common" }
shard = { path = "../shard" }
tracing = { workspace = true }
diff --git a/core/simulator/src/bus.rs b/core/simulator/src/bus.rs
index 5e6769b77..a6c65915d 100644
--- a/core/simulator/src/bus.rs
+++ b/core/simulator/src/bus.rs
@@ -15,10 +15,12 @@
// specific language governing permissions and limitations
// under the License.
-use iggy_binary_protocol::consensus::MESSAGE_ALIGN;
-use iggy_binary_protocol::consensus::iobuf::{Frozen, Owned};
-use iggy_binary_protocol::{GenericHeader, Message};
+use iggy_binary_protocol::GenericHeader;
use message_bus::{MessageBus, SendError};
+use server_common::{
+ MESSAGE_ALIGN, Message,
+ iobuf::{Frozen, Owned},
+};
use std::collections::{HashSet, VecDeque};
use std::ops::Deref;
use std::sync::{Arc, Mutex};
diff --git a/core/simulator/src/client.rs b/core/simulator/src/client.rs
index 0c2c5b588..67c6d81d8 100644
--- a/core/simulator/src/client.rs
+++ b/core/simulator/src/client.rs
@@ -16,15 +16,16 @@
// under the License.
use bytes::Bytes;
-use iggy_binary_protocol::consensus::iobuf::Owned;
use iggy_binary_protocol::requests::streams::{CreateStreamRequest, DeleteStreamRequest};
use iggy_binary_protocol::{
- AckLevel, Message, Operation, RequestHeader, WireEncode, WireIdentifier, WireName,
+ AckLevel, Operation, RequestHeader, WireEncode, WireIdentifier, WireName,
};
-use iggy_common::send_messages2::{
- IggyMessage2, IggyMessage2Header, IggyMessages2, SendMessages2Owned,
+use server_common::sharding::IggyNamespace;
+use server_common::{
+ Message,
+ iobuf::Owned,
+ send_messages2::{IggyMessage2, IggyMessage2Header, IggyMessages2, SendMessages2Owned},
};
-use iggy_common::sharding::IggyNamespace;
use std::cell::Cell;
// TODO: Proper client which implements the full client SDK API
diff --git a/core/simulator/src/deps.rs b/core/simulator/src/deps.rs
index 5033d88a4..4a998f731 100644
--- a/core/simulator/src/deps.rs
+++ b/core/simulator/src/deps.rs
@@ -15,14 +15,14 @@
// specific language governing permissions and limitations
// under the License.
-use iggy_binary_protocol::consensus::iobuf::Owned;
-use iggy_binary_protocol::{Message, PrepareHeader};
+use iggy_binary_protocol::PrepareHeader;
use iggy_common::variadic;
use journal::{Journal, JournalHandle, Storage};
use metadata::MuxStateMachine;
use metadata::stm::consumer_group::ConsumerGroups;
use metadata::stm::stream::Streams;
use metadata::stm::user::Users;
+use server_common::{Message, iobuf::Owned};
use std::cell::{Cell, RefCell, UnsafeCell};
use std::collections::HashMap;
diff --git a/core/simulator/src/lib.rs b/core/simulator/src/lib.rs
index 28d777a5f..c9556cde3 100644
--- a/core/simulator/src/lib.rs
+++ b/core/simulator/src/lib.rs
@@ -26,13 +26,14 @@ pub mod replica;
use bus::SimOutbox;
use client::SimClient;
use consensus::PartitionsHandle;
-use iggy_binary_protocol::{GenericHeader, Message, ReplyHeader};
+use iggy_binary_protocol::{GenericHeader, ReplyHeader};
use iggy_common::IggyError;
-use iggy_common::sharding::IggyNamespace;
use network::Network;
use packet::{PacketSimulatorOptions, ProcessId};
use partitions::{Partition, PartitionOffsets, PollQueryResult, PollingArgs, PollingConsumer};
use replica::{Replica, new_replica};
+use server_common::Message;
+use server_common::sharding::IggyNamespace;
use std::collections::HashSet;
use std::sync::Arc;
@@ -348,14 +349,14 @@ mod tests {
use super::*;
use crate::client::SimClient;
use consensus::Status;
- use iggy_common::sharding::IggyNamespace;
+ use server_common::sharding::IggyNamespace;
/// After crashing the primary in a 5-node cluster, the 4 remaining
/// backups should detect the failure via heartbeat timeout and elect
/// a new primary through the view change protocol.
#[test]
fn view_change_after_primary_crash() {
- iggy_common::MemoryPool::init_pool(&iggy_common::MemoryPoolConfigOther {
+ server_common::MemoryPool::init_pool(&server_common::MemoryPoolConfigOther {
enabled: false,
size: iggy_common::IggyByteSize::from(0u64),
bucket_capacity: 1,
@@ -461,7 +462,7 @@ mod tests {
/// key / content if they need at-most-once-per-payload.
#[test]
fn failover_retry_re_executes_under_at_least_once() {
- iggy_common::MemoryPool::init_pool(&iggy_common::MemoryPoolConfigOther {
+ server_common::MemoryPool::init_pool(&server_common::MemoryPoolConfigOther {
enabled: false,
size: iggy_common::IggyByteSize::from(0u64),
bucket_capacity: 1,
@@ -577,7 +578,7 @@ mod tests {
/// which fails when the new primary hasn't caught up yet.
#[test]
fn view_change_behind_backup_becomes_primary() {
- iggy_common::MemoryPool::init_pool(&iggy_common::MemoryPoolConfigOther {
+ server_common::MemoryPool::init_pool(&server_common::MemoryPoolConfigOther {
enabled: false,
size: iggy_common::IggyByteSize::from(0u64),
bucket_capacity: 1,
diff --git a/core/simulator/src/main.rs b/core/simulator/src/main.rs
index dc34745ea..bf0edf17d 100644
--- a/core/simulator/src/main.rs
+++ b/core/simulator/src/main.rs
@@ -15,10 +15,11 @@
// specific language governing permissions and limitations
// under the License.
-use iggy_binary_protocol::{Message, ReplyHeader};
-use iggy_common::sharding::IggyNamespace;
-use iggy_common::{IggyByteSize, MemoryPool, MemoryPoolConfigOther, PollingStrategy};
+use iggy_binary_protocol::ReplyHeader;
+use iggy_common::{IggyByteSize, PollingStrategy};
use partitions::{PollingArgs, PollingConsumer};
+use server_common::sharding::IggyNamespace;
+use server_common::{MemoryPool, MemoryPoolConfigOther, Message};
use simulator::Simulator;
use simulator::client::SimClient;
use simulator::packet::PacketSimulatorOptions;
diff --git a/core/simulator/src/network.rs b/core/simulator/src/network.rs
index 269541a10..6a4aa795e 100644
--- a/core/simulator/src/network.rs
+++ b/core/simulator/src/network.rs
@@ -24,7 +24,8 @@
use crate::packet::{
ALLOW_ALL, BLOCK_ALL, LinkFilter, Packet, PacketSimulator, PacketSimulatorOptions, ProcessId,
};
-use iggy_binary_protocol::{GenericHeader, Message};
+use iggy_binary_protocol::GenericHeader;
+use server_common::Message;
/// Network layer for the cluster simulation.
///
diff --git a/core/simulator/src/packet.rs b/core/simulator/src/packet.rs
index c523ee45e..be6a54829 100644
--- a/core/simulator/src/packet.rs
+++ b/core/simulator/src/packet.rs
@@ -40,10 +40,11 @@
use crate::ready_queue::{Ready, ReadyQueue};
use enumset::EnumSet;
-use iggy_binary_protocol::{Command2, GenericHeader, Message};
+use iggy_binary_protocol::{Command2, GenericHeader};
use rand::RngExt;
use rand_xoshiro::Xoshiro256Plus;
use rand_xoshiro::rand_core::SeedableRng;
+use server_common::Message;
use std::collections::HashMap;
/// Per-link command filter. An `EnumSet<Command2>` where:
@@ -781,10 +782,8 @@ mod tests {
let header: &mut GenericHeader =
bytemuck::checked::try_from_bytes_mut(&mut buf).expect("zeroed bytes are valid");
header.command = command;
- Message::try_from(
- iggy_binary_protocol::consensus::iobuf::Owned::<4096>::copy_from_slice(&buf),
- )
- .expect("generic test buffer must contain a valid generic message")
+ Message::try_from(server_common::iobuf::Owned::<4096>::copy_from_slice(&buf))
+ .expect("generic test buffer must contain a valid generic message")
}
/// Helper: disable all links to/from a given replica (isolate it).
diff --git a/core/simulator/src/replica.rs b/core/simulator/src/replica.rs
index ae1f65ea0..399706d76 100644
--- a/core/simulator/src/replica.rs
+++ b/core/simulator/src/replica.rs
@@ -19,13 +19,13 @@ use crate::bus::{SharedSimOutbox, SimOutbox};
use crate::deps::{MemStorage, SimJournal, SimMuxStateMachine, SimSnapshot};
use consensus::{LocalPipeline, VsrConsensus};
use iggy_common::IggyByteSize;
-use iggy_common::sharding::ShardId;
use iggy_common::variadic;
use metadata::IggyMetadata;
use metadata::stm::consumer_group::{ConsumerGroups, ConsumerGroupsInner};
use metadata::stm::stream::{Streams, StreamsInner};
use metadata::stm::user::{Users, UsersInner};
use partitions::{IggyPartitions, PartitionsConfig};
+use server_common::sharding::ShardId;
use std::sync::Arc;
// TODO: Make configurable