You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by pi...@apache.org on 2025/05/05 18:38:54 UTC
(iggy) branch master updated: chore(repo): migrate java sdk (#1735)
This is an automated email from the ASF dual-hosted git repository.
piotr pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new c1a41aad chore(repo): migrate java sdk (#1735)
c1a41aad is described below
commit c1a41aad0db1d2a7721bd80907f40deffaa16c30
Author: Nandakumar Vadivelu <na...@apache.org>
AuthorDate: Tue May 6 00:08:51 2025 +0530
chore(repo): migrate java sdk (#1735)
Move the existing [Java SDK](https://github.com/iggy-rs/iggy-java-client) to the main Iggy repository under `foreign` directory. Original Java SDK for Iggy was created by mmodzelewski (Maciej Modzelewski)
This will resolve #1731
---
.gitignore | 4 +
foreign/java/.java-version | 1 +
foreign/java/LICENSE | 201 +++++++++++
foreign/java/NOTICE | 13 +
foreign/java/README.md | 27 ++
.../java/examples/simple-consumer/build.gradle.kts | 16 +
.../src/main/java/rs/iggy/SimpleConsumer.java | 93 +++++
.../java/examples/simple-producer/build.gradle.kts | 16 +
.../src/main/java/rs/iggy/SimpleProducer.java | 65 ++++
foreign/java/gradle/wrapper/gradle-wrapper.jar | Bin 0 -> 60756 bytes
.../java/gradle/wrapper/gradle-wrapper.properties | 6 +
foreign/java/gradlew | 234 +++++++++++++
foreign/java/gradlew.bat | 89 +++++
foreign/java/java-sdk/build.gradle.kts | 96 +++++
.../java/java-sdk/src/main/java/rs/iggy/Iggy.java | 14 +
.../java-sdk/src/main/java/rs/iggy/NonNullApi.java | 13 +
.../clients/blocking/ConsumerGroupsClient.java | 49 +++
.../clients/blocking/ConsumerOffsetsClient.java | 24 ++
.../rs/iggy/clients/blocking/IggyBaseClient.java | 23 ++
.../java/rs/iggy/clients/blocking/IggyClient.java | 15 +
.../iggy/clients/blocking/IggyClientBuilder.java | 19 +
.../rs/iggy/clients/blocking/MessagesClient.java | 28 ++
.../rs/iggy/clients/blocking/PartitionsClient.java | 20 ++
.../blocking/PersonalAccessTokensClient.java | 19 +
.../rs/iggy/clients/blocking/StreamsClient.java | 33 ++
.../rs/iggy/clients/blocking/SystemClient.java | 20 ++
.../rs/iggy/clients/blocking/TopicsClient.java | 83 +++++
.../java/rs/iggy/clients/blocking/UsersClient.java | 48 +++
.../blocking/http/ConsumerGroupsHttpClient.java | 65 ++++
.../blocking/http/ConsumerOffsetsHttpClient.java | 42 +++
.../iggy/clients/blocking/http/IggyHttpClient.java | 75 ++++
.../clients/blocking/http/InternalHttpClient.java | 143 ++++++++
.../clients/blocking/http/MessagesHttpClient.java | 47 +++
.../clients/blocking/http/ObjectMapperFactory.java | 28 ++
.../blocking/http/PartitionsHttpClient.java | 35 ++
.../http/PersonalAccessTokensHttpClient.java | 57 +++
.../clients/blocking/http/StreamsHttpClient.java | 57 +++
.../clients/blocking/http/SystemHttpClient.java | 50 +++
.../clients/blocking/http/TopicsHttpClient.java | 96 +++++
.../clients/blocking/http/UsersHttpClient.java | 96 +++++
.../clients/blocking/http/error/IggyHttpError.java | 9 +
.../blocking/http/error/IggyHttpException.java | 15 +
.../iggy/clients/blocking/http/package-info.java | 4 +
.../clients/blocking/tcp/BytesDeserializer.java | 388 +++++++++++++++++++++
.../iggy/clients/blocking/tcp/BytesSerializer.java | 200 +++++++++++
.../blocking/tcp/ConsumerGroupsTcpClient.java | 99 ++++++
.../blocking/tcp/ConsumerOffsetTcpClient.java | 50 +++
.../iggy/clients/blocking/tcp/IggyTcpClient.java | 76 ++++
.../clients/blocking/tcp/InternalTcpClient.java | 87 +++++
.../clients/blocking/tcp/MessagesTcpClient.java | 52 +++
.../clients/blocking/tcp/PartitionsTcpClient.java | 33 ++
.../tcp/PersonalAccessTokensTcpClient.java | 62 ++++
.../clients/blocking/tcp/StreamsTcpClient.java | 78 +++++
.../iggy/clients/blocking/tcp/SystemTcpClient.java | 60 ++++
.../iggy/clients/blocking/tcp/TopicsTcpClient.java | 98 ++++++
.../iggy/clients/blocking/tcp/UsersTcpClient.java | 137 ++++++++
.../rs/iggy/clients/blocking/tcp/package-info.java | 4 +
.../main/java/rs/iggy/consumergroup/Consumer.java | 38 ++
.../java/rs/iggy/consumergroup/ConsumerGroup.java | 9 +
.../iggy/consumergroup/ConsumerGroupDetails.java | 19 +
.../rs/iggy/consumergroup/ConsumerGroupMember.java | 10 +
.../java/rs/iggy/consumergroup/package-info.java | 4 +
.../rs/iggy/consumeroffset/ConsumerOffsetInfo.java | 10 +
.../java/rs/iggy/consumeroffset/package-info.java | 4 +
.../main/java/rs/iggy/identifier/ConsumerId.java | 19 +
.../main/java/rs/iggy/identifier/Identifier.java | 51 +++
.../src/main/java/rs/iggy/identifier/StreamId.java | 19 +
.../src/main/java/rs/iggy/identifier/TopicId.java | 19 +
.../src/main/java/rs/iggy/identifier/UserId.java | 19 +
.../main/java/rs/iggy/identifier/package-info.java | 4 +
.../java/rs/iggy/message/BigIntegerMessageId.java | 34 ++
.../main/java/rs/iggy/message/BytesMessageId.java | 34 ++
.../src/main/java/rs/iggy/message/HeaderKind.java | 38 ++
.../src/main/java/rs/iggy/message/HeaderValue.java | 4 +
.../src/main/java/rs/iggy/message/Message.java | 16 +
.../src/main/java/rs/iggy/message/MessageId.java | 24 ++
.../main/java/rs/iggy/message/MessageState.java | 23 ++
.../main/java/rs/iggy/message/Partitioning.java | 29 ++
.../java/rs/iggy/message/PartitioningKind.java | 22 ++
.../main/java/rs/iggy/message/PolledMessage.java | 16 +
.../main/java/rs/iggy/message/PolledMessages.java | 11 +
.../src/main/java/rs/iggy/message/PollingKind.java | 19 +
.../main/java/rs/iggy/message/PollingStrategy.java | 27 ++
.../main/java/rs/iggy/message/UuidMessageId.java | 37 ++
.../main/java/rs/iggy/message/package-info.java | 4 +
.../src/main/java/rs/iggy/partition/Partition.java | 13 +
.../main/java/rs/iggy/partition/package-info.java | 4 +
.../PersonalAccessTokenInfo.java | 7 +
.../RawPersonalAccessToken.java | 4 +
.../src/main/java/rs/iggy/stream/StreamBase.java | 13 +
.../main/java/rs/iggy/stream/StreamDetails.java | 19 +
.../src/main/java/rs/iggy/stream/package-info.java | 4 +
.../src/main/java/rs/iggy/system/ClientInfo.java | 12 +
.../java/rs/iggy/system/ClientInfoDetails.java | 18 +
.../java/rs/iggy/system/ConsumerGroupInfo.java | 8 +
.../src/main/java/rs/iggy/system/Stats.java | 29 ++
.../src/main/java/rs/iggy/system/package-info.java | 4 +
.../java/rs/iggy/topic/CompressionAlgorithm.java | 25 ++
.../src/main/java/rs/iggy/topic/Topic.java | 17 +
.../src/main/java/rs/iggy/topic/TopicDetails.java | 33 ++
.../src/main/java/rs/iggy/topic/package-info.java | 4 +
.../main/java/rs/iggy/user/GlobalPermissions.java | 15 +
.../src/main/java/rs/iggy/user/IdentityInfo.java | 6 +
.../src/main/java/rs/iggy/user/Permissions.java | 12 +
.../main/java/rs/iggy/user/StreamPermissions.java | 17 +
.../src/main/java/rs/iggy/user/TokenInfo.java | 6 +
.../main/java/rs/iggy/user/TopicPermissions.java | 9 +
.../src/main/java/rs/iggy/user/UserInfo.java | 11 +
.../main/java/rs/iggy/user/UserInfoDetails.java | 16 +
.../src/main/java/rs/iggy/user/UserStatus.java | 29 ++
.../src/main/java/rs/iggy/user/package-info.java | 4 +
.../java-sdk/src/test/java/rs/iggy/IggyTest.java | 20 ++
.../blocking/ConsumerGroupsClientBaseTest.java | 90 +++++
.../blocking/ConsumerOffsetsClientBaseTest.java | 50 +++
.../rs/iggy/clients/blocking/IntegrationTest.java | 53 +++
.../clients/blocking/MessagesClientBaseTest.java | 74 ++++
.../clients/blocking/PartitionsClientBaseTest.java | 43 +++
.../blocking/PersonalAccessTokensBaseTest.java | 64 ++++
.../clients/blocking/StreamClientBaseTest.java | 70 ++++
.../clients/blocking/SystemClientBaseTest.java | 27 ++
.../clients/blocking/TopicsClientBaseTest.java | 88 +++++
.../iggy/clients/blocking/UsersClientBaseTest.java | 137 ++++++++
.../http/ConsumerGroupsHttpClientTest.java | 13 +
.../http/ConsumerOffsetsHttpClientTest.java | 13 +
.../clients/blocking/http/HttpClientFactory.java | 14 +
.../blocking/http/MessagesHttpClientTest.java | 13 +
.../blocking/http/PartitionsHttpClientTest.java | 13 +
.../http/PersonalAccessTokensHttpClientTest.java | 13 +
.../blocking/http/StreamHttpClientTest.java | 13 +
.../blocking/http/SystemHttpClientTest.java | 13 +
.../blocking/http/TopicsHttpClientTest.java | 13 +
.../clients/blocking/http/UsersHttpClientTest.java | 27 ++
.../clients/blocking/tcp/BytesSerializerTest.java | 101 ++++++
.../blocking/tcp/ConsumerGroupsTcpClientTest.java | 40 +++
.../blocking/tcp/ConsumerOffsetsTcpClientTest.java | 13 +
.../blocking/tcp/MessagesTcpClientTest.java | 13 +
.../blocking/tcp/PartitionsTcpClientTest.java | 13 +
.../tcp/PersonalAccessTokensTcpClientTest.java | 13 +
.../clients/blocking/tcp/StreamTcpClientTest.java | 13 +
.../clients/blocking/tcp/SystemTcpClientTest.java | 42 +++
.../clients/blocking/tcp/TcpClientFactory.java | 14 +
.../clients/blocking/tcp/TopicsTcpClientTest.java | 13 +
.../clients/blocking/tcp/UsersTcpClientTest.java | 22 ++
.../java-sdk/src/test/resources/logback-test.xml | 19 +
foreign/java/settings.gradle.kts | 8 +
145 files changed, 5599 insertions(+)
diff --git a/.gitignore b/.gitignore
index be843e58..71dcd361 100644
--- a/.gitignore
+++ b/.gitignore
@@ -2,6 +2,7 @@ target*
.idea/
.vscode/
.DS_Store
+.gradle
local_data*
bench_*
/sdk/errors_table/
@@ -12,3 +13,6 @@ bench_*
/iggy_release_tmp
/**/obj
/**/bin
+*.iml
+foreign/java/**/build
+foreign/java/**/out
diff --git a/foreign/java/.java-version b/foreign/java/.java-version
new file mode 100644
index 00000000..98d9bcb7
--- /dev/null
+++ b/foreign/java/.java-version
@@ -0,0 +1 @@
+17
diff --git a/foreign/java/LICENSE b/foreign/java/LICENSE
new file mode 100644
index 00000000..9c8f3ea0
--- /dev/null
+++ b/foreign/java/LICENSE
@@ -0,0 +1,201 @@
+ Apache License
+ Version 2.0, January 2004
+ http://www.apache.org/licenses/
+
+ TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
+
+ 1. Definitions.
+
+ "License" shall mean the terms and conditions for use, reproduction,
+ and distribution as defined by Sections 1 through 9 of this document.
+
+ "Licensor" shall mean the copyright owner or entity authorized by
+ the copyright owner that is granting the License.
+
+ "Legal Entity" shall mean the union of the acting entity and all
+ other entities that control, are controlled by, or are under common
+ control with that entity. For the purposes of this definition,
+ "control" means (i) the power, direct or indirect, to cause the
+ direction or management of such entity, whether by contract or
+ otherwise, or (ii) ownership of fifty percent (50%) or more of the
+ outstanding shares, or (iii) beneficial ownership of such entity.
+
+ "You" (or "Your") shall mean an individual or Legal Entity
+ exercising permissions granted by this License.
+
+ "Source" form shall mean the preferred form for making modifications,
+ including but not limited to software source code, documentation
+ source, and configuration files.
+
+ "Object" form shall mean any form resulting from mechanical
+ transformation or translation of a Source form, including but
+ not limited to compiled object code, generated documentation,
+ and conversions to other media types.
+
+ "Work" shall mean the work of authorship, whether in Source or
+ Object form, made available under the License, as indicated by a
+ copyright notice that is included in or attached to the work
+ (an example is provided in the Appendix below).
+
+ "Derivative Works" shall mean any work, whether in Source or Object
+ form, that is based on (or derived from) the Work and for which the
+ editorial revisions, annotations, elaborations, or other modifications
+ represent, as a whole, an original work of authorship. For the purposes
+ of this License, Derivative Works shall not include works that remain
+ separable from, or merely link (or bind by name) to the interfaces of,
+ the Work and Derivative Works thereof.
+
+ "Contribution" shall mean any work of authorship, including
+ the original version of the Work and any modifications or additions
+ to that Work or Derivative Works thereof, that is intentionally
+ submitted to Licensor for inclusion in the Work by the copyright owner
+ or by an individual or Legal Entity authorized to submit on behalf of
+ the copyright owner. For the purposes of this definition, "submitted"
+ means any form of electronic, verbal, or written communication sent
+ to the Licensor or its representatives, including but not limited to
+ communication on electronic mailing lists, source code control systems,
+ and issue tracking systems that are managed by, or on behalf of, the
+ Licensor for the purpose of discussing and improving the Work, but
+ excluding communication that is conspicuously marked or otherwise
+ designated in writing by the copyright owner as "Not a Contribution."
+
+ "Contributor" shall mean Licensor and any individual or Legal Entity
+ on behalf of whom a Contribution has been received by Licensor and
+ subsequently incorporated within the Work.
+
+ 2. Grant of Copyright License. Subject to the terms and conditions of
+ this License, each Contributor hereby grants to You a perpetual,
+ worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+ copyright license to reproduce, prepare Derivative Works of,
+ publicly display, publicly perform, sublicense, and distribute the
+ Work and such Derivative Works in Source or Object form.
+
+ 3. Grant of Patent License. Subject to the terms and conditions of
+ this License, each Contributor hereby grants to You a perpetual,
+ worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+ (except as stated in this section) patent license to make, have made,
+ use, offer to sell, sell, import, and otherwise transfer the Work,
+ where such license applies only to those patent claims licensable
+ by such Contributor that are necessarily infringed by their
+ Contribution(s) alone or by combination of their Contribution(s)
+ with the Work to which such Contribution(s) was submitted. If You
+ institute patent litigation against any entity (including a
+ cross-claim or counterclaim in a lawsuit) alleging that the Work
+ or a Contribution incorporated within the Work constitutes direct
+ or contributory patent infringement, then any patent licenses
+ granted to You under this License for that Work shall terminate
+ as of the date such litigation is filed.
+
+ 4. Redistribution. You may reproduce and distribute copies of the
+ Work or Derivative Works thereof in any medium, with or without
+ modifications, and in Source or Object form, provided that You
+ meet the following conditions:
+
+ (a) You must give any other recipients of the Work or
+ Derivative Works a copy of this License; and
+
+ (b) You must cause any modified files to carry prominent notices
+ stating that You changed the files; and
+
+ (c) You must retain, in the Source form of any Derivative Works
+ that You distribute, all copyright, patent, trademark, and
+ attribution notices from the Source form of the Work,
+ excluding those notices that do not pertain to any part of
+ the Derivative Works; and
+
+ (d) If the Work includes a "NOTICE" text file as part of its
+ distribution, then any Derivative Works that You distribute must
+ include a readable copy of the attribution notices contained
+ within such NOTICE file, excluding those notices that do not
+ pertain to any part of the Derivative Works, in at least one
+ of the following places: within a NOTICE text file distributed
+ as part of the Derivative Works; within the Source form or
+ documentation, if provided along with the Derivative Works; or,
+ within a display generated by the Derivative Works, if and
+ wherever such third-party notices normally appear. The contents
+ of the NOTICE file are for informational purposes only and
+ do not modify the License. You may add Your own attribution
+ notices within Derivative Works that You distribute, alongside
+ or as an addendum to the NOTICE text from the Work, provided
+ that such additional attribution notices cannot be construed
+ as modifying the License.
+
+ You may add Your own copyright statement to Your modifications and
+ may provide additional or different license terms and conditions
+ for use, reproduction, or distribution of Your modifications, or
+ for any such Derivative Works as a whole, provided Your use,
+ reproduction, and distribution of the Work otherwise complies with
+ the conditions stated in this License.
+
+ 5. Submission of Contributions. Unless You explicitly state otherwise,
+ any Contribution intentionally submitted for inclusion in the Work
+ by You to the Licensor shall be under the terms and conditions of
+ this License, without any additional terms or conditions.
+ Notwithstanding the above, nothing herein shall supersede or modify
+ the terms of any separate license agreement you may have executed
+ with Licensor regarding such Contributions.
+
+ 6. Trademarks. This License does not grant permission to use the trade
+ names, trademarks, service marks, or product names of the Licensor,
+ except as required for reasonable and customary use in describing the
+ origin of the Work and reproducing the content of the NOTICE file.
+
+ 7. Disclaimer of Warranty. Unless required by applicable law or
+ agreed to in writing, Licensor provides the Work (and each
+ Contributor provides its Contributions) on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+ implied, including, without limitation, any warranties or conditions
+ of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
+ PARTICULAR PURPOSE. You are solely responsible for determining the
+ appropriateness of using or redistributing the Work and assume any
+ risks associated with Your exercise of permissions under this License.
+
+ 8. Limitation of Liability. In no event and under no legal theory,
+ whether in tort (including negligence), contract, or otherwise,
+ unless required by applicable law (such as deliberate and grossly
+ negligent acts) or agreed to in writing, shall any Contributor be
+ liable to You for damages, including any direct, indirect, special,
+ incidental, or consequential damages of any character arising as a
+ result of this License or out of the use or inability to use the
+ Work (including but not limited to damages for loss of goodwill,
+ work stoppage, computer failure or malfunction, or any and all
+ other commercial damages or losses), even if such Contributor
+ has been advised of the possibility of such damages.
+
+ 9. Accepting Warranty or Additional Liability. While redistributing
+ the Work or Derivative Works thereof, You may choose to offer,
+ and charge a fee for, acceptance of support, warranty, indemnity,
+ or other liability obligations and/or rights consistent with this
+ License. However, in accepting such obligations, You may act only
+ on Your own behalf and on Your sole responsibility, not on behalf
+ of any other Contributor, and only if You agree to indemnify,
+ defend, and hold each Contributor harmless for any liability
+ incurred by, or claims asserted against, such Contributor by reason
+ of your accepting any such warranty or additional liability.
+
+ END OF TERMS AND CONDITIONS
+
+ APPENDIX: How to apply the Apache License to your work.
+
+ To apply the Apache License to your work, attach the following
+ boilerplate notice, with the fields enclosed by brackets "{}"
+ replaced with your own identifying information. (Don't include
+ the brackets!) The text should be enclosed in the appropriate
+ comment syntax for the file format. We also recommend that a
+ file or class name and description of purpose be included on the
+ same "printed page" as the copyright notice for easier
+ identification within third-party archives.
+
+ Copyright {yyyy} {name of copyright owner}
+
+ Licensed 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.
\ No newline at end of file
diff --git a/foreign/java/NOTICE b/foreign/java/NOTICE
new file mode 100644
index 00000000..63d71ff1
--- /dev/null
+++ b/foreign/java/NOTICE
@@ -0,0 +1,13 @@
+Copyright 2023-2025 Piotr Gankiewicz, LaserData, Inc.
+
+Licensed 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.
\ No newline at end of file
diff --git a/foreign/java/README.md b/foreign/java/README.md
new file mode 100644
index 00000000..7811ec63
--- /dev/null
+++ b/foreign/java/README.md
@@ -0,0 +1,27 @@
+# Java SDK for Iggy
+
+[Iggy GitHub](https://github.com/iggy-rs/iggy) | [Website](https://iggy.rs) | [Getting started](https://docs.iggy.rs/introduction/getting-started/) | [Documentation](https://docs.iggy.rs) | [Blog](https://blog.iggy.rs) | [Discord](https://iggy.rs/discord)
+
+[](https://github.com/iggy-rs/iggy-java-client/actions/workflows/tests.yml)
+[](https://twitter.com/iggy_rs_)
+
+---
+
+Official Java client SDK for [Iggy.rs](https://iggy.rs) message streaming.
+
+The client currently supports HTTP and TCP protocols with blocking implementation.
+
+### Adding the client to your project
+
+Add dependency to `pom.xml` or `build.gradle` file.
+
+You can find the latest version in Maven Central repository:
+
+https://central.sonatype.com/artifact/rs.iggy/iggy-java-sdk
+
+### Implement consumer and producer
+
+You can find examples for
+simple [consumer](https://github.com/iggy-rs/iggy-java-client/blob/main/examples/simple-consumer/src/main/java/rs/iggy/SimpleConsumer.java)
+and [producer](https://github.com/iggy-rs/iggy-java-client/blob/main/examples/simple-producer/src/main/java/rs/iggy/SimpleProducer.java)
+in the repository.
diff --git a/foreign/java/examples/simple-consumer/build.gradle.kts b/foreign/java/examples/simple-consumer/build.gradle.kts
new file mode 100644
index 00000000..167e15ef
--- /dev/null
+++ b/foreign/java/examples/simple-consumer/build.gradle.kts
@@ -0,0 +1,16 @@
+plugins {
+ id("java")
+}
+
+group = "rs.iggy"
+version = "0.0.1-SNAPSHOT"
+
+repositories {
+ mavenCentral()
+}
+
+dependencies {
+ implementation(project(":java-sdk"))
+ implementation("org.slf4j:slf4j-api:2.0.9")
+ runtimeOnly("ch.qos.logback:logback-classic:1.4.12")
+}
diff --git a/foreign/java/examples/simple-consumer/src/main/java/rs/iggy/SimpleConsumer.java b/foreign/java/examples/simple-consumer/src/main/java/rs/iggy/SimpleConsumer.java
new file mode 100644
index 00000000..64ea7816
--- /dev/null
+++ b/foreign/java/examples/simple-consumer/src/main/java/rs/iggy/SimpleConsumer.java
@@ -0,0 +1,93 @@
+package rs.iggy;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import rs.iggy.clients.blocking.IggyBaseClient;
+import rs.iggy.clients.blocking.tcp.IggyTcpClient;
+import rs.iggy.consumergroup.Consumer;
+import rs.iggy.consumergroup.ConsumerGroupDetails;
+import rs.iggy.identifier.ConsumerId;
+import rs.iggy.identifier.StreamId;
+import rs.iggy.identifier.TopicId;
+import rs.iggy.message.PolledMessage;
+import rs.iggy.message.PollingStrategy;
+import rs.iggy.stream.StreamDetails;
+import rs.iggy.topic.CompressionAlgorithm;
+import rs.iggy.topic.TopicDetails;
+import java.math.BigInteger;
+import java.util.ArrayList;
+import java.util.Optional;
+import static java.util.Optional.empty;
+
+public class SimpleConsumer {
+
+ private static final String STREAM_NAME = "dev01";
+ private static final StreamId STREAM_ID = StreamId.of(STREAM_NAME);
+ private static final String TOPIC_NAME = "events";
+ private static final TopicId TOPIC_ID = TopicId.of(TOPIC_NAME);
+ private static final String GROUP_NAME = "simple-consumer";
+ private static final ConsumerId GROUP_ID = ConsumerId.of(GROUP_NAME);
+ private static final Logger log = LoggerFactory.getLogger(SimpleConsumer.class);
+
+ public static void main(String[] args) {
+ var client = new IggyTcpClient("localhost", 8090);
+ client.users().login("iggy", "iggy");
+
+ createStream(client);
+ createTopic(client);
+ createConsumerGroup(client);
+ client.consumerGroups().joinConsumerGroup(STREAM_ID, TOPIC_ID, GROUP_ID);
+
+ var messages = new ArrayList<PolledMessage>();
+ while (messages.size() < 1000) {
+ var polledMessages = client.messages()
+ .pollMessages(STREAM_ID,
+ TOPIC_ID,
+ empty(),
+ Consumer.group(GROUP_ID),
+ PollingStrategy.next(),
+ 10L,
+ true);
+ messages.addAll(polledMessages.messages());
+ log.debug("Fetched {} messages from partition {}, current offset {}",
+ polledMessages.messages().size(),
+ polledMessages.partitionId(),
+ polledMessages.currentOffset());
+ }
+ }
+
+ private static void createStream(IggyBaseClient client) {
+ Optional<StreamDetails> stream = client.streams().getStream(STREAM_ID);
+ if (stream.isPresent()) {
+ return;
+ }
+ client.streams().createStream(empty(), STREAM_NAME);
+ }
+
+ private static void createTopic(IggyBaseClient client) {
+ Optional<TopicDetails> topic = client.topics().getTopic(STREAM_ID, TOPIC_ID);
+ if (topic.isPresent()) {
+ return;
+ }
+ client.topics()
+ .createTopic(STREAM_ID,
+ empty(),
+ 1L,
+ CompressionAlgorithm.None,
+ BigInteger.ZERO,
+ BigInteger.ZERO,
+ empty(),
+ TOPIC_NAME);
+
+ }
+
+ private static void createConsumerGroup(IggyBaseClient client) {
+ Optional<ConsumerGroupDetails> consumerGroup = client.consumerGroups()
+ .getConsumerGroup(STREAM_ID, TOPIC_ID, GROUP_ID);
+ if (consumerGroup.isPresent()) {
+ return;
+ }
+ client.consumerGroups().createConsumerGroup(STREAM_ID, TOPIC_ID, empty(), GROUP_NAME);
+ }
+
+}
diff --git a/foreign/java/examples/simple-producer/build.gradle.kts b/foreign/java/examples/simple-producer/build.gradle.kts
new file mode 100644
index 00000000..167e15ef
--- /dev/null
+++ b/foreign/java/examples/simple-producer/build.gradle.kts
@@ -0,0 +1,16 @@
+plugins {
+ id("java")
+}
+
+group = "rs.iggy"
+version = "0.0.1-SNAPSHOT"
+
+repositories {
+ mavenCentral()
+}
+
+dependencies {
+ implementation(project(":java-sdk"))
+ implementation("org.slf4j:slf4j-api:2.0.9")
+ runtimeOnly("ch.qos.logback:logback-classic:1.4.12")
+}
diff --git a/foreign/java/examples/simple-producer/src/main/java/rs/iggy/SimpleProducer.java b/foreign/java/examples/simple-producer/src/main/java/rs/iggy/SimpleProducer.java
new file mode 100644
index 00000000..806b08b9
--- /dev/null
+++ b/foreign/java/examples/simple-producer/src/main/java/rs/iggy/SimpleProducer.java
@@ -0,0 +1,65 @@
+package rs.iggy;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import rs.iggy.clients.blocking.tcp.IggyTcpClient;
+import rs.iggy.identifier.StreamId;
+import rs.iggy.identifier.TopicId;
+import rs.iggy.message.Message;
+import rs.iggy.message.Partitioning;
+import rs.iggy.stream.StreamDetails;
+import rs.iggy.topic.CompressionAlgorithm;
+import rs.iggy.topic.TopicDetails;
+import java.math.BigInteger;
+import java.util.Optional;
+import static java.util.Collections.singletonList;
+import static java.util.Optional.empty;
+
+public class SimpleProducer {
+ private static final String STREAM_NAME = "dev01";
+ private static final StreamId STREAM_ID = StreamId.of(STREAM_NAME);
+ private static final String TOPIC_NAME = "events";
+ private static final TopicId TOPIC_ID = TopicId.of(TOPIC_NAME);
+ private static final Logger log = LoggerFactory.getLogger(SimpleProducer.class);
+
+ public static void main(String[] args) {
+ var client = new IggyTcpClient("localhost", 8090);
+ client.users().login("iggy", "iggy");
+
+ createStream(client);
+ createTopic(client);
+
+ int counter = 0;
+ while (counter++ < 1000) {
+ var message = Message.of("message from simple producer " + counter);
+ client.messages().sendMessages(STREAM_ID, TOPIC_ID, Partitioning.balanced(), singletonList(message));
+ log.debug("Message {} sent", counter);
+ }
+
+ }
+
+ private static void createStream(IggyTcpClient client) {
+ Optional<StreamDetails> stream = client.streams().getStream(STREAM_ID);
+ if (stream.isPresent()) {
+ return;
+ }
+ client.streams().createStream(empty(), STREAM_NAME);
+ }
+
+ private static void createTopic(IggyTcpClient client) {
+ Optional<TopicDetails> topic = client.topics().getTopic(STREAM_ID, TOPIC_ID);
+ if (topic.isPresent()) {
+ return;
+ }
+ client.topics()
+ .createTopic(STREAM_ID,
+ empty(),
+ 1L,
+ CompressionAlgorithm.None,
+ BigInteger.ZERO,
+ BigInteger.ZERO,
+ empty(),
+ TOPIC_NAME);
+ }
+
+}
diff --git a/foreign/java/gradle/wrapper/gradle-wrapper.jar b/foreign/java/gradle/wrapper/gradle-wrapper.jar
new file mode 100644
index 00000000..249e5832
Binary files /dev/null and b/foreign/java/gradle/wrapper/gradle-wrapper.jar differ
diff --git a/foreign/java/gradle/wrapper/gradle-wrapper.properties b/foreign/java/gradle/wrapper/gradle-wrapper.properties
new file mode 100644
index 00000000..004e84f5
--- /dev/null
+++ b/foreign/java/gradle/wrapper/gradle-wrapper.properties
@@ -0,0 +1,6 @@
+#Mon Sep 18 13:34:16 CEST 2023
+distributionBase=GRADLE_USER_HOME
+distributionPath=wrapper/dists
+distributionUrl=https\://services.gradle.org/distributions/gradle-8.3-bin.zip
+zipStoreBase=GRADLE_USER_HOME
+zipStorePath=wrapper/dists
diff --git a/foreign/java/gradlew b/foreign/java/gradlew
new file mode 100755
index 00000000..1b6c7873
--- /dev/null
+++ b/foreign/java/gradlew
@@ -0,0 +1,234 @@
+#!/bin/sh
+
+#
+# Copyright © 2015-2021 the original authors.
+#
+# Licensed 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
+#
+# https://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.
+#
+
+##############################################################################
+#
+# Gradle start up script for POSIX generated by Gradle.
+#
+# Important for running:
+#
+# (1) You need a POSIX-compliant shell to run this script. If your /bin/sh is
+# noncompliant, but you have some other compliant shell such as ksh or
+# bash, then to run this script, type that shell name before the whole
+# command line, like:
+#
+# ksh Gradle
+#
+# Busybox and similar reduced shells will NOT work, because this script
+# requires all of these POSIX shell features:
+# * functions;
+# * expansions «$var», «${var}», «${var:-default}», «${var+SET}»,
+# «${var#prefix}», «${var%suffix}», and «$( cmd )»;
+# * compound commands having a testable exit status, especially «case»;
+# * various built-in commands including «command», «set», and «ulimit».
+#
+# Important for patching:
+#
+# (2) This script targets any POSIX shell, so it avoids extensions provided
+# by Bash, Ksh, etc; in particular arrays are avoided.
+#
+# The "traditional" practice of packing multiple parameters into a
+# space-separated string is a well documented source of bugs and security
+# problems, so this is (mostly) avoided, by progressively accumulating
+# options in "$@", and eventually passing that to Java.
+#
+# Where the inherited environment variables (DEFAULT_JVM_OPTS, JAVA_OPTS,
+# and GRADLE_OPTS) rely on word-splitting, this is performed explicitly;
+# see the in-line comments for details.
+#
+# There are tweaks for specific operating systems such as AIX, CygWin,
+# Darwin, MinGW, and NonStop.
+#
+# (3) This script is generated from the Groovy template
+# https://github.com/gradle/gradle/blob/master/subprojects/plugins/src/main/resources/org/gradle/api/internal/plugins/unixStartScript.txt
+# within the Gradle project.
+#
+# You can find Gradle at https://github.com/gradle/gradle/.
+#
+##############################################################################
+
+# Attempt to set APP_HOME
+
+# Resolve links: $0 may be a link
+app_path=$0
+
+# Need this for daisy-chained symlinks.
+while
+ APP_HOME=${app_path%"${app_path##*/}"} # leaves a trailing /; empty if no leading path
+ [ -h "$app_path" ]
+do
+ ls=$( ls -ld "$app_path" )
+ link=${ls#*' -> '}
+ case $link in #(
+ /*) app_path=$link ;; #(
+ *) app_path=$APP_HOME$link ;;
+ esac
+done
+
+APP_HOME=$( cd "${APP_HOME:-./}" && pwd -P ) || exit
+
+APP_NAME="Gradle"
+APP_BASE_NAME=${0##*/}
+
+# Add default JVM options here. You can also use JAVA_OPTS and GRADLE_OPTS to pass JVM options to this script.
+DEFAULT_JVM_OPTS='"-Xmx64m" "-Xms64m"'
+
+# Use the maximum available, or set MAX_FD != -1 to use that value.
+MAX_FD=maximum
+
+warn () {
+ echo "$*"
+} >&2
+
+die () {
+ echo
+ echo "$*"
+ echo
+ exit 1
+} >&2
+
+# OS specific support (must be 'true' or 'false').
+cygwin=false
+msys=false
+darwin=false
+nonstop=false
+case "$( uname )" in #(
+ CYGWIN* ) cygwin=true ;; #(
+ Darwin* ) darwin=true ;; #(
+ MSYS* | MINGW* ) msys=true ;; #(
+ NONSTOP* ) nonstop=true ;;
+esac
+
+CLASSPATH=$APP_HOME/gradle/wrapper/gradle-wrapper.jar
+
+
+# Determine the Java command to use to start the JVM.
+if [ -n "$JAVA_HOME" ] ; then
+ if [ -x "$JAVA_HOME/jre/sh/java" ] ; then
+ # IBM's JDK on AIX uses strange locations for the executables
+ JAVACMD=$JAVA_HOME/jre/sh/java
+ else
+ JAVACMD=$JAVA_HOME/bin/java
+ fi
+ if [ ! -x "$JAVACMD" ] ; then
+ die "ERROR: JAVA_HOME is set to an invalid directory: $JAVA_HOME
+
+Please set the JAVA_HOME variable in your environment to match the
+location of your Java installation."
+ fi
+else
+ JAVACMD=java
+ which java >/dev/null 2>&1 || die "ERROR: JAVA_HOME is not set and no 'java' command could be found in your PATH.
+
+Please set the JAVA_HOME variable in your environment to match the
+location of your Java installation."
+fi
+
+# Increase the maximum file descriptors if we can.
+if ! "$cygwin" && ! "$darwin" && ! "$nonstop" ; then
+ case $MAX_FD in #(
+ max*)
+ MAX_FD=$( ulimit -H -n ) ||
+ warn "Could not query maximum file descriptor limit"
+ esac
+ case $MAX_FD in #(
+ '' | soft) :;; #(
+ *)
+ ulimit -n "$MAX_FD" ||
+ warn "Could not set maximum file descriptor limit to $MAX_FD"
+ esac
+fi
+
+# Collect all arguments for the java command, stacking in reverse order:
+# * args from the command line
+# * the main class name
+# * -classpath
+# * -D...appname settings
+# * --module-path (only if needed)
+# * DEFAULT_JVM_OPTS, JAVA_OPTS, and GRADLE_OPTS environment variables.
+
+# For Cygwin or MSYS, switch paths to Windows format before running java
+if "$cygwin" || "$msys" ; then
+ APP_HOME=$( cygpath --path --mixed "$APP_HOME" )
+ CLASSPATH=$( cygpath --path --mixed "$CLASSPATH" )
+
+ JAVACMD=$( cygpath --unix "$JAVACMD" )
+
+ # Now convert the arguments - kludge to limit ourselves to /bin/sh
+ for arg do
+ if
+ case $arg in #(
+ -*) false ;; # don't mess with options #(
+ /?*) t=${arg#/} t=/${t%%/*} # looks like a POSIX filepath
+ [ -e "$t" ] ;; #(
+ *) false ;;
+ esac
+ then
+ arg=$( cygpath --path --ignore --mixed "$arg" )
+ fi
+ # Roll the args list around exactly as many times as the number of
+ # args, so each arg winds up back in the position where it started, but
+ # possibly modified.
+ #
+ # NB: a `for` loop captures its iteration list before it begins, so
+ # changing the positional parameters here affects neither the number of
+ # iterations, nor the values presented in `arg`.
+ shift # remove old arg
+ set -- "$@" "$arg" # push replacement arg
+ done
+fi
+
+# Collect all arguments for the java command;
+# * $DEFAULT_JVM_OPTS, $JAVA_OPTS, and $GRADLE_OPTS can contain fragments of
+# shell script including quotes and variable substitutions, so put them in
+# double quotes to make sure that they get re-expanded; and
+# * put everything else in single quotes, so that it's not re-expanded.
+
+set -- \
+ "-Dorg.gradle.appname=$APP_BASE_NAME" \
+ -classpath "$CLASSPATH" \
+ org.gradle.wrapper.GradleWrapperMain \
+ "$@"
+
+# Use "xargs" to parse quoted args.
+#
+# With -n1 it outputs one arg per line, with the quotes and backslashes removed.
+#
+# In Bash we could simply go:
+#
+# readarray ARGS < <( xargs -n1 <<<"$var" ) &&
+# set -- "${ARGS[@]}" "$@"
+#
+# but POSIX shell has neither arrays nor command substitution, so instead we
+# post-process each arg (as a line of input to sed) to backslash-escape any
+# character that might be a shell metacharacter, then use eval to reverse
+# that process (while maintaining the separation between arguments), and wrap
+# the whole thing up as a single "set" statement.
+#
+# This will of course break if any of these variables contains a newline or
+# an unmatched quote.
+#
+
+eval "set -- $(
+ printf '%s\n' "$DEFAULT_JVM_OPTS $JAVA_OPTS $GRADLE_OPTS" |
+ xargs -n1 |
+ sed ' s~[^-[:alnum:]+,./:=@_]~\\&~g; ' |
+ tr '\n' ' '
+ )" '"$@"'
+
+exec "$JAVACMD" "$@"
diff --git a/foreign/java/gradlew.bat b/foreign/java/gradlew.bat
new file mode 100644
index 00000000..107acd32
--- /dev/null
+++ b/foreign/java/gradlew.bat
@@ -0,0 +1,89 @@
+@rem
+@rem Copyright 2015 the original author or authors.
+@rem
+@rem Licensed under the Apache License, Version 2.0 (the "License");
+@rem you may not use this file except in compliance with the License.
+@rem You may obtain a copy of the License at
+@rem
+@rem https://www.apache.org/licenses/LICENSE-2.0
+@rem
+@rem Unless required by applicable law or agreed to in writing, software
+@rem distributed under the License is distributed on an "AS IS" BASIS,
+@rem WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+@rem See the License for the specific language governing permissions and
+@rem limitations under the License.
+@rem
+
+@if "%DEBUG%" == "" @echo off
+@rem ##########################################################################
+@rem
+@rem Gradle startup script for Windows
+@rem
+@rem ##########################################################################
+
+@rem Set local scope for the variables with windows NT shell
+if "%OS%"=="Windows_NT" setlocal
+
+set DIRNAME=%~dp0
+if "%DIRNAME%" == "" set DIRNAME=.
+set APP_BASE_NAME=%~n0
+set APP_HOME=%DIRNAME%
+
+@rem Resolve any "." and ".." in APP_HOME to make it shorter.
+for %%i in ("%APP_HOME%") do set APP_HOME=%%~fi
+
+@rem Add default JVM options here. You can also use JAVA_OPTS and GRADLE_OPTS to pass JVM options to this script.
+set DEFAULT_JVM_OPTS="-Xmx64m" "-Xms64m"
+
+@rem Find java.exe
+if defined JAVA_HOME goto findJavaFromJavaHome
+
+set JAVA_EXE=java.exe
+%JAVA_EXE% -version >NUL 2>&1
+if "%ERRORLEVEL%" == "0" goto execute
+
+echo.
+echo ERROR: JAVA_HOME is not set and no 'java' command could be found in your PATH.
+echo.
+echo Please set the JAVA_HOME variable in your environment to match the
+echo location of your Java installation.
+
+goto fail
+
+:findJavaFromJavaHome
+set JAVA_HOME=%JAVA_HOME:"=%
+set JAVA_EXE=%JAVA_HOME%/bin/java.exe
+
+if exist "%JAVA_EXE%" goto execute
+
+echo.
+echo ERROR: JAVA_HOME is set to an invalid directory: %JAVA_HOME%
+echo.
+echo Please set the JAVA_HOME variable in your environment to match the
+echo location of your Java installation.
+
+goto fail
+
+:execute
+@rem Setup the command line
+
+set CLASSPATH=%APP_HOME%\gradle\wrapper\gradle-wrapper.jar
+
+
+@rem Execute Gradle
+"%JAVA_EXE%" %DEFAULT_JVM_OPTS% %JAVA_OPTS% %GRADLE_OPTS% "-Dorg.gradle.appname=%APP_BASE_NAME%" -classpath "%CLASSPATH%" org.gradle.wrapper.GradleWrapperMain %*
+
+:end
+@rem End local scope for the variables with windows NT shell
+if "%ERRORLEVEL%"=="0" goto mainEnd
+
+:fail
+rem Set variable GRADLE_EXIT_CONSOLE if you need the _script_ return code instead of
+rem the _cmd.exe /c_ return code!
+if not "" == "%GRADLE_EXIT_CONSOLE%" exit 1
+exit /b 1
+
+:mainEnd
+if "%OS%"=="Windows_NT" endlocal
+
+:omega
diff --git a/foreign/java/java-sdk/build.gradle.kts b/foreign/java/java-sdk/build.gradle.kts
new file mode 100644
index 00000000..a1e5932d
--- /dev/null
+++ b/foreign/java/java-sdk/build.gradle.kts
@@ -0,0 +1,96 @@
+plugins {
+ id("java-library")
+ id("maven-publish")
+ id("org.jreleaser") version ("1.14.0")
+}
+
+group = "rs.iggy"
+version = "0.1.1"
+
+repositories {
+ mavenCentral()
+}
+
+java {
+ withJavadocJar()
+ withSourcesJar()
+}
+
+dependencies {
+ implementation("org.apache.httpcomponents.client5:httpclient5:5.4")
+ implementation("com.fasterxml.jackson.core:jackson-databind:2.18.0")
+ implementation("com.fasterxml.jackson.datatype:jackson-datatype-jdk8:2.18.0")
+ implementation("org.apache.commons:commons-lang3:3.17.0")
+ implementation("org.slf4j:slf4j-api:2.0.16")
+ implementation("com.google.code.findbugs:jsr305:3.0.2")
+ implementation("io.projectreactor:reactor-core:3.6.11")
+ implementation("io.projectreactor.netty:reactor-netty-core:1.1.23")
+ testImplementation("org.testcontainers:testcontainers:1.20.3")
+ testImplementation("org.testcontainers:junit-jupiter:1.20.3")
+ testImplementation(platform("org.junit:junit-bom:5.11.3"))
+ testImplementation("org.junit.jupiter:junit-jupiter")
+ testImplementation("org.assertj:assertj-core:3.26.3")
+ testRuntimeOnly("ch.qos.logback:logback-classic:1.5.11")
+}
+
+tasks.withType<Test> {
+ useJUnitPlatform()
+}
+
+publishing {
+ publications {
+ create<MavenPublication>("maven") {
+ artifactId = "iggy-java-sdk"
+
+ from(components["java"])
+
+ pom {
+ name = "Apache Iggy Java Client SDK"
+ description = "Official Java client SDK for Apache Iggy message streaming"
+ url = "https://github.com/iggy-rs/iggy-java-client"
+ licenses {
+ license {
+ name = "Apache License, Version 2.0"
+ url = "https://github.com/iggy-rs/iggy-java-client/blob/main/LICENSE"
+ }
+ }
+ developers {
+ developer {
+ id = "mmodzelewski"
+ name = "Maciej Modzelewski"
+ email = "maciej@modzelewski.dev"
+ }
+ }
+ scm {
+ url = "https://github.com/iggy-rs/iggy-java-client"
+ connection = "scm:git:git://github.com/iggy-rs/iggy-java-client.git"
+ developerConnection = "scm:git:git://github.com/iggy-rs/iggy-java-client.git"
+ }
+ }
+ }
+ }
+
+ repositories {
+ maven {
+ url = uri(layout.buildDirectory.dir("staging-deploy"))
+ }
+ }
+}
+
+jreleaser {
+ signing {
+ setActive("ALWAYS")
+ armored = true
+ }
+ deploy {
+ maven {
+ mavenCentral {
+ create("sonatype") {
+ setActive("ALWAYS")
+ url = "https://central.sonatype.com/api/v1/publisher"
+ stagingRepository("build/staging-deploy")
+ }
+ }
+ }
+ }
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/Iggy.java b/foreign/java/java-sdk/src/main/java/rs/iggy/Iggy.java
new file mode 100644
index 00000000..84d10bfc
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/Iggy.java
@@ -0,0 +1,14 @@
+package rs.iggy;
+
+import rs.iggy.clients.blocking.IggyClientBuilder;
+
+public final class Iggy {
+
+ private Iggy() {
+ }
+
+ public static IggyClientBuilder clientBuilder() {
+ return new IggyClientBuilder();
+ }
+
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/NonNullApi.java b/foreign/java/java-sdk/src/main/java/rs/iggy/NonNullApi.java
new file mode 100644
index 00000000..dfcbe6e5
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/NonNullApi.java
@@ -0,0 +1,13 @@
+package rs.iggy;
+
+import javax.annotation.Nonnull;
+import javax.annotation.meta.TypeQualifierDefault;
+import java.lang.annotation.*;
+
+@Target(ElementType.PACKAGE)
+@Retention(RetentionPolicy.RUNTIME)
+@Documented
+@Nonnull
+@TypeQualifierDefault({ElementType.METHOD, ElementType.PARAMETER})
+public @interface NonNullApi {
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/ConsumerGroupsClient.java b/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/ConsumerGroupsClient.java
new file mode 100644
index 00000000..09b01440
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/ConsumerGroupsClient.java
@@ -0,0 +1,49 @@
+package rs.iggy.clients.blocking;
+
+import rs.iggy.consumergroup.ConsumerGroup;
+import rs.iggy.consumergroup.ConsumerGroupDetails;
+import rs.iggy.identifier.ConsumerId;
+import rs.iggy.identifier.StreamId;
+import rs.iggy.identifier.TopicId;
+import java.util.List;
+import java.util.Optional;
+
+public interface ConsumerGroupsClient {
+
+ default Optional<ConsumerGroupDetails> getConsumerGroup(Long streamId, Long topicId, Long groupId) {
+ return getConsumerGroup(StreamId.of(streamId), TopicId.of(topicId), ConsumerId.of(groupId));
+ }
+
+ Optional<ConsumerGroupDetails> getConsumerGroup(StreamId streamId, TopicId topicId, ConsumerId groupId);
+
+ default List<ConsumerGroup> getConsumerGroups(Long streamId, Long topicId) {
+ return getConsumerGroups(StreamId.of(streamId), TopicId.of(topicId));
+ }
+
+ List<ConsumerGroup> getConsumerGroups(StreamId streamId, TopicId topicId);
+
+ default ConsumerGroupDetails createConsumerGroup(Long streamId, Long topicId, Optional<Long> groupId, String name) {
+ return createConsumerGroup(StreamId.of(streamId), TopicId.of(topicId), groupId, name);
+ }
+
+ ConsumerGroupDetails createConsumerGroup(StreamId streamId, TopicId topicId, Optional<Long> groupId, String name);
+
+ default void deleteConsumerGroup(Long streamId, Long topicId, Long groupId) {
+ deleteConsumerGroup(StreamId.of(streamId), TopicId.of(topicId), ConsumerId.of(groupId));
+ }
+
+ void deleteConsumerGroup(StreamId streamId, TopicId topicId, ConsumerId groupId);
+
+ default void joinConsumerGroup(Long streamId, Long topicId, Long groupId) {
+ joinConsumerGroup(StreamId.of(streamId), TopicId.of(topicId), ConsumerId.of(groupId));
+ }
+
+ void joinConsumerGroup(StreamId streamId, TopicId topicId, ConsumerId groupId);
+
+ default void leaveConsumerGroup(Long streamId, Long topicId, Long groupId) {
+ leaveConsumerGroup(StreamId.of(streamId), TopicId.of(topicId), ConsumerId.of(groupId));
+ }
+
+ void leaveConsumerGroup(StreamId streamId, TopicId topicId, ConsumerId groupId);
+
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/ConsumerOffsetsClient.java b/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/ConsumerOffsetsClient.java
new file mode 100644
index 00000000..3fa53acc
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/ConsumerOffsetsClient.java
@@ -0,0 +1,24 @@
+package rs.iggy.clients.blocking;
+
+import rs.iggy.consumergroup.Consumer;
+import rs.iggy.consumeroffset.ConsumerOffsetInfo;
+import rs.iggy.identifier.StreamId;
+import rs.iggy.identifier.TopicId;
+import java.math.BigInteger;
+import java.util.Optional;
+
+public interface ConsumerOffsetsClient {
+
+ default void storeConsumerOffset(Long streamId, Long topicId, Optional<Long> partitionId, Long consumerId, BigInteger offset) {
+ storeConsumerOffset(StreamId.of(streamId), TopicId.of(topicId), partitionId, Consumer.of(consumerId), offset);
+ }
+
+ void storeConsumerOffset(StreamId streamId, TopicId topicId, Optional<Long> partitionId, Consumer consumer, BigInteger offset);
+
+ default Optional<ConsumerOffsetInfo> getConsumerOffset(Long streamId, Long topicId, Optional<Long> partitionId, Long consumerId) {
+ return getConsumerOffset(StreamId.of(streamId), TopicId.of(topicId), partitionId, Consumer.of(consumerId));
+ }
+
+ Optional<ConsumerOffsetInfo> getConsumerOffset(StreamId streamId, TopicId topicId, Optional<Long> partitionId, Consumer consumer);
+
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/IggyBaseClient.java b/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/IggyBaseClient.java
new file mode 100644
index 00000000..3467cfd7
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/IggyBaseClient.java
@@ -0,0 +1,23 @@
+package rs.iggy.clients.blocking;
+
+public interface IggyBaseClient {
+
+ SystemClient system();
+
+ StreamsClient streams();
+
+ UsersClient users();
+
+ TopicsClient topics();
+
+ PartitionsClient partitions();
+
+ ConsumerGroupsClient consumerGroups();
+
+ ConsumerOffsetsClient consumerOffsets();
+
+ MessagesClient messages();
+
+ PersonalAccessTokensClient personalAccessTokens();
+
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/IggyClient.java b/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/IggyClient.java
new file mode 100644
index 00000000..37c5872d
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/IggyClient.java
@@ -0,0 +1,15 @@
+package rs.iggy.clients.blocking;
+
+public class IggyClient {
+
+ private final IggyBaseClient baseClient;
+
+ public IggyClient(IggyBaseClient baseClient) {
+ this.baseClient = baseClient;
+ }
+
+ public IggyBaseClient getBaseClient() {
+ return baseClient;
+ }
+
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/IggyClientBuilder.java b/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/IggyClientBuilder.java
new file mode 100644
index 00000000..247d3635
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/IggyClientBuilder.java
@@ -0,0 +1,19 @@
+package rs.iggy.clients.blocking;
+
+public class IggyClientBuilder {
+
+ private IggyBaseClient client;
+
+ public IggyClientBuilder withBaseClient(IggyBaseClient client) {
+ this.client = client;
+ return this;
+ }
+
+ public IggyClient build() {
+ if (client == null) {
+ throw new IllegalArgumentException("Base client not provided");
+ }
+ return new IggyClient(client);
+ }
+
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/MessagesClient.java b/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/MessagesClient.java
new file mode 100644
index 00000000..f4faee87
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/MessagesClient.java
@@ -0,0 +1,28 @@
+package rs.iggy.clients.blocking;
+
+import rs.iggy.consumergroup.Consumer;
+import rs.iggy.identifier.StreamId;
+import rs.iggy.identifier.TopicId;
+import rs.iggy.message.Message;
+import rs.iggy.message.Partitioning;
+import rs.iggy.message.PolledMessages;
+import rs.iggy.message.PollingStrategy;
+import java.util.List;
+import java.util.Optional;
+
+public interface MessagesClient {
+
+ default PolledMessages pollMessages(Long streamId, Long topicId, Optional<Long> partitionId, Long consumerId, PollingStrategy strategy, Long count, boolean autoCommit) {
+ return pollMessages(StreamId.of(streamId), TopicId.of(topicId), partitionId, Consumer.of(consumerId),
+ strategy, count, autoCommit);
+ }
+
+ PolledMessages pollMessages(StreamId streamId, TopicId topicId, Optional<Long> partitionId, Consumer consumer, PollingStrategy strategy, Long count, boolean autoCommit);
+
+ default void sendMessages(Long streamId, Long topicId, Partitioning partitioning, List<Message> messages) {
+ sendMessages(StreamId.of(streamId), TopicId.of(topicId), partitioning, messages);
+ }
+
+ void sendMessages(StreamId streamId, TopicId topicId, Partitioning partitioning, List<Message> messages);
+
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/PartitionsClient.java b/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/PartitionsClient.java
new file mode 100644
index 00000000..4a5386e7
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/PartitionsClient.java
@@ -0,0 +1,20 @@
+package rs.iggy.clients.blocking;
+
+import rs.iggy.identifier.StreamId;
+import rs.iggy.identifier.TopicId;
+
+public interface PartitionsClient {
+
+ default void createPartitions(Long streamId, Long topicId, Long partitionsCount) {
+ createPartitions(StreamId.of(streamId), TopicId.of(topicId), partitionsCount);
+ }
+
+ void createPartitions(StreamId streamId, TopicId topicId, Long partitionsCount);
+
+ default void deletePartitions(Long streamId, Long topicId, Long partitionsCount) {
+ deletePartitions(StreamId.of(streamId), TopicId.of(topicId), partitionsCount);
+ }
+
+ void deletePartitions(StreamId streamId, TopicId topicId, Long partitionsCount);
+
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/PersonalAccessTokensClient.java b/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/PersonalAccessTokensClient.java
new file mode 100644
index 00000000..bcca66c8
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/PersonalAccessTokensClient.java
@@ -0,0 +1,19 @@
+package rs.iggy.clients.blocking;
+
+import rs.iggy.personalaccesstoken.PersonalAccessTokenInfo;
+import rs.iggy.personalaccesstoken.RawPersonalAccessToken;
+import rs.iggy.user.IdentityInfo;
+import java.math.BigInteger;
+import java.util.List;
+
+public interface PersonalAccessTokensClient {
+
+ RawPersonalAccessToken createPersonalAccessToken(String name, BigInteger expiry);
+
+ List<PersonalAccessTokenInfo> getPersonalAccessTokens();
+
+ void deletePersonalAccessToken(String name);
+
+ IdentityInfo loginWithPersonalAccessToken(String token);
+
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/StreamsClient.java b/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/StreamsClient.java
new file mode 100644
index 00000000..2aa3e871
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/StreamsClient.java
@@ -0,0 +1,33 @@
+package rs.iggy.clients.blocking;
+
+import rs.iggy.identifier.StreamId;
+import rs.iggy.stream.StreamBase;
+import rs.iggy.stream.StreamDetails;
+import java.util.List;
+import java.util.Optional;
+
+public interface StreamsClient {
+
+ default Optional<StreamDetails> getStream(Long streamId) {
+ return getStream(StreamId.of(streamId));
+ }
+
+ Optional<StreamDetails> getStream(StreamId streamId);
+
+ List<StreamBase> getStreams();
+
+ StreamDetails createStream(Optional<Long> streamId, String name);
+
+ default void updateStream(Long streamId, String name) {
+ updateStream(StreamId.of(streamId), name);
+ }
+
+ void updateStream(StreamId streamId, String name);
+
+ default void deleteStream(Long streamId) {
+ deleteStream(StreamId.of(streamId));
+ }
+
+ void deleteStream(StreamId streamId);
+
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/SystemClient.java b/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/SystemClient.java
new file mode 100644
index 00000000..27b6c6b4
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/SystemClient.java
@@ -0,0 +1,20 @@
+package rs.iggy.clients.blocking;
+
+import rs.iggy.system.ClientInfo;
+import rs.iggy.system.ClientInfoDetails;
+import rs.iggy.system.Stats;
+import java.util.List;
+
+public interface SystemClient {
+
+ Stats getStats();
+
+ ClientInfoDetails getMe();
+
+ ClientInfoDetails getClient(Long clientId);
+
+ List<ClientInfo> getClients();
+
+ String ping();
+
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/TopicsClient.java b/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/TopicsClient.java
new file mode 100644
index 00000000..96fec30a
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/TopicsClient.java
@@ -0,0 +1,83 @@
+package rs.iggy.clients.blocking;
+
+import rs.iggy.identifier.StreamId;
+import rs.iggy.identifier.TopicId;
+import rs.iggy.topic.CompressionAlgorithm;
+import rs.iggy.topic.Topic;
+import rs.iggy.topic.TopicDetails;
+import java.math.BigInteger;
+import java.util.List;
+import java.util.Optional;
+
+public interface TopicsClient {
+
+ default Optional<TopicDetails> getTopic(Long streamId, Long topicId) {
+ return getTopic(StreamId.of(streamId), TopicId.of(topicId));
+ }
+
+ Optional<TopicDetails> getTopic(StreamId streamId, TopicId topicId);
+
+ default List<Topic> getTopics(Long streamId) {
+ return getTopics(StreamId.of(streamId));
+ }
+
+ List<Topic> getTopics(StreamId streamId);
+
+ default TopicDetails createTopic(Long streamId,
+ Optional<Long> topicId,
+ Long partitionsCount,
+ CompressionAlgorithm compressionAlgorithm,
+ BigInteger messageExpiry,
+ BigInteger maxTopicSize,
+ Optional<Short> replicationFactor,
+ String name) {
+ return createTopic(StreamId.of(streamId),
+ topicId,
+ partitionsCount,
+ compressionAlgorithm,
+ messageExpiry,
+ maxTopicSize,
+ replicationFactor,
+ name);
+ }
+
+ TopicDetails createTopic(StreamId streamId,
+ Optional<Long> topicId,
+ Long partitionsCount,
+ CompressionAlgorithm compressionAlgorithm,
+ BigInteger messageExpiry,
+ BigInteger maxTopicSize,
+ Optional<Short> replicationFactor,
+ String name);
+
+ default void updateTopic(Long streamId,
+ Long topicId,
+ CompressionAlgorithm compressionAlgorithm,
+ BigInteger messageExpiry,
+ BigInteger maxTopicSize,
+ Optional<Short> replicationFactor,
+ String name) {
+ updateTopic(StreamId.of(streamId),
+ TopicId.of(topicId),
+ compressionAlgorithm,
+ messageExpiry,
+ maxTopicSize,
+ replicationFactor,
+ name);
+ }
+
+ void updateTopic(StreamId streamId,
+ TopicId topicId,
+ CompressionAlgorithm compressionAlgorithm,
+ BigInteger messageExpiry,
+ BigInteger maxTopicSize,
+ Optional<Short> replicationFactor,
+ String name);
+
+ default void deleteTopic(Long streamId, Long topicId) {
+ deleteTopic(StreamId.of(streamId), TopicId.of(topicId));
+ }
+
+ void deleteTopic(StreamId streamId, TopicId topicId);
+
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/UsersClient.java b/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/UsersClient.java
new file mode 100644
index 00000000..085c44a2
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/UsersClient.java
@@ -0,0 +1,48 @@
+package rs.iggy.clients.blocking;
+
+import rs.iggy.identifier.UserId;
+import rs.iggy.user.*;
+import java.util.List;
+import java.util.Optional;
+
+public interface UsersClient {
+
+ default Optional<UserInfoDetails> getUser(Long userId) {
+ return getUser(UserId.of(userId));
+ }
+
+ Optional<UserInfoDetails> getUser(UserId userId);
+
+ List<UserInfo> getUsers();
+
+ UserInfoDetails createUser(String username, String password, UserStatus status, Optional<Permissions> permissions);
+
+ default void deleteUser(Long userId) {
+ deleteUser(UserId.of(userId));
+ }
+
+ void deleteUser(UserId userId);
+
+ default void updateUser(Long userId, Optional<String> username, Optional<UserStatus> status) {
+ updateUser(UserId.of(userId), username, status);
+ }
+
+ void updateUser(UserId userId, Optional<String> username, Optional<UserStatus> status);
+
+ default void updatePermissions(Long userId, Optional<Permissions> permissions) {
+ updatePermissions(UserId.of(userId), permissions);
+ }
+
+ void updatePermissions(UserId userId, Optional<Permissions> permissions);
+
+ default void changePassword(Long userId, String currentPassword, String newPassword) {
+ changePassword(UserId.of(userId), currentPassword, newPassword);
+ }
+
+ void changePassword(UserId userId, String currentPassword, String newPassword);
+
+ IdentityInfo login(String username, String password);
+
+ void logout();
+
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/http/ConsumerGroupsHttpClient.java b/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/http/ConsumerGroupsHttpClient.java
new file mode 100644
index 00000000..a4fc1086
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/http/ConsumerGroupsHttpClient.java
@@ -0,0 +1,65 @@
+package rs.iggy.clients.blocking.http;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import rs.iggy.clients.blocking.ConsumerGroupsClient;
+import rs.iggy.consumergroup.ConsumerGroup;
+import rs.iggy.consumergroup.ConsumerGroupDetails;
+import rs.iggy.identifier.ConsumerId;
+import rs.iggy.identifier.StreamId;
+import rs.iggy.identifier.TopicId;
+import java.util.List;
+import java.util.Optional;
+
+class ConsumerGroupsHttpClient implements ConsumerGroupsClient {
+
+ private final InternalHttpClient httpClient;
+
+ public ConsumerGroupsHttpClient(InternalHttpClient httpClient) {
+ this.httpClient = httpClient;
+ }
+
+ @Override
+ public Optional<ConsumerGroupDetails> getConsumerGroup(StreamId streamId, TopicId topicId, ConsumerId groupId) {
+ var request = httpClient.prepareGetRequest(path(streamId, topicId) + "/" + groupId);
+ return httpClient.executeWithOptionalResponse(request, ConsumerGroupDetails.class);
+ }
+
+ @Override
+ public List<ConsumerGroup> getConsumerGroups(StreamId streamId, TopicId topicId) {
+ var request = httpClient.prepareGetRequest(path(streamId, topicId));
+ return httpClient.execute(request, new TypeReference<>() {
+ });
+ }
+
+ @Override
+ public ConsumerGroupDetails createConsumerGroup(StreamId streamId, TopicId topicId, Optional<Long> groupId, String name) {
+ var request = httpClient.preparePostRequest(path(streamId, topicId),
+ new CreateConsumerGroup(groupId, name));
+ return httpClient.execute(request, new TypeReference<>() {
+ });
+ }
+
+ @Override
+ public void deleteConsumerGroup(StreamId streamId, TopicId topicId, ConsumerId groupId) {
+ var request = httpClient.prepareDeleteRequest(path(streamId, topicId) + "/" + groupId);
+ httpClient.execute(request);
+ }
+
+ @Override
+ public void joinConsumerGroup(StreamId streamId, TopicId topicId, ConsumerId groupId) {
+ throw new UnsupportedOperationException("Method not available in HTTP client");
+ }
+
+ @Override
+ public void leaveConsumerGroup(StreamId streamId, TopicId topicId, ConsumerId groupId) {
+ throw new UnsupportedOperationException("Method not available in HTTP client");
+ }
+
+ private static String path(StreamId streamId, TopicId topicId) {
+ return "/streams/" + streamId + "/topics/" + topicId + "/consumer-groups";
+ }
+
+ private record CreateConsumerGroup(Optional<Long> groupId, String name) {
+ }
+
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/http/ConsumerOffsetsHttpClient.java b/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/http/ConsumerOffsetsHttpClient.java
new file mode 100644
index 00000000..843f535b
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/http/ConsumerOffsetsHttpClient.java
@@ -0,0 +1,42 @@
+package rs.iggy.clients.blocking.http;
+
+import org.apache.hc.core5.http.message.BasicNameValuePair;
+import rs.iggy.clients.blocking.ConsumerOffsetsClient;
+import rs.iggy.consumergroup.Consumer;
+import rs.iggy.consumeroffset.ConsumerOffsetInfo;
+import rs.iggy.identifier.StreamId;
+import rs.iggy.identifier.TopicId;
+import java.math.BigInteger;
+import java.util.Optional;
+
+class ConsumerOffsetsHttpClient implements ConsumerOffsetsClient {
+
+ private static final String DEFAULT_PARTITION_ID = "1";
+ private final InternalHttpClient httpClient;
+
+ public ConsumerOffsetsHttpClient(InternalHttpClient httpClient) {
+ this.httpClient = httpClient;
+ }
+
+ @Override
+ public void storeConsumerOffset(StreamId streamId, TopicId topicId, Optional<Long> partitionId, Consumer consumer, BigInteger offset) {
+ var request = httpClient.preparePutRequest(path(streamId, topicId),
+ new StoreConsumerOffset(consumer.id().toString(), partitionId, offset));
+ httpClient.execute(request);
+ }
+
+ @Override
+ public Optional<ConsumerOffsetInfo> getConsumerOffset(StreamId streamId, TopicId topicId, Optional<Long> partitionId, Consumer consumer) {
+ var request = httpClient.prepareGetRequest(path(streamId, topicId),
+ new BasicNameValuePair("consumer_id", consumer.id().toString()),
+ new BasicNameValuePair("partition_id", partitionId.map(Object::toString).orElse(DEFAULT_PARTITION_ID)));
+ return httpClient.executeWithOptionalResponse(request, ConsumerOffsetInfo.class);
+ }
+
+ private static String path(StreamId streamId, TopicId topicId) {
+ return "/streams/" + streamId + "/topics/" + topicId + "/consumer-offsets";
+ }
+
+ private record StoreConsumerOffset(String consumerId, Optional<Long> partitionId, BigInteger offset) {
+ }
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/http/IggyHttpClient.java b/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/http/IggyHttpClient.java
new file mode 100644
index 00000000..be95757d
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/http/IggyHttpClient.java
@@ -0,0 +1,75 @@
+package rs.iggy.clients.blocking.http;
+
+import rs.iggy.clients.blocking.*;
+
+public class IggyHttpClient implements IggyBaseClient {
+
+ private final SystemHttpClient systemClient;
+ private final StreamsHttpClient streamsClient;
+ private final UsersHttpClient usersClient;
+ private final TopicsHttpClient topicsClient;
+ private final PartitionsHttpClient partitionsClient;
+ private final ConsumerGroupsHttpClient consumerGroupsClient;
+ private final ConsumerOffsetsHttpClient consumerOffsetsClient;
+ private final MessagesHttpClient messagesClient;
+ private final PersonalAccessTokensHttpClient personalAccessTokensHttpClient;
+
+ public IggyHttpClient(String url) {
+ InternalHttpClient httpClient = new InternalHttpClient(url);
+ systemClient = new SystemHttpClient(httpClient);
+ streamsClient = new StreamsHttpClient(httpClient);
+ usersClient = new UsersHttpClient(httpClient);
+ topicsClient = new TopicsHttpClient(httpClient);
+ partitionsClient = new PartitionsHttpClient(httpClient);
+ consumerGroupsClient = new ConsumerGroupsHttpClient(httpClient);
+ consumerOffsetsClient = new ConsumerOffsetsHttpClient(httpClient);
+ messagesClient = new MessagesHttpClient(httpClient);
+ personalAccessTokensHttpClient = new PersonalAccessTokensHttpClient(httpClient);
+ }
+
+ @Override
+ public SystemClient system() {
+ return systemClient;
+ }
+
+ @Override
+ public StreamsClient streams() {
+ return streamsClient;
+ }
+
+ @Override
+ public UsersClient users() {
+ return usersClient;
+ }
+
+ @Override
+ public TopicsClient topics() {
+ return topicsClient;
+ }
+
+ @Override
+ public PartitionsClient partitions() {
+ return partitionsClient;
+ }
+
+ @Override
+ public ConsumerGroupsClient consumerGroups() {
+ return consumerGroupsClient;
+ }
+
+ @Override
+ public ConsumerOffsetsClient consumerOffsets() {
+ return consumerOffsetsClient;
+ }
+
+ @Override
+ public MessagesClient messages() {
+ return messagesClient;
+ }
+
+ @Override
+ public PersonalAccessTokensClient personalAccessTokens() {
+ return personalAccessTokensHttpClient;
+ }
+
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/http/InternalHttpClient.java b/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/http/InternalHttpClient.java
new file mode 100644
index 00000000..e4830a46
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/http/InternalHttpClient.java
@@ -0,0 +1,143 @@
+package rs.iggy.clients.blocking.http;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.JavaType;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.hc.client5.http.impl.classic.HttpClients;
+import org.apache.hc.core5.http.ClassicHttpRequest;
+import org.apache.hc.core5.http.ClassicHttpResponse;
+import org.apache.hc.core5.http.NameValuePair;
+import org.apache.hc.core5.http.io.HttpClientResponseHandler;
+import org.apache.hc.core5.http.io.support.ClassicRequestBuilder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import rs.iggy.clients.blocking.http.error.IggyHttpError;
+import rs.iggy.clients.blocking.http.error.IggyHttpException;
+import java.io.IOException;
+import java.util.Optional;
+
+final class InternalHttpClient {
+
+ private static final Logger log = LoggerFactory.getLogger(InternalHttpClient.class);
+
+ private static final String AUTHORIZATION = "Authorization";
+ private final String url;
+ private final ObjectMapper objectMapper = ObjectMapperFactory.getInstance();
+ private Optional<String> token = Optional.empty();
+
+ InternalHttpClient(String url) {
+ this.url = url;
+ }
+
+ void setToken(Optional<String> token) {
+ this.token = token;
+ }
+
+ <T> T execute(ClassicHttpRequest request, Class<T> clazz) {
+ return execute(request, objectMapper.constructType(clazz));
+ }
+
+ <T> T execute(ClassicHttpRequest request, TypeReference<T> typeReference) {
+ return execute(request, objectMapper.constructType(typeReference));
+ }
+
+ private <T> T execute(ClassicHttpRequest request, JavaType type) {
+ return executeRequest(request, response -> handleTypedResponse(response, type));
+ }
+
+ public <T> Optional<T> executeWithOptionalResponse(ClassicHttpRequest request, Class<T> clazz) {
+ return executeWithOptionalResponse(request, objectMapper.constructType(clazz));
+ }
+
+ private <T> Optional<T> executeWithOptionalResponse(ClassicHttpRequest request, JavaType type) {
+ return executeRequest(request, response -> {
+ if (response.getCode() == 404) {
+ return Optional.empty();
+ }
+ return Optional.of(handleTypedResponse(response, type));
+ });
+ }
+
+ String executeWithStringResponse(ClassicHttpRequest request) {
+ return executeRequest(request, response -> {
+ handleErrorResponse(response);
+ return new String(response.getEntity().getContent().readAllBytes());
+ });
+ }
+
+ void execute(ClassicHttpRequest request) {
+ executeRequest(request, response -> {
+ handleErrorResponse(response);
+ return "";
+ });
+ }
+
+ private <T> T executeRequest(ClassicHttpRequest request, HttpClientResponseHandler<T> responseHandler) {
+ try (var client = HttpClients.createDefault()) {
+ return client.execute(request, responseHandler);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ ClassicHttpRequest prepareGetRequest(String path, NameValuePair... params) {
+ return ClassicRequestBuilder.get(url + path)
+ .setHeader(AUTHORIZATION, getBearerToken())
+ .addParameters(params)
+ .build();
+ }
+
+ ClassicHttpRequest preparePostRequest(String path, Object body) {
+ var builder = ClassicRequestBuilder.post(url + path)
+ .setHeader(AUTHORIZATION, getBearerToken());
+ return addRequestBody(builder, body);
+ }
+
+ ClassicHttpRequest preparePutRequest(String path, Object body) {
+ var builder = ClassicRequestBuilder.put(url + path)
+ .setHeader(AUTHORIZATION, getBearerToken());
+ return addRequestBody(builder, body);
+ }
+
+ ClassicHttpRequest prepareDeleteRequest(String path, NameValuePair... params) {
+ return ClassicRequestBuilder.delete(url + path)
+ .setHeader(AUTHORIZATION, getBearerToken())
+ .addParameters(params)
+ .build();
+ }
+
+ private ClassicHttpRequest addRequestBody(ClassicRequestBuilder requestBuilder, Object body) {
+ try {
+ var encodedBody = objectMapper.writeValueAsString(body);
+ log.debug("Request body: {}", encodedBody);
+ return requestBuilder
+ .setHeader("Content-Type", "application/json")
+ .setEntity(encodedBody)
+ .build();
+ } catch (JsonProcessingException e) {
+ throw new RuntimeException(e);
+ }
+
+ }
+
+ private <T> T handleTypedResponse(ClassicHttpResponse response, JavaType type) throws IOException {
+ handleErrorResponse(response);
+ return objectMapper.readValue(response.getEntity().getContent(), type);
+ }
+
+ private void handleErrorResponse(ClassicHttpResponse response) throws IOException {
+ if (!isSuccessful(response.getCode())) {
+ var error = objectMapper.readValue(response.getEntity().getContent(), IggyHttpError.class);
+ throw new IggyHttpException(error);
+ }
+ }
+
+ private String getBearerToken() {
+ return token.map(t -> "Bearer " + t).orElse("");
+ }
+
+ private static boolean isSuccessful(int statusCode) {
+ return statusCode >= 200 && statusCode < 300;
+ }
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/http/MessagesHttpClient.java b/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/http/MessagesHttpClient.java
new file mode 100644
index 00000000..70e89866
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/http/MessagesHttpClient.java
@@ -0,0 +1,47 @@
+package rs.iggy.clients.blocking.http;
+
+import org.apache.hc.core5.http.message.BasicNameValuePair;
+import rs.iggy.clients.blocking.MessagesClient;
+import rs.iggy.consumergroup.Consumer;
+import rs.iggy.identifier.StreamId;
+import rs.iggy.identifier.TopicId;
+import rs.iggy.message.Message;
+import rs.iggy.message.Partitioning;
+import rs.iggy.message.PolledMessages;
+import rs.iggy.message.PollingStrategy;
+import java.util.List;
+import java.util.Optional;
+
+class MessagesHttpClient implements MessagesClient {
+
+ private final InternalHttpClient httpClient;
+
+ public MessagesHttpClient(InternalHttpClient httpClient) {
+ this.httpClient = httpClient;
+ }
+
+ @Override
+ public PolledMessages pollMessages(StreamId streamId, TopicId topicId, Optional<Long> partitionId, Consumer consumer, PollingStrategy strategy, Long count, boolean autoCommit) {
+ var request = httpClient.prepareGetRequest(path(streamId, topicId),
+ new BasicNameValuePair("consumer_id", consumer.id().toString()),
+ partitionId.map(id -> new BasicNameValuePair("partition_id", id.toString())).orElse(null),
+ new BasicNameValuePair("strategy_kind", strategy.kind().name()),
+ new BasicNameValuePair("strategy_value", strategy.value().toString()),
+ new BasicNameValuePair("count", count.toString()),
+ new BasicNameValuePair("auto_commit", Boolean.toString(autoCommit)));
+ return httpClient.execute(request, PolledMessages.class);
+ }
+
+ @Override
+ public void sendMessages(StreamId streamId, TopicId topicId, Partitioning partitioning, List<Message> messages) {
+ var request = httpClient.preparePostRequest(path(streamId, topicId), new SendMessages(partitioning, messages));
+ httpClient.execute(request);
+ }
+
+ private static String path(StreamId streamId, TopicId topicId) {
+ return "/streams/" + streamId + "/topics/" + topicId + "/messages";
+ }
+
+ private record SendMessages(Partitioning partitioning, List<Message> messages) {
+ }
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/http/ObjectMapperFactory.java b/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/http/ObjectMapperFactory.java
new file mode 100644
index 00000000..96dd88e8
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/http/ObjectMapperFactory.java
@@ -0,0 +1,28 @@
+package rs.iggy.clients.blocking.http;
+
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.MapperFeature;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.PropertyNamingStrategies;
+import com.fasterxml.jackson.databind.json.JsonMapper;
+import com.fasterxml.jackson.datatype.jdk8.Jdk8Module;
+
+final class ObjectMapperFactory {
+
+ private static final ObjectMapper INSTANCE = JsonMapper.builder()
+ .enable(MapperFeature.ACCEPT_CASE_INSENSITIVE_ENUMS)
+ .configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false)
+ .configure(DeserializationFeature.FAIL_ON_NULL_CREATOR_PROPERTIES, true)
+ .configure(DeserializationFeature.FAIL_ON_NULL_FOR_PRIMITIVES, true)
+ .build()
+ .registerModule(new Jdk8Module())
+ .setPropertyNamingStrategy(PropertyNamingStrategies.SNAKE_CASE);
+
+ private ObjectMapperFactory() {
+ }
+
+ static ObjectMapper getInstance() {
+ return INSTANCE;
+ }
+
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/http/PartitionsHttpClient.java b/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/http/PartitionsHttpClient.java
new file mode 100644
index 00000000..d8999e41
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/http/PartitionsHttpClient.java
@@ -0,0 +1,35 @@
+package rs.iggy.clients.blocking.http;
+
+import org.apache.hc.core5.http.message.BasicNameValuePair;
+import rs.iggy.clients.blocking.PartitionsClient;
+import rs.iggy.identifier.StreamId;
+import rs.iggy.identifier.TopicId;
+
+class PartitionsHttpClient implements PartitionsClient {
+
+ private static final String STREAMS = "/streams";
+ private static final String TOPICS = "/topics";
+ private static final String PARTITIONS = "/partitions";
+ private final InternalHttpClient httpClient;
+
+ public PartitionsHttpClient(InternalHttpClient httpClient) {
+ this.httpClient = httpClient;
+ }
+
+ @Override
+ public void createPartitions(StreamId streamId, TopicId topicId, Long partitionsCount) {
+ var request = httpClient.preparePostRequest(STREAMS + "/" + streamId + TOPICS + "/" + topicId + PARTITIONS,
+ new CreatePartitions(partitionsCount));
+ httpClient.execute(request);
+ }
+
+ @Override
+ public void deletePartitions(StreamId streamId, TopicId topicId, Long partitionsCount) {
+ var request = httpClient.prepareDeleteRequest(STREAMS + "/" + streamId + TOPICS + "/" + topicId + PARTITIONS,
+ new BasicNameValuePair("partitions_count", partitionsCount.toString()));
+ httpClient.execute(request);
+ }
+
+ private record CreatePartitions(Long partitionsCount) {
+ }
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/http/PersonalAccessTokensHttpClient.java b/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/http/PersonalAccessTokensHttpClient.java
new file mode 100644
index 00000000..00287831
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/http/PersonalAccessTokensHttpClient.java
@@ -0,0 +1,57 @@
+package rs.iggy.clients.blocking.http;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import rs.iggy.clients.blocking.PersonalAccessTokensClient;
+import rs.iggy.personalaccesstoken.PersonalAccessTokenInfo;
+import rs.iggy.personalaccesstoken.RawPersonalAccessToken;
+import rs.iggy.user.IdentityInfo;
+import rs.iggy.user.TokenInfo;
+import java.math.BigInteger;
+import java.util.List;
+
+class PersonalAccessTokensHttpClient implements PersonalAccessTokensClient {
+
+ private static final String PERSONAL_ACCESS_TOKENS = "/personal-access-tokens";
+ private final InternalHttpClient httpClient;
+
+ public PersonalAccessTokensHttpClient(InternalHttpClient httpClient) {
+ this.httpClient = httpClient;
+ }
+
+ @Override
+ public RawPersonalAccessToken createPersonalAccessToken(String name, BigInteger expiry) {
+ var request = httpClient.preparePostRequest(PERSONAL_ACCESS_TOKENS,
+ new CreatePersonalAccessToken(name, expiry));
+ return httpClient.execute(request, new TypeReference<>() {
+ });
+ }
+
+ @Override
+ public List<PersonalAccessTokenInfo> getPersonalAccessTokens() {
+ var request = httpClient.prepareGetRequest(PERSONAL_ACCESS_TOKENS);
+ return httpClient.execute(request, new TypeReference<>() {
+ });
+ }
+
+ @Override
+ public void deletePersonalAccessToken(String name) {
+ var request = httpClient.prepareDeleteRequest(PERSONAL_ACCESS_TOKENS + "/" + name);
+ httpClient.execute(request);
+ }
+
+ @Override
+ public IdentityInfo loginWithPersonalAccessToken(String token) {
+ var request = httpClient.preparePostRequest(PERSONAL_ACCESS_TOKENS + "/login",
+ new LoginWithPersonalAccessToken(token));
+ var response = httpClient.execute(request, IdentityInfo.class);
+ httpClient.setToken(response.accessToken().map(TokenInfo::token));
+ return response;
+ }
+
+ record CreatePersonalAccessToken(String name, BigInteger expiry) {
+ }
+
+ record LoginWithPersonalAccessToken(String token) {
+ }
+
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/http/StreamsHttpClient.java b/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/http/StreamsHttpClient.java
new file mode 100644
index 00000000..3baeb49a
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/http/StreamsHttpClient.java
@@ -0,0 +1,57 @@
+package rs.iggy.clients.blocking.http;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import rs.iggy.clients.blocking.StreamsClient;
+import rs.iggy.identifier.StreamId;
+import rs.iggy.stream.StreamBase;
+import rs.iggy.stream.StreamDetails;
+import java.util.List;
+import java.util.Optional;
+
+class StreamsHttpClient implements StreamsClient {
+
+ private static final String STREAMS = "/streams";
+ private final InternalHttpClient httpClient;
+
+ public StreamsHttpClient(InternalHttpClient httpClient) {
+ this.httpClient = httpClient;
+ }
+
+ @Override
+ public Optional<StreamDetails> getStream(StreamId streamId) {
+ var request = httpClient.prepareGetRequest(STREAMS + "/" + streamId);
+ return httpClient.executeWithOptionalResponse(request, StreamDetails.class);
+ }
+
+ @Override
+ public List<StreamBase> getStreams() {
+ var request = httpClient.prepareGetRequest(STREAMS);
+ return httpClient.execute(request, new TypeReference<>() {
+ });
+ }
+
+ @Override
+ public StreamDetails createStream(Optional<Long> streamId, String name) {
+ var request = httpClient.preparePostRequest(STREAMS, new CreateStream(streamId, name));
+ return httpClient.execute(request, new TypeReference<>() {
+ });
+ }
+
+ @Override
+ public void updateStream(StreamId streamId, String name) {
+ var request = httpClient.preparePutRequest(STREAMS + "/" + streamId, new UpdateStream(name));
+ httpClient.execute(request);
+ }
+
+ @Override
+ public void deleteStream(StreamId streamId) {
+ var request = httpClient.prepareDeleteRequest(STREAMS + "/" + streamId);
+ httpClient.execute(request);
+ }
+
+ record CreateStream(Optional<Long> streamId, String name) {
+ }
+
+ record UpdateStream(String name) {
+ }
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/http/SystemHttpClient.java b/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/http/SystemHttpClient.java
new file mode 100644
index 00000000..1de904f3
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/http/SystemHttpClient.java
@@ -0,0 +1,50 @@
+package rs.iggy.clients.blocking.http;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import rs.iggy.clients.blocking.SystemClient;
+import rs.iggy.system.ClientInfo;
+import rs.iggy.system.ClientInfoDetails;
+import rs.iggy.system.Stats;
+import java.util.List;
+
+class SystemHttpClient implements SystemClient {
+
+ private static final String STATS = "/stats";
+ private static final String CLIENTS = "/clients";
+ private static final String PING = "/ping";
+ private final InternalHttpClient httpClient;
+
+ public SystemHttpClient(InternalHttpClient httpClient) {
+ this.httpClient = httpClient;
+ }
+
+ @Override
+ public Stats getStats() {
+ var request = httpClient.prepareGetRequest(STATS);
+ return httpClient.execute(request, Stats.class);
+ }
+
+ @Override
+ public ClientInfoDetails getMe() {
+ throw new UnsupportedOperationException("Method not available in HTTP client");
+ }
+
+ @Override
+ public ClientInfoDetails getClient(Long clientId) {
+ var request = httpClient.prepareGetRequest(CLIENTS + "/" + clientId);
+ return httpClient.execute(request, ClientInfoDetails.class);
+ }
+
+ @Override
+ public List<ClientInfo> getClients() {
+ var request = httpClient.prepareGetRequest(CLIENTS);
+ return httpClient.execute(request, new TypeReference<>() {
+ });
+ }
+
+ @Override
+ public String ping() {
+ var request = httpClient.prepareGetRequest(PING);
+ return httpClient.executeWithStringResponse(request);
+ }
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/http/TopicsHttpClient.java b/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/http/TopicsHttpClient.java
new file mode 100644
index 00000000..f4f545bf
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/http/TopicsHttpClient.java
@@ -0,0 +1,96 @@
+package rs.iggy.clients.blocking.http;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import rs.iggy.clients.blocking.TopicsClient;
+import rs.iggy.identifier.StreamId;
+import rs.iggy.identifier.TopicId;
+import rs.iggy.topic.CompressionAlgorithm;
+import rs.iggy.topic.Topic;
+import rs.iggy.topic.TopicDetails;
+import java.math.BigInteger;
+import java.util.List;
+import java.util.Optional;
+
+class TopicsHttpClient implements TopicsClient {
+
+ private static final String STREAMS = "/streams";
+ private static final String TOPICS = "/topics";
+ private final InternalHttpClient httpClient;
+
+ public TopicsHttpClient(InternalHttpClient httpClient) {
+ this.httpClient = httpClient;
+ }
+
+ @Override
+ public Optional<TopicDetails> getTopic(StreamId streamId, TopicId topicId) {
+ var request = httpClient.prepareGetRequest(STREAMS + "/" + streamId + TOPICS + "/" + topicId);
+ return httpClient.executeWithOptionalResponse(request, TopicDetails.class);
+ }
+
+ @Override
+ public List<Topic> getTopics(StreamId streamId) {
+ var request = httpClient.prepareGetRequest(STREAMS + "/" + streamId + TOPICS);
+ return httpClient.execute(request, new TypeReference<>() {
+ });
+ }
+
+ @Override
+ public TopicDetails createTopic(StreamId streamId,
+ Optional<Long> topicId,
+ Long partitionsCount,
+ CompressionAlgorithm compressionAlgorithm,
+ BigInteger messageExpiry,
+ BigInteger maxTopicSize,
+ Optional<Short> replicationFactor,
+ String name) {
+ var request = httpClient.preparePostRequest(STREAMS + "/" + streamId + TOPICS,
+ new CreateTopic(topicId,
+ partitionsCount,
+ compressionAlgorithm,
+ messageExpiry,
+ maxTopicSize,
+ replicationFactor,
+ name));
+ return httpClient.execute(request, new TypeReference<>() {
+ });
+ }
+
+ @Override
+ public void updateTopic(StreamId streamId,
+ TopicId topicId,
+ CompressionAlgorithm compressionAlgorithm,
+ BigInteger messageExpiry,
+ BigInteger maxTopicSize,
+ Optional<Short> replicationFactor,
+ String name) {
+ var request = httpClient.preparePutRequest(STREAMS + "/" + streamId + TOPICS + "/" + topicId,
+ new UpdateTopic(compressionAlgorithm, messageExpiry, maxTopicSize, replicationFactor, name));
+ httpClient.execute(request);
+ }
+
+ @Override
+ public void deleteTopic(StreamId streamId, TopicId topicId) {
+ var request = httpClient.prepareDeleteRequest(STREAMS + "/" + streamId + TOPICS + "/" + topicId);
+ httpClient.execute(request);
+ }
+
+ record CreateTopic(
+ Optional<Long> topicId,
+ Long partitionsCount,
+ CompressionAlgorithm compressionAlgorithm,
+ BigInteger messageExpiry,
+ BigInteger maxTopicSize,
+ Optional<Short> replicationFactor,
+ String name
+ ) {
+ }
+
+ record UpdateTopic(
+ CompressionAlgorithm compressionAlgorithm,
+ BigInteger messageExpiry,
+ BigInteger maxTopicSize,
+ Optional<Short> replicationFactor,
+ String name
+ ) {
+ }
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/http/UsersHttpClient.java b/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/http/UsersHttpClient.java
new file mode 100644
index 00000000..d7c78054
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/http/UsersHttpClient.java
@@ -0,0 +1,96 @@
+package rs.iggy.clients.blocking.http;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import rs.iggy.clients.blocking.UsersClient;
+import rs.iggy.identifier.UserId;
+import rs.iggy.user.*;
+import java.util.List;
+import java.util.Optional;
+
+class UsersHttpClient implements UsersClient {
+
+ private static final String USERS = "/users";
+ private final InternalHttpClient httpClient;
+
+ public UsersHttpClient(InternalHttpClient httpClient) {
+ this.httpClient = httpClient;
+ }
+
+ @Override
+ public Optional<UserInfoDetails> getUser(UserId userId) {
+ var request = httpClient.prepareGetRequest(USERS + "/" + userId);
+ return httpClient.executeWithOptionalResponse(request, UserInfoDetails.class);
+ }
+
+ @Override
+ public List<UserInfo> getUsers() {
+ var request = httpClient.prepareGetRequest(USERS);
+ return httpClient.execute(request, new TypeReference<>() {
+ });
+ }
+
+ @Override
+ public UserInfoDetails createUser(String username, String password, UserStatus status, Optional<Permissions> permissions) {
+ var request = httpClient.preparePostRequest(USERS, new CreateUser(username, password, status, permissions));
+ return httpClient.execute(request, new TypeReference<>() {
+ });
+ }
+
+ @Override
+ public void deleteUser(UserId userId) {
+ var request = httpClient.prepareDeleteRequest(USERS + "/" + userId);
+ httpClient.execute(request);
+ }
+
+ @Override
+ public void updateUser(UserId userId, Optional<String> username, Optional<UserStatus> status) {
+ var request = httpClient.preparePutRequest(USERS + "/" + userId, new UpdateUser(username, status));
+ httpClient.execute(request);
+ }
+
+ @Override
+ public void updatePermissions(UserId userId, Optional<Permissions> permissions) {
+ var request = httpClient.preparePutRequest(USERS + "/" + userId + "/permissions",
+ new UpdatePermissions(permissions));
+ httpClient.execute(request);
+ }
+
+ @Override
+ public void changePassword(UserId userId, String currentPassword, String newPassword) {
+ var request = httpClient.preparePutRequest(USERS + "/" + userId + "/password",
+ new ChangePassword(currentPassword, newPassword));
+ httpClient.execute(request);
+ }
+
+ @Override
+ public IdentityInfo login(String username, String password) {
+ String version = "0.6.30";
+ String context = "java-sdk";
+ var request = httpClient.preparePostRequest(USERS + "/login", new Login(username, password, version, context));
+ var response = httpClient.execute(request, IdentityInfo.class);
+ httpClient.setToken(response.accessToken().map(TokenInfo::token));
+ return response;
+ }
+
+ @Override
+ public void logout() {
+ var request = httpClient.prepareDeleteRequest(USERS + "/logout");
+ httpClient.execute(request);
+ httpClient.setToken(Optional.empty());
+ }
+
+ record Login(String username, String password, String version, String context) {
+ }
+
+ private record CreateUser(String username, String password, UserStatus status, Optional<Permissions> permissions) {
+ }
+
+ private record UpdateUser(Optional<String> username, Optional<UserStatus> status) {
+ }
+
+ private record UpdatePermissions(Optional<Permissions> permissions) {
+ }
+
+ private record ChangePassword(String currentPassword, String newPassword) {
+ }
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/http/error/IggyHttpError.java b/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/http/error/IggyHttpError.java
new file mode 100644
index 00000000..dacebe7b
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/http/error/IggyHttpError.java
@@ -0,0 +1,9 @@
+package rs.iggy.clients.blocking.http.error;
+
+public record IggyHttpError(
+ String id,
+ String code,
+ String reason,
+ String field
+) {
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/http/error/IggyHttpException.java b/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/http/error/IggyHttpException.java
new file mode 100644
index 00000000..01b2e522
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/http/error/IggyHttpException.java
@@ -0,0 +1,15 @@
+package rs.iggy.clients.blocking.http.error;
+
+public class IggyHttpException extends RuntimeException {
+ private final IggyHttpError error;
+
+ public IggyHttpException(IggyHttpError error) {
+ super(error.reason());
+ this.error = error;
+ }
+
+ public IggyHttpError getError() {
+ return error;
+ }
+
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/http/package-info.java b/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/http/package-info.java
new file mode 100644
index 00000000..675da73c
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/http/package-info.java
@@ -0,0 +1,4 @@
+@NonNullApi
+package rs.iggy.clients.blocking.http;
+
+import rs.iggy.NonNullApi;
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/tcp/BytesDeserializer.java b/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/tcp/BytesDeserializer.java
new file mode 100644
index 00000000..2fbdb924
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/tcp/BytesDeserializer.java
@@ -0,0 +1,388 @@
+package rs.iggy.clients.blocking.tcp;
+
+import io.netty.buffer.ByteBuf;
+import org.apache.commons.lang3.ArrayUtils;
+import rs.iggy.consumergroup.ConsumerGroup;
+import rs.iggy.consumergroup.ConsumerGroupDetails;
+import rs.iggy.consumergroup.ConsumerGroupMember;
+import rs.iggy.consumeroffset.ConsumerOffsetInfo;
+import rs.iggy.message.*;
+import rs.iggy.partition.Partition;
+import rs.iggy.personalaccesstoken.PersonalAccessTokenInfo;
+import rs.iggy.personalaccesstoken.RawPersonalAccessToken;
+import rs.iggy.stream.StreamBase;
+import rs.iggy.stream.StreamDetails;
+import rs.iggy.system.ClientInfo;
+import rs.iggy.system.ClientInfoDetails;
+import rs.iggy.system.ConsumerGroupInfo;
+import rs.iggy.system.Stats;
+import rs.iggy.topic.CompressionAlgorithm;
+import rs.iggy.topic.Topic;
+import rs.iggy.topic.TopicDetails;
+import rs.iggy.user.*;
+import java.math.BigInteger;
+import java.nio.charset.StandardCharsets;
+import java.util.*;
+
+final class BytesDeserializer {
+
+ private BytesDeserializer() {
+ }
+
+ static StreamBase readStreamBase(ByteBuf response) {
+ var streamId = response.readUnsignedIntLE();
+ var createdAt = readU64AsBigInteger(response);
+ var topicsCount = response.readUnsignedIntLE();
+ var size = readU64AsBigInteger(response);
+ var messagesCount = readU64AsBigInteger(response);
+ var nameLength = response.readByte();
+ var name = response.readCharSequence(nameLength, StandardCharsets.UTF_8).toString();
+
+ return new StreamBase(streamId, createdAt, name, size.toString(), messagesCount, topicsCount);
+ }
+
+ static StreamDetails readStreamDetails(ByteBuf response) {
+ var streamBase = readStreamBase(response);
+
+ List<Topic> topics = new ArrayList<>();
+ if (response.isReadable()) {
+ topics.add(readTopic(response));
+ }
+
+ return new StreamDetails(streamBase, topics);
+ }
+
+ public static TopicDetails readTopicDetails(ByteBuf response) {
+ var topic = readTopic(response);
+
+ List<Partition> partitions = new ArrayList<>();
+ while (response.isReadable()) {
+ partitions.add(readPartition(response));
+ }
+
+ return new TopicDetails(topic, partitions);
+ }
+
+ static Partition readPartition(ByteBuf response) {
+ var partitionId = response.readUnsignedIntLE();
+ var createdAt = readU64AsBigInteger(response);
+ var segmentsCount = response.readUnsignedIntLE();
+ var currentOffset = readU64AsBigInteger(response);
+ var size = readU64AsBigInteger(response);
+ var messagesCount = readU64AsBigInteger(response);
+ return new Partition(partitionId, createdAt, segmentsCount, currentOffset, size.toString(), messagesCount);
+ }
+
+ public static Topic readTopic(ByteBuf response) {
+ var topicId = response.readUnsignedIntLE();
+ var createdAt = readU64AsBigInteger(response);
+ var partitionsCount = response.readUnsignedIntLE();
+ var messageExpiry = readU64AsBigInteger(response);
+ var compressionAlgorithmCode = response.readByte();
+ var maxTopicSize = readU64AsBigInteger(response);
+ var replicationFactor = response.readByte();
+ var size = readU64AsBigInteger(response);
+ var messagesCount = readU64AsBigInteger(response);
+ var nameLength = response.readByte();
+ var name = response.readCharSequence(nameLength, StandardCharsets.UTF_8).toString();
+ return new Topic(topicId,
+ createdAt,
+ name,
+ size.toString(),
+ messageExpiry,
+ CompressionAlgorithm.fromCode(compressionAlgorithmCode),
+ maxTopicSize,
+ (short) replicationFactor,
+ messagesCount,
+ partitionsCount);
+ }
+
+ public static ConsumerGroupDetails readConsumerGroupDetails(ByteBuf response) {
+ var consumerGroup = readConsumerGroup(response);
+
+ List<ConsumerGroupMember> members = new ArrayList<>();
+ while (response.isReadable()) {
+ members.add(readConsumerGroupMember(response));
+ }
+
+ return new ConsumerGroupDetails(consumerGroup, members);
+ }
+
+ static ConsumerGroupMember readConsumerGroupMember(ByteBuf response) {
+ var memberId = response.readUnsignedIntLE();
+ var partitionsCount = response.readUnsignedIntLE();
+ List<Long> partitionIds = new ArrayList<>();
+ for (int i = 0; i < partitionsCount; i++) {
+ partitionIds.add(response.readUnsignedIntLE());
+ }
+ return new ConsumerGroupMember(memberId, partitionsCount, partitionIds);
+ }
+
+ public static ConsumerGroup readConsumerGroup(ByteBuf response) {
+ var groupId = response.readUnsignedIntLE();
+ var partitionsCount = response.readUnsignedIntLE();
+ var membersCount = response.readUnsignedIntLE();
+ var nameLength = response.readByte();
+ var name = response.readCharSequence(nameLength, StandardCharsets.UTF_8).toString();
+ return new ConsumerGroup(groupId, name, partitionsCount, membersCount);
+ }
+
+ public static ConsumerOffsetInfo readConsumerOffsetInfo(ByteBuf response) {
+ var partitionId = response.readUnsignedIntLE();
+ var currentOffset = readU64AsBigInteger(response);
+ var storedOffset = readU64AsBigInteger(response);
+ return new ConsumerOffsetInfo(partitionId, currentOffset, storedOffset);
+ }
+
+ public static PolledMessages readPolledMessages(ByteBuf response) {
+ var partitionId = response.readUnsignedIntLE();
+ var currentOffset = readU64AsBigInteger(response);
+ var _messagesCount = response.readUnsignedIntLE();
+ var messages = new ArrayList<PolledMessage>();
+ while (response.isReadable()) {
+ messages.add(readPolledMessage(response));
+ }
+ return new PolledMessages(partitionId, currentOffset, messages);
+ }
+
+ static PolledMessage readPolledMessage(ByteBuf response) {
+ var offset = readU64AsBigInteger(response);
+ var stateCode = response.readByte();
+ var state = MessageState.fromCode(stateCode);
+ var timestamp = readU64AsBigInteger(response);
+ var id = readBytesMessageId(response);
+ var checksum = response.readUnsignedIntLE();
+ var headersLength = response.readUnsignedIntLE();
+ var headers = Optional.<Map<String, HeaderValue>>empty();
+ if (headersLength > 0) {
+ var headersMap = new HashMap<String, HeaderValue>();
+ ByteBuf headersBytes = response.readBytes(toInt(headersLength));
+ while (headersBytes.isReadable()) {
+ var keyLength = headersBytes.readUnsignedIntLE();
+ var key = headersBytes.readCharSequence(toInt(keyLength), StandardCharsets.UTF_8).toString();
+ var kindCode = headersBytes.readByte();
+ var kind = HeaderKind.fromCode(kindCode);
+ var valueLength = headersBytes.readUnsignedIntLE();
+ var value = headersBytes.readCharSequence(toInt(valueLength), StandardCharsets.UTF_8);
+ headersMap.put(key, new HeaderValue(kind, String.valueOf(value)));
+ }
+ headers = Optional.of(headersMap);
+ }
+ var payloadLength = response.readUnsignedIntLE();
+ var payload = newByteArray(payloadLength);
+ response.readBytes(payload);
+ return new PolledMessage(offset, state, timestamp, id, checksum, headers, payload);
+ }
+
+ static Stats readStats(ByteBuf response) {
+ var processId = response.readUnsignedIntLE();
+ var cpuUsage = response.readFloatLE();
+ var totalCpuUsage = response.readFloatLE();
+ var memoryUsage = readU64AsBigInteger(response);
+ var totalMemory = readU64AsBigInteger(response);
+ var availableMemory = readU64AsBigInteger(response);
+ var runTime = readU64AsBigInteger(response);
+ var startTime = readU64AsBigInteger(response);
+ var readBytes = readU64AsBigInteger(response);
+ var writtenBytes = readU64AsBigInteger(response);
+ var messagesSizeBytes = readU64AsBigInteger(response);
+ var streamsCount = response.readUnsignedIntLE();
+ var topicsCount = response.readUnsignedIntLE();
+ var partitionsCount = response.readUnsignedIntLE();
+ var segmentsCount = response.readUnsignedIntLE();
+ var messagesCount = readU64AsBigInteger(response);
+ var clientsCount = response.readUnsignedIntLE();
+ var consumerGroupsCount = response.readUnsignedIntLE();
+ var hostnameLength = response.readUnsignedIntLE();
+ var hostname = response.readCharSequence(toInt(hostnameLength), StandardCharsets.UTF_8).toString();
+ var osNameLength = response.readUnsignedIntLE();
+ var osName = response.readCharSequence(toInt(osNameLength), StandardCharsets.UTF_8).toString();
+ var osVersionLength = response.readUnsignedIntLE();
+ var osVersion = response.readCharSequence(toInt(osVersionLength), StandardCharsets.UTF_8).toString();
+ var kernelVersionLength = response.readUnsignedIntLE();
+ var kernelVersion = response.readCharSequence(toInt(kernelVersionLength), StandardCharsets.UTF_8).toString();
+
+ return new Stats(processId,
+ cpuUsage,
+ totalCpuUsage,
+ memoryUsage.toString(),
+ totalMemory.toString(),
+ availableMemory.toString(),
+ runTime,
+ startTime,
+ readBytes.toString(),
+ writtenBytes.toString(),
+ messagesSizeBytes.toString(),
+ streamsCount,
+ topicsCount,
+ partitionsCount,
+ segmentsCount,
+ messagesCount,
+ clientsCount,
+ consumerGroupsCount,
+ hostname,
+ osName,
+ osVersion,
+ kernelVersion);
+ }
+
+ static ClientInfoDetails readClientInfoDetails(ByteBuf response) {
+ var clientInfo = readClientInfo(response);
+ var consumerGroups = new ArrayList<ConsumerGroupInfo>();
+ for (int i = 0; i < clientInfo.consumerGroupsCount(); i++) {
+ consumerGroups.add(readConsumerGroupInfo(response));
+ }
+
+ return new ClientInfoDetails(clientInfo, consumerGroups);
+ }
+
+ static ClientInfo readClientInfo(ByteBuf response) {
+ var clientId = response.readUnsignedIntLE();
+ var userId = response.readUnsignedIntLE();
+ var userIdOptional = Optional.<Long>empty();
+ if (userId != 0) {
+ userIdOptional = Optional.of(userId);
+ }
+ var transport = response.readByte();
+ var transportString = "Tcp";
+ if (transport == 2) {
+ transportString = "Quic";
+ }
+ var addressLength = response.readUnsignedIntLE();
+ var address = response.readCharSequence(toInt(addressLength), StandardCharsets.UTF_8).toString();
+ var consumerGroupsCount = response.readUnsignedIntLE();
+ return new ClientInfo(clientId, userIdOptional, address, transportString, consumerGroupsCount);
+ }
+
+ static ConsumerGroupInfo readConsumerGroupInfo(ByteBuf response) {
+ var streamId = response.readUnsignedIntLE();
+ var topicId = response.readUnsignedIntLE();
+ var groupId = response.readUnsignedIntLE();
+
+ return new ConsumerGroupInfo(streamId, topicId, groupId);
+ }
+
+ static UserInfoDetails readUserInfoDetails(ByteBuf response) {
+ var userInfo = readUserInfo(response);
+
+ Optional<Permissions> permissionsOptional = Optional.empty();
+ if (response.readBoolean()) {
+ var permissions = readPermissions(response);
+ permissionsOptional = Optional.of(permissions);
+ }
+
+ return new UserInfoDetails(userInfo, permissionsOptional);
+ }
+
+ static Permissions readPermissions(ByteBuf response) {
+ var _permissionsLength = response.readUnsignedIntLE();
+ var globalPermissions = readGlobalPermissions(response);
+ Map<Long, StreamPermissions> streamPermissionsMap = new HashMap<>();
+ while (response.readBoolean()) {
+ var streamId = response.readUnsignedIntLE();
+ var streamPermissions = readStreamPermissions(response);
+ streamPermissionsMap.put(streamId, streamPermissions);
+ }
+ return new Permissions(globalPermissions, streamPermissionsMap);
+ }
+
+ static StreamPermissions readStreamPermissions(ByteBuf response) {
+ var manageStream = response.readBoolean();
+ var readStream = response.readBoolean();
+ var manageTopics = response.readBoolean();
+ var readTopics = response.readBoolean();
+ var pollMessages = response.readBoolean();
+ var sendMessages = response.readBoolean();
+ Map<Long, TopicPermissions> topicPermissionsMap = new HashMap<>();
+ while (response.readBoolean()) {
+ var topicId = response.readUnsignedIntLE();
+ var topicPermissions = readTopicPermissions(response);
+ topicPermissionsMap.put(topicId, topicPermissions);
+ }
+ return new StreamPermissions(manageStream,
+ readStream,
+ manageTopics,
+ readTopics,
+ pollMessages,
+ sendMessages,
+ topicPermissionsMap);
+ }
+
+ static TopicPermissions readTopicPermissions(ByteBuf response) {
+ var manageTopic = response.readBoolean();
+ var readTopic = response.readBoolean();
+ var pollMessages = response.readBoolean();
+ var sendMessages = response.readBoolean();
+ return new TopicPermissions(manageTopic, readTopic, pollMessages, sendMessages);
+ }
+
+ static GlobalPermissions readGlobalPermissions(ByteBuf response) {
+ var manageServers = response.readBoolean();
+ var readServers = response.readBoolean();
+ var manageUsers = response.readBoolean();
+ var readUsers = response.readBoolean();
+ var manageStreams = response.readBoolean();
+ var readStreams = response.readBoolean();
+ var manageTopics = response.readBoolean();
+ var readTopics = response.readBoolean();
+ var pollMessages = response.readBoolean();
+ var sendMessages = response.readBoolean();
+ return new GlobalPermissions(manageServers,
+ readServers,
+ manageUsers,
+ readUsers,
+ manageStreams,
+ readStreams,
+ manageTopics,
+ readTopics,
+ pollMessages,
+ sendMessages);
+ }
+
+ static UserInfo readUserInfo(ByteBuf response) {
+ var userId = response.readUnsignedIntLE();
+ var createdAt = readU64AsBigInteger(response);
+ var statusCode = response.readByte();
+ var status = UserStatus.fromCode(statusCode);
+ var usernameLength = response.readByte();
+ var username = response.readCharSequence(usernameLength, StandardCharsets.UTF_8).toString();
+ return new UserInfo(userId, createdAt, status, username);
+ }
+
+ static RawPersonalAccessToken readRawPersonalAccessToken(ByteBuf response) {
+ var tokenLength = response.readByte();
+ var token = response.readCharSequence(tokenLength, StandardCharsets.UTF_8).toString();
+ return new RawPersonalAccessToken(token);
+ }
+
+ static PersonalAccessTokenInfo readPersonalAccessTokenInfo(ByteBuf response) {
+ var nameLength = response.readByte();
+ var name = response.readCharSequence(nameLength, StandardCharsets.UTF_8).toString();
+ var expiry = readU64AsBigInteger(response);
+ Optional<BigInteger> expiryOptional = expiry.equals(BigInteger.ZERO) ? Optional.empty() : Optional.of(expiry);
+ return new PersonalAccessTokenInfo(name, expiryOptional);
+ }
+
+ private static BigInteger readU64AsBigInteger(ByteBuf buffer) {
+ var bytesArray = new byte[9];
+ buffer.readBytes(bytesArray, 0, 8);
+ ArrayUtils.reverse(bytesArray);
+ return new BigInteger(bytesArray);
+ }
+
+ private static BytesMessageId readBytesMessageId(ByteBuf buffer) {
+ var bytesArray = new byte[16];
+ buffer.readBytes(bytesArray);
+ ArrayUtils.reverse(bytesArray);
+ return new BytesMessageId(bytesArray);
+ }
+
+ private static int toInt(Long size) {
+ return size.intValue();
+ }
+
+ private static byte[] newByteArray(Long size) {
+ return new byte[size.intValue()];
+ }
+
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/tcp/BytesSerializer.java b/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/tcp/BytesSerializer.java
new file mode 100644
index 00000000..f8db45fa
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/tcp/BytesSerializer.java
@@ -0,0 +1,200 @@
+package rs.iggy.clients.blocking.tcp;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import org.apache.commons.lang3.ArrayUtils;
+import rs.iggy.consumergroup.Consumer;
+import rs.iggy.identifier.Identifier;
+import rs.iggy.message.HeaderValue;
+import rs.iggy.message.Message;
+import rs.iggy.message.Partitioning;
+import rs.iggy.message.PollingStrategy;
+import rs.iggy.user.GlobalPermissions;
+import rs.iggy.user.Permissions;
+import rs.iggy.user.StreamPermissions;
+import rs.iggy.user.TopicPermissions;
+import java.math.BigInteger;
+import java.util.Map;
+
+public final class BytesSerializer {
+
+ private BytesSerializer() {
+ }
+
+ static ByteBuf toBytes(Consumer consumer) {
+ ByteBuf buffer = Unpooled.buffer();
+ buffer.writeByte(consumer.kind().asCode());
+ buffer.writeBytes(toBytes(consumer.id()));
+ return buffer;
+ }
+
+ static ByteBuf toBytes(Identifier identifier) {
+ if (identifier.getKind() == 1) {
+ ByteBuf buffer = Unpooled.buffer(6);
+ buffer.writeByte(1);
+ buffer.writeByte(4);
+ buffer.writeIntLE(identifier.getId().intValue());
+ return buffer;
+ } else if (identifier.getKind() == 2) {
+ ByteBuf buffer = Unpooled.buffer(2 + identifier.getName().length());
+ buffer.writeByte(2);
+ buffer.writeByte(identifier.getName().length());
+ buffer.writeBytes(identifier.getName().getBytes());
+ return buffer;
+ } else {
+ throw new IllegalArgumentException("Unknown identifier kind: " + identifier.getKind());
+ }
+ }
+
+ static ByteBuf nameToBytes(String name) {
+ ByteBuf buffer = Unpooled.buffer(1 + name.length());
+ buffer.writeByte(name.length());
+ buffer.writeBytes(name.getBytes());
+ return buffer;
+ }
+
+ static ByteBuf toBytes(Partitioning partitioning) {
+ ByteBuf buffer = Unpooled.buffer(2 + partitioning.value().length);
+ buffer.writeByte(partitioning.kind().asCode());
+ buffer.writeByte(partitioning.value().length);
+ buffer.writeBytes(partitioning.value());
+ return buffer;
+ }
+
+ static ByteBuf toBytes(Message message) {
+ var buffer = Unpooled.buffer();
+ buffer.writeBytes(message.id().toBytes());
+ message.headers().ifPresentOrElse((headers) -> {
+ var headersBytes = toBytes(headers);
+ buffer.writeIntLE(headersBytes.readableBytes());
+ buffer.writeBytes(headersBytes);
+ }, () -> buffer.writeIntLE(0));
+ buffer.writeIntLE(message.payload().length);
+ buffer.writeBytes(message.payload());
+ return buffer;
+ }
+
+ static ByteBuf toBytes(PollingStrategy strategy) {
+ var buffer = Unpooled.buffer(9);
+ buffer.writeByte(strategy.kind().asCode());
+ buffer.writeBytes(toBytesAsU64(strategy.value()));
+ return buffer;
+ }
+
+ static ByteBuf toBytes(Map<String, HeaderValue> headers) {
+ if (headers.isEmpty()) {
+ return Unpooled.EMPTY_BUFFER;
+ }
+ var buffer = Unpooled.buffer();
+ for (Map.Entry<String, HeaderValue> entry : headers.entrySet()) {
+ String key = entry.getKey();
+ buffer.writeIntLE(key.length());
+ buffer.writeBytes(key.getBytes());
+
+ HeaderValue value = entry.getValue();
+ buffer.writeByte(value.kind().asCode());
+ buffer.writeIntLE(value.value().length());
+ buffer.writeBytes(value.value().getBytes());
+ }
+ return buffer;
+ }
+
+ static ByteBuf toBytes(Permissions permissions) {
+ var buffer = Unpooled.buffer();
+ buffer.writeBytes(toBytes(permissions.global()));
+ if (permissions.streams().isEmpty()) {
+ buffer.writeByte(0);
+ } else {
+ for (Map.Entry<Long, StreamPermissions> entry : permissions.streams().entrySet()) {
+ buffer.writeByte(1);
+ buffer.writeIntLE(entry.getKey().intValue());
+ buffer.writeBytes(toBytes(entry.getValue()));
+ }
+ buffer.writeByte(0);
+ }
+
+ return buffer;
+ }
+
+ static ByteBuf toBytes(GlobalPermissions permissions) {
+ var buffer = Unpooled.buffer();
+ buffer.writeBoolean(permissions.manageServers());
+ buffer.writeBoolean(permissions.readServers());
+ buffer.writeBoolean(permissions.manageUsers());
+ buffer.writeBoolean(permissions.readUsers());
+ buffer.writeBoolean(permissions.manageStreams());
+ buffer.writeBoolean(permissions.readStreams());
+ buffer.writeBoolean(permissions.manageTopics());
+ buffer.writeBoolean(permissions.readTopics());
+ buffer.writeBoolean(permissions.pollMessages());
+ buffer.writeBoolean(permissions.sendMessages());
+ return buffer;
+ }
+
+ static ByteBuf toBytes(StreamPermissions permissions) {
+ var buffer = Unpooled.buffer();
+ buffer.writeBoolean(permissions.manageStream());
+ buffer.writeBoolean(permissions.readStream());
+ buffer.writeBoolean(permissions.manageTopics());
+ buffer.writeBoolean(permissions.readTopics());
+ buffer.writeBoolean(permissions.pollMessages());
+ buffer.writeBoolean(permissions.sendMessages());
+
+ if (permissions.topics().isEmpty()) {
+ buffer.writeByte(0);
+ } else {
+ for (Map.Entry<Long, TopicPermissions> entry : permissions.topics().entrySet()) {
+ buffer.writeByte(1);
+ buffer.writeIntLE(entry.getKey().intValue());
+ buffer.writeBytes(toBytes(entry.getValue()));
+ }
+ buffer.writeByte(0);
+ }
+
+ return buffer;
+ }
+
+ static ByteBuf toBytes(TopicPermissions permissions) {
+ var buffer = Unpooled.buffer();
+ buffer.writeBoolean(permissions.manageTopic());
+ buffer.writeBoolean(permissions.readTopic());
+ buffer.writeBoolean(permissions.pollMessages());
+ buffer.writeBoolean(permissions.sendMessages());
+ return buffer;
+ }
+
+ static ByteBuf toBytesAsU64(BigInteger value) {
+ if (value.signum() == -1) {
+ throw new IllegalArgumentException("Negative value cannot be serialized to unsigned 64: " + value);
+ }
+ ByteBuf buffer = Unpooled.buffer(8, 8);
+ byte[] valueAsBytes = value.toByteArray();
+ if (valueAsBytes.length > 9) {
+ throw new IllegalArgumentException();
+ }
+ ArrayUtils.reverse(valueAsBytes);
+ buffer.writeBytes(valueAsBytes, 0, Math.min(8, valueAsBytes.length));
+ if (valueAsBytes.length < 8) {
+ buffer.writeZero(8 - valueAsBytes.length);
+ }
+ return buffer;
+ }
+
+ public static ByteBuf toBytesAsU128(BigInteger value) {
+ if (value.signum() == -1) {
+ throw new IllegalArgumentException("Negative value cannot be serialized to unsigned 128: " + value);
+ }
+ ByteBuf buffer = Unpooled.buffer(16, 16);
+ byte[] valueAsBytes = value.toByteArray();
+ if (valueAsBytes.length > 17) {
+ throw new IllegalArgumentException();
+ }
+ ArrayUtils.reverse(valueAsBytes);
+ buffer.writeBytes(valueAsBytes, 0, Math.min(16, valueAsBytes.length));
+ if (valueAsBytes.length < 16) {
+ buffer.writeZero(16 - valueAsBytes.length);
+ }
+ return buffer;
+ }
+
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/tcp/ConsumerGroupsTcpClient.java b/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/tcp/ConsumerGroupsTcpClient.java
new file mode 100644
index 00000000..11da588a
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/tcp/ConsumerGroupsTcpClient.java
@@ -0,0 +1,99 @@
+package rs.iggy.clients.blocking.tcp;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import rs.iggy.clients.blocking.ConsumerGroupsClient;
+import rs.iggy.consumergroup.ConsumerGroup;
+import rs.iggy.consumergroup.ConsumerGroupDetails;
+import rs.iggy.identifier.ConsumerId;
+import rs.iggy.identifier.StreamId;
+import rs.iggy.identifier.TopicId;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Optional;
+import static rs.iggy.clients.blocking.tcp.BytesDeserializer.readConsumerGroup;
+import static rs.iggy.clients.blocking.tcp.BytesDeserializer.readConsumerGroupDetails;
+import static rs.iggy.clients.blocking.tcp.BytesSerializer.nameToBytes;
+import static rs.iggy.clients.blocking.tcp.BytesSerializer.toBytes;
+
+class ConsumerGroupsTcpClient implements ConsumerGroupsClient {
+
+ private static final int GET_CONSUMER_GROUP_CODE = 600;
+ private static final int GET_CONSUMER_GROUPS_CODE = 601;
+ private static final int CREATE_CONSUMER_GROUP_CODE = 602;
+ private static final int DELETE_CONSUMER_GROUP_CODE = 603;
+ private static final int JOIN_CONSUMER_GROUP_CODE = 604;
+ private static final int LEAVE_CONSUMER_GROUP_CODE = 605;
+
+ private final InternalTcpClient tcpClient;
+
+ public ConsumerGroupsTcpClient(InternalTcpClient tcpClient) {
+ this.tcpClient = tcpClient;
+ }
+
+ @Override
+ public Optional<ConsumerGroupDetails> getConsumerGroup(StreamId streamId, TopicId topicId, ConsumerId groupId) {
+ var payload = toBytes(streamId);
+ payload.writeBytes(toBytes(topicId));
+ payload.writeBytes(toBytes(groupId));
+ var response = tcpClient.send(GET_CONSUMER_GROUP_CODE, payload);
+ if (response.isReadable()) {
+ return Optional.of(readConsumerGroupDetails(response));
+ }
+ return Optional.empty();
+ }
+
+ @Override
+ public List<ConsumerGroup> getConsumerGroups(StreamId streamId, TopicId topicId) {
+ var payload = toBytes(streamId);
+ payload.writeBytes(toBytes(topicId));
+ var response = tcpClient.send(GET_CONSUMER_GROUPS_CODE, payload);
+ List<ConsumerGroup> groups = new ArrayList<>();
+ while (response.isReadable()) {
+ groups.add(readConsumerGroup(response));
+ }
+ return groups;
+ }
+
+ @Override
+ public ConsumerGroupDetails createConsumerGroup(StreamId streamId, TopicId topicId, Optional<Long> groupId, String name) {
+ var streamIdBytes = toBytes(streamId);
+ var topicIdBytes = toBytes(topicId);
+ var payload = Unpooled.buffer(5 + streamIdBytes.readableBytes() + topicIdBytes.readableBytes() + name.length());
+
+ payload.writeBytes(streamIdBytes);
+ payload.writeBytes(topicIdBytes);
+ payload.writeIntLE(groupId.orElse(0L).intValue());
+ payload.writeBytes(nameToBytes(name));
+
+ ByteBuf response = tcpClient.send(CREATE_CONSUMER_GROUP_CODE, payload);
+ return readConsumerGroupDetails(response);
+ }
+
+ @Override
+ public void deleteConsumerGroup(StreamId streamId, TopicId topicId, ConsumerId groupId) {
+ var payload = toBytes(streamId);
+ payload.writeBytes(toBytes(topicId));
+ payload.writeBytes(toBytes(groupId));
+ tcpClient.send(DELETE_CONSUMER_GROUP_CODE, payload);
+ }
+
+ @Override
+ public void joinConsumerGroup(StreamId streamId, TopicId topicId, ConsumerId groupId) {
+ var payload = toBytes(streamId);
+ payload.writeBytes(toBytes(topicId));
+ payload.writeBytes(toBytes(groupId));
+
+ tcpClient.send(JOIN_CONSUMER_GROUP_CODE, payload);
+ }
+
+ @Override
+ public void leaveConsumerGroup(StreamId streamId, TopicId topicId, ConsumerId groupId) {
+ var payload = toBytes(streamId);
+ payload.writeBytes(toBytes(topicId));
+ payload.writeBytes(toBytes(groupId));
+
+ tcpClient.send(LEAVE_CONSUMER_GROUP_CODE, payload);
+ }
+
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/tcp/ConsumerOffsetTcpClient.java b/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/tcp/ConsumerOffsetTcpClient.java
new file mode 100644
index 00000000..b5afe41d
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/tcp/ConsumerOffsetTcpClient.java
@@ -0,0 +1,50 @@
+package rs.iggy.clients.blocking.tcp;
+
+import rs.iggy.clients.blocking.ConsumerOffsetsClient;
+import rs.iggy.consumergroup.Consumer;
+import rs.iggy.consumeroffset.ConsumerOffsetInfo;
+import rs.iggy.identifier.StreamId;
+import rs.iggy.identifier.TopicId;
+import java.math.BigInteger;
+import java.util.Optional;
+import static rs.iggy.clients.blocking.tcp.BytesDeserializer.readConsumerOffsetInfo;
+import static rs.iggy.clients.blocking.tcp.BytesSerializer.toBytes;
+import static rs.iggy.clients.blocking.tcp.BytesSerializer.toBytesAsU64;
+
+class ConsumerOffsetTcpClient implements ConsumerOffsetsClient {
+
+ private static final int GET_CONSUMER_OFFSET_CODE = 120;
+ private static final int STORE_CONSUMER_OFFSET_CODE = 121;
+
+ private final InternalTcpClient tcpClient;
+
+ public ConsumerOffsetTcpClient(InternalTcpClient tcpClient) {
+ this.tcpClient = tcpClient;
+ }
+
+ @Override
+ public void storeConsumerOffset(StreamId streamId, TopicId topicId, Optional<Long> partitionId, Consumer consumer, BigInteger offset) {
+ var payload = toBytes(consumer);
+ payload.writeBytes(toBytes(streamId));
+ payload.writeBytes(toBytes(topicId));
+ payload.writeIntLE(partitionId.orElse(0L).intValue());
+ payload.writeBytes(toBytesAsU64(offset));
+
+ tcpClient.send(STORE_CONSUMER_OFFSET_CODE, payload);
+ }
+
+ @Override
+ public Optional<ConsumerOffsetInfo> getConsumerOffset(StreamId streamId, TopicId topicId, Optional<Long> partitionId, Consumer consumer) {
+ var payload = toBytes(consumer);
+ payload.writeBytes(toBytes(streamId));
+ payload.writeBytes(toBytes(topicId));
+ payload.writeIntLE(partitionId.orElse(0L).intValue());
+
+ var response = tcpClient.send(GET_CONSUMER_OFFSET_CODE, payload);
+ if (response.isReadable()) {
+ return Optional.of(readConsumerOffsetInfo(response));
+ }
+ return Optional.empty();
+ }
+
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/tcp/IggyTcpClient.java b/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/tcp/IggyTcpClient.java
new file mode 100644
index 00000000..dd9cace7
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/tcp/IggyTcpClient.java
@@ -0,0 +1,76 @@
+package rs.iggy.clients.blocking.tcp;
+
+import rs.iggy.clients.blocking.*;
+
+public class IggyTcpClient implements IggyBaseClient {
+
+ private final UsersTcpClient usersClient;
+ private final StreamsTcpClient streamsClient;
+ private final TopicsTcpClient topicsClient;
+ private final PartitionsTcpClient partitionsClient;
+ private final ConsumerGroupsTcpClient consumerGroupsClient;
+ private final ConsumerOffsetTcpClient consumerOffsetsClient;
+ private final MessagesTcpClient messagesClient;
+ private final SystemTcpClient systemClient;
+ private final PersonalAccessTokensTcpClient personalAccessTokensClient;
+
+ public IggyTcpClient(String host, Integer port) {
+ InternalTcpClient tcpClient = new InternalTcpClient(host, port);
+ tcpClient.connect();
+ usersClient = new UsersTcpClient(tcpClient);
+ streamsClient = new StreamsTcpClient(tcpClient);
+ topicsClient = new TopicsTcpClient(tcpClient);
+ partitionsClient = new PartitionsTcpClient(tcpClient);
+ consumerGroupsClient = new ConsumerGroupsTcpClient(tcpClient);
+ consumerOffsetsClient = new ConsumerOffsetTcpClient(tcpClient);
+ messagesClient = new MessagesTcpClient(tcpClient);
+ systemClient = new SystemTcpClient(tcpClient);
+ personalAccessTokensClient = new PersonalAccessTokensTcpClient(tcpClient);
+ }
+
+ @Override
+ public SystemClient system() {
+ return systemClient;
+ }
+
+ @Override
+ public StreamsClient streams() {
+ return streamsClient;
+ }
+
+ @Override
+ public UsersClient users() {
+ return usersClient;
+ }
+
+ @Override
+ public TopicsClient topics() {
+ return topicsClient;
+ }
+
+ @Override
+ public PartitionsClient partitions() {
+ return partitionsClient;
+ }
+
+ @Override
+ public ConsumerGroupsClient consumerGroups() {
+ return consumerGroupsClient;
+ }
+
+ @Override
+ public ConsumerOffsetsClient consumerOffsets() {
+ return consumerOffsetsClient;
+ }
+
+ @Override
+ public MessagesClient messages() {
+ return messagesClient;
+ }
+
+ @Override
+ public PersonalAccessTokensClient personalAccessTokens() {
+ return personalAccessTokensClient;
+ }
+
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/tcp/InternalTcpClient.java b/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/tcp/InternalTcpClient.java
new file mode 100644
index 00000000..ce595f49
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/tcp/InternalTcpClient.java
@@ -0,0 +1,87 @@
+package rs.iggy.clients.blocking.tcp;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.ByteToMessageDecoder;
+import reactor.core.publisher.Mono;
+import reactor.netty.Connection;
+import reactor.netty.tcp.TcpClient;
+import java.util.List;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+
+final class InternalTcpClient {
+
+ private static final int REQUEST_INITIAL_BYTES_LENGTH = 4;
+ private static final int COMMAND_LENGTH = 4;
+ private static final int RESPONSE_INITIAL_BYTES_LENGTH = 8;
+
+ private final TcpClient client;
+ private final BlockingQueue<IggyResponse> responses = new LinkedBlockingQueue<>();
+ private Connection connection;
+
+ InternalTcpClient(String host, Integer port) {
+ client = TcpClient.create()
+ .host(host)
+ .port(port)
+ .doOnConnected(conn -> conn.addHandlerLast(new IggyResponseDecoder()));
+ }
+
+ void connect() {
+ this.connection = client.connectNow();
+ this.connection.inbound().receiveObject().ofType(IggyResponse.class).subscribe(responses::add);
+ }
+
+ ByteBuf send(int command) {
+ return send(command, Unpooled.EMPTY_BUFFER);
+ }
+
+ ByteBuf send(int command, ByteBuf payload) {
+ var payloadSize = payload.readableBytes() + COMMAND_LENGTH;
+ var buffer = Unpooled.buffer(REQUEST_INITIAL_BYTES_LENGTH + payloadSize);
+ buffer.writeIntLE(payloadSize);
+ buffer.writeIntLE(command);
+ buffer.writeBytes(payload);
+
+ connection.outbound().send(Mono.just(buffer)).then().block();
+ try {
+ IggyResponse response = responses.take();
+ return handleResponse(response);
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private ByteBuf handleResponse(IggyResponse response) {
+ if (response.status() != 0) {
+ throw new RuntimeException("Received an invalid response with status " + response.status());
+ }
+ if (response.length() == 0) {
+ return Unpooled.EMPTY_BUFFER;
+ }
+ return response.payload();
+ }
+
+ static class IggyResponseDecoder extends ByteToMessageDecoder {
+ @Override
+ protected void decode(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf, List<Object> list) {
+ if (byteBuf.readableBytes() < RESPONSE_INITIAL_BYTES_LENGTH) {
+ return;
+ }
+ byteBuf.markReaderIndex();
+ var status = byteBuf.readUnsignedIntLE();
+ var responseLength = byteBuf.readUnsignedIntLE();
+ if (byteBuf.readableBytes() < responseLength) {
+ byteBuf.resetReaderIndex();
+ return;
+ }
+ var length = Long.valueOf(responseLength).intValue();
+ list.add(new IggyResponse(status, length, byteBuf.readBytes(length)));
+ }
+ }
+
+ record IggyResponse(long status, int length, ByteBuf payload) {
+ }
+
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/tcp/MessagesTcpClient.java b/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/tcp/MessagesTcpClient.java
new file mode 100644
index 00000000..70e12438
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/tcp/MessagesTcpClient.java
@@ -0,0 +1,52 @@
+package rs.iggy.clients.blocking.tcp;
+
+import rs.iggy.clients.blocking.MessagesClient;
+import rs.iggy.consumergroup.Consumer;
+import rs.iggy.identifier.StreamId;
+import rs.iggy.identifier.TopicId;
+import rs.iggy.message.Message;
+import rs.iggy.message.Partitioning;
+import rs.iggy.message.PolledMessages;
+import rs.iggy.message.PollingStrategy;
+import java.util.List;
+import java.util.Optional;
+import static rs.iggy.clients.blocking.tcp.BytesSerializer.toBytes;
+
+class MessagesTcpClient implements MessagesClient {
+
+ private static final int POLL_MESSAGES_CODE = 100;
+ private static final int SEND_MESSAGES_CODE = 101;
+
+ private final InternalTcpClient tcpClient;
+
+ public MessagesTcpClient(InternalTcpClient tcpClient) {
+ this.tcpClient = tcpClient;
+ }
+
+ @Override
+ public PolledMessages pollMessages(StreamId streamId, TopicId topicId, Optional<Long> partitionId, Consumer consumer, PollingStrategy strategy, Long count, boolean autoCommit) {
+ var payload = toBytes(consumer);
+ payload.writeBytes(toBytes(streamId));
+ payload.writeBytes(toBytes(topicId));
+ payload.writeIntLE(partitionId.orElse(0L).intValue());
+ payload.writeBytes(toBytes(strategy));
+ payload.writeIntLE(count.intValue());
+ payload.writeByte(autoCommit ? 1 : 0);
+
+ var response = tcpClient.send(POLL_MESSAGES_CODE, payload);
+
+ return BytesDeserializer.readPolledMessages(response);
+ }
+
+ @Override
+ public void sendMessages(StreamId streamId, TopicId topicId, Partitioning partitioning, List<Message> messages) {
+ var payload = toBytes(streamId);
+ payload.writeBytes(toBytes(topicId));
+ payload.writeBytes(toBytes(partitioning));
+ for (var message : messages) {
+ payload.writeBytes(toBytes(message));
+ }
+
+ tcpClient.send(SEND_MESSAGES_CODE, payload);
+ }
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/tcp/PartitionsTcpClient.java b/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/tcp/PartitionsTcpClient.java
new file mode 100644
index 00000000..44bb9f3f
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/tcp/PartitionsTcpClient.java
@@ -0,0 +1,33 @@
+package rs.iggy.clients.blocking.tcp;
+
+import rs.iggy.clients.blocking.PartitionsClient;
+import rs.iggy.identifier.StreamId;
+import rs.iggy.identifier.TopicId;
+import static rs.iggy.clients.blocking.tcp.BytesSerializer.toBytes;
+
+class PartitionsTcpClient implements PartitionsClient {
+
+ private static final int CREATE_PARTITION_CODE = 402;
+ private static final int DELETE_PARTITION_CODE = 403;
+ private final InternalTcpClient tcpClient;
+
+ PartitionsTcpClient(InternalTcpClient tcpClient) {
+ this.tcpClient = tcpClient;
+ }
+
+ @Override
+ public void createPartitions(StreamId streamId, TopicId topicId, Long partitionsCount) {
+ var payload = toBytes(streamId);
+ payload.writeBytes(toBytes(topicId));
+ payload.writeIntLE(partitionsCount.intValue());
+ tcpClient.send(CREATE_PARTITION_CODE, payload);
+ }
+
+ @Override
+ public void deletePartitions(StreamId streamId, TopicId topicId, Long partitionsCount) {
+ var payload = toBytes(streamId);
+ payload.writeBytes(toBytes(topicId));
+ payload.writeIntLE(partitionsCount.intValue());
+ tcpClient.send(DELETE_PARTITION_CODE, payload);
+ }
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/tcp/PersonalAccessTokensTcpClient.java b/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/tcp/PersonalAccessTokensTcpClient.java
new file mode 100644
index 00000000..c482df06
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/tcp/PersonalAccessTokensTcpClient.java
@@ -0,0 +1,62 @@
+package rs.iggy.clients.blocking.tcp;
+
+import io.netty.buffer.Unpooled;
+import rs.iggy.clients.blocking.PersonalAccessTokensClient;
+import rs.iggy.personalaccesstoken.PersonalAccessTokenInfo;
+import rs.iggy.personalaccesstoken.RawPersonalAccessToken;
+import rs.iggy.user.IdentityInfo;
+import java.math.BigInteger;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Optional;
+import static rs.iggy.clients.blocking.tcp.BytesDeserializer.readPersonalAccessTokenInfo;
+import static rs.iggy.clients.blocking.tcp.BytesDeserializer.readRawPersonalAccessToken;
+import static rs.iggy.clients.blocking.tcp.BytesSerializer.nameToBytes;
+import static rs.iggy.clients.blocking.tcp.BytesSerializer.toBytesAsU64;
+
+class PersonalAccessTokensTcpClient implements PersonalAccessTokensClient {
+
+ private static final int GET_PERSONAL_ACCESS_TOKENS_CODE = 41;
+ private static final int CREATE_PERSONAL_ACCESS_TOKEN_CODE = 42;
+ private static final int DELETE_PERSONAL_ACCESS_TOKEN_CODE = 43;
+ private static final int LOGIN_WITH_PERSONAL_ACCESS_TOKEN_CODE = 44;
+
+ private final InternalTcpClient tcpClient;
+
+ public PersonalAccessTokensTcpClient(InternalTcpClient tcpClient) {
+ this.tcpClient = tcpClient;
+ }
+
+ @Override
+ public RawPersonalAccessToken createPersonalAccessToken(String name, BigInteger expiry) {
+ var payload = Unpooled.buffer();
+ payload.writeBytes(nameToBytes(name));
+ payload.writeBytes(toBytesAsU64(expiry));
+ var response = tcpClient.send(CREATE_PERSONAL_ACCESS_TOKEN_CODE, payload);
+ return readRawPersonalAccessToken(response);
+ }
+
+ @Override
+ public List<PersonalAccessTokenInfo> getPersonalAccessTokens() {
+ var response = tcpClient.send(GET_PERSONAL_ACCESS_TOKENS_CODE);
+ var tokens = new ArrayList<PersonalAccessTokenInfo>();
+ while (response.isReadable()) {
+ tokens.add(readPersonalAccessTokenInfo(response));
+ }
+ return tokens;
+ }
+
+ @Override
+ public void deletePersonalAccessToken(String name) {
+ var payload = nameToBytes(name);
+ tcpClient.send(DELETE_PERSONAL_ACCESS_TOKEN_CODE, payload);
+ }
+
+ @Override
+ public IdentityInfo loginWithPersonalAccessToken(String token) {
+ var payload = nameToBytes(token);
+ var response = tcpClient.send(LOGIN_WITH_PERSONAL_ACCESS_TOKEN_CODE, payload);
+ var userId = response.readUnsignedIntLE();
+ return new IdentityInfo(userId, Optional.empty());
+ }
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/tcp/StreamsTcpClient.java b/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/tcp/StreamsTcpClient.java
new file mode 100644
index 00000000..15278931
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/tcp/StreamsTcpClient.java
@@ -0,0 +1,78 @@
+package rs.iggy.clients.blocking.tcp;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import rs.iggy.clients.blocking.StreamsClient;
+import rs.iggy.identifier.StreamId;
+import rs.iggy.stream.StreamBase;
+import rs.iggy.stream.StreamDetails;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Optional;
+import static rs.iggy.clients.blocking.tcp.BytesDeserializer.readStreamBase;
+import static rs.iggy.clients.blocking.tcp.BytesDeserializer.readStreamDetails;
+import static rs.iggy.clients.blocking.tcp.BytesSerializer.nameToBytes;
+import static rs.iggy.clients.blocking.tcp.BytesSerializer.toBytes;
+
+class StreamsTcpClient implements StreamsClient {
+
+ private static final int GET_STREAM_CODE = 200;
+ private static final int GET_STREAMS_CODE = 201;
+ private static final int CREATE_STREAM_CODE = 202;
+ private static final int DELETE_STREAM_CODE = 203;
+ private static final int UPDATE_STREAM_CODE = 204;
+ private final InternalTcpClient tcpClient;
+
+ StreamsTcpClient(InternalTcpClient tcpClient) {
+ this.tcpClient = tcpClient;
+ }
+
+ @Override
+ public StreamDetails createStream(Optional<Long> streamId, String name) {
+ var payloadSize = 4 + 1 + name.length();
+ var payload = Unpooled.buffer(payloadSize);
+
+ payload.writeIntLE(streamId.orElse(0L).intValue());
+ payload.writeBytes(nameToBytes(name));
+ var response = tcpClient.send(CREATE_STREAM_CODE, payload);
+ return readStreamDetails(response);
+ }
+
+ @Override
+ public Optional<StreamDetails> getStream(StreamId streamId) {
+ var payload = toBytes(streamId);
+ var response = tcpClient.send(GET_STREAM_CODE, payload);
+ if (response.isReadable()) {
+ return Optional.of(readStreamDetails(response));
+ }
+ return Optional.empty();
+ }
+
+ @Override
+ public List<StreamBase> getStreams() {
+ ByteBuf response = tcpClient.send(GET_STREAMS_CODE);
+ List<StreamBase> streams = new ArrayList<>();
+ while (response.isReadable()) {
+ streams.add(readStreamBase(response));
+ }
+ return streams;
+ }
+
+ @Override
+ public void updateStream(StreamId streamId, String name) {
+ var payloadSize = 1 + name.length();
+ var idBytes = toBytes(streamId);
+ var payload = Unpooled.buffer(payloadSize + idBytes.capacity());
+
+ payload.writeBytes(idBytes);
+ payload.writeBytes(nameToBytes(name));
+ tcpClient.send(UPDATE_STREAM_CODE, payload);
+ }
+
+ @Override
+ public void deleteStream(StreamId streamId) {
+ var payload = toBytes(streamId);
+ tcpClient.send(DELETE_STREAM_CODE, payload);
+ }
+
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/tcp/SystemTcpClient.java b/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/tcp/SystemTcpClient.java
new file mode 100644
index 00000000..bf18b185
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/tcp/SystemTcpClient.java
@@ -0,0 +1,60 @@
+package rs.iggy.clients.blocking.tcp;
+
+import io.netty.buffer.Unpooled;
+import rs.iggy.clients.blocking.SystemClient;
+import rs.iggy.system.ClientInfo;
+import rs.iggy.system.ClientInfoDetails;
+import rs.iggy.system.Stats;
+import java.util.ArrayList;
+import java.util.List;
+import static rs.iggy.clients.blocking.tcp.BytesDeserializer.*;
+
+class SystemTcpClient implements SystemClient {
+ private static final int PING_CODE = 1;
+ private static final int GET_STATS_CODE = 10;
+ private static final int GET_ME_CODE = 20;
+ private static final int GET_CLIENT_CODE = 21;
+ private static final int GET_CLIENTS_CODE = 22;
+
+ private final InternalTcpClient tcpClient;
+
+ SystemTcpClient(InternalTcpClient tcpClient) {
+ this.tcpClient = tcpClient;
+ }
+
+ @Override
+ public Stats getStats() {
+ var response = tcpClient.send(GET_STATS_CODE);
+ return readStats(response);
+ }
+
+ @Override
+ public ClientInfoDetails getMe() {
+ var response = tcpClient.send(GET_ME_CODE);
+ return readClientInfoDetails(response);
+ }
+
+ @Override
+ public ClientInfoDetails getClient(Long clientId) {
+ var payload = Unpooled.buffer(4);
+ payload.writeIntLE(clientId.intValue());
+ var response = tcpClient.send(GET_CLIENT_CODE, payload);
+ return readClientInfoDetails(response);
+ }
+
+ @Override
+ public List<ClientInfo> getClients() {
+ var response = tcpClient.send(GET_CLIENTS_CODE);
+ List<ClientInfo> clients = new ArrayList<>();
+ while (response.isReadable()) {
+ clients.add(readClientInfo(response));
+ }
+ return clients;
+ }
+
+ @Override
+ public String ping() {
+ tcpClient.send(PING_CODE);
+ return "";
+ }
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/tcp/TopicsTcpClient.java b/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/tcp/TopicsTcpClient.java
new file mode 100644
index 00000000..849d1eca
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/tcp/TopicsTcpClient.java
@@ -0,0 +1,98 @@
+package rs.iggy.clients.blocking.tcp;
+
+import io.netty.buffer.Unpooled;
+import rs.iggy.clients.blocking.TopicsClient;
+import rs.iggy.identifier.StreamId;
+import rs.iggy.identifier.TopicId;
+import rs.iggy.topic.CompressionAlgorithm;
+import rs.iggy.topic.Topic;
+import rs.iggy.topic.TopicDetails;
+import java.math.BigInteger;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Optional;
+import static rs.iggy.clients.blocking.tcp.BytesDeserializer.readTopic;
+import static rs.iggy.clients.blocking.tcp.BytesDeserializer.readTopicDetails;
+import static rs.iggy.clients.blocking.tcp.BytesSerializer.*;
+
+class TopicsTcpClient implements TopicsClient {
+
+ private static final int GET_TOPIC_CODE = 300;
+ private static final int GET_TOPICS_CODE = 301;
+ private static final int CREATE_TOPIC_CODE = 302;
+ private static final int DELETE_TOPIC_CODE = 303;
+ private static final int UPDATE_TOPIC_CODE = 304;
+ private static final int PURGE_TOPIC_CODE = 305;
+ private final InternalTcpClient tcpClient;
+
+ TopicsTcpClient(InternalTcpClient tcpClient) {
+ this.tcpClient = tcpClient;
+ }
+
+ @Override
+ public Optional<TopicDetails> getTopic(StreamId streamId, TopicId topicId) {
+ var payload = toBytes(streamId);
+ payload.writeBytes(toBytes(topicId));
+ var response = tcpClient.send(GET_TOPIC_CODE, payload);
+ if (response.isReadable()) {
+ return Optional.of(readTopicDetails(response));
+ }
+ return Optional.empty();
+ }
+
+ @Override
+ public List<Topic> getTopics(StreamId streamId) {
+ var payload = toBytes(streamId);
+ var response = tcpClient.send(GET_TOPICS_CODE, payload);
+ List<Topic> topics = new ArrayList<>();
+ while (response.isReadable()) {
+ topics.add(readTopic(response));
+ }
+ return topics;
+ }
+
+ @Override
+ public TopicDetails createTopic(StreamId streamId, Optional<Long> topicId, Long partitionsCount, CompressionAlgorithm compressionAlgorithm, BigInteger messageExpiry, BigInteger maxTopicSize, Optional<Short> replicationFactor, String name) {
+ var streamIdBytes = toBytes(streamId);
+ var payload = Unpooled.buffer(27 + streamIdBytes.readableBytes() + name.length());
+
+ payload.writeBytes(streamIdBytes);
+ payload.writeIntLE(topicId.orElse(0L).intValue());
+ payload.writeIntLE(partitionsCount.intValue());
+ payload.writeByte(compressionAlgorithm.asCode());
+ payload.writeBytes(toBytesAsU64(messageExpiry));
+ payload.writeBytes(toBytesAsU64(maxTopicSize));
+ payload.writeByte(replicationFactor.orElse((short) 0));
+ payload.writeBytes(nameToBytes(name));
+
+ var response = tcpClient.send(CREATE_TOPIC_CODE, payload);
+ return readTopicDetails(response);
+ }
+
+ @Override
+ public void updateTopic(StreamId streamId,
+ TopicId topicId,
+ CompressionAlgorithm compressionAlgorithm,
+ BigInteger messageExpiry,
+ BigInteger maxTopicSize,
+ Optional<Short> replicationFactor,
+ String name) {
+ var payload = Unpooled.buffer();
+ payload.writeBytes(toBytes(streamId));
+ payload.writeBytes(toBytes(topicId));
+ payload.writeByte(compressionAlgorithm.asCode());
+ payload.writeBytes(toBytesAsU64(messageExpiry));
+ payload.writeBytes(toBytesAsU64(maxTopicSize));
+ payload.writeByte(replicationFactor.orElse((short) 0));
+ payload.writeBytes(nameToBytes(name));
+
+ tcpClient.send(UPDATE_TOPIC_CODE, payload);
+ }
+
+ @Override
+ public void deleteTopic(StreamId streamId, TopicId topicId) {
+ var payload = toBytes(streamId);
+ payload.writeBytes(toBytes(topicId));
+ tcpClient.send(DELETE_TOPIC_CODE, payload);
+ }
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/tcp/UsersTcpClient.java b/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/tcp/UsersTcpClient.java
new file mode 100644
index 00000000..41ec6aab
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/tcp/UsersTcpClient.java
@@ -0,0 +1,137 @@
+package rs.iggy.clients.blocking.tcp;
+
+import io.netty.buffer.Unpooled;
+import rs.iggy.clients.blocking.UsersClient;
+import rs.iggy.identifier.UserId;
+import rs.iggy.user.*;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Optional;
+import static rs.iggy.clients.blocking.tcp.BytesDeserializer.readUserInfoDetails;
+import static rs.iggy.clients.blocking.tcp.BytesSerializer.nameToBytes;
+import static rs.iggy.clients.blocking.tcp.BytesSerializer.toBytes;
+
+class UsersTcpClient implements UsersClient {
+
+ private static final int GET_USER_CODE = 31;
+ private static final int GET_USERS_CODE = 32;
+ private static final int CREATE_USER_CODE = 33;
+ private static final int DELETE_USER_CODE = 34;
+ private static final int UPDATE_USER_CODE = 35;
+ private static final int UPDATE_PERMISSIONS_CODE = 36;
+ private static final int CHANGE_PASSWORD_CODE = 37;
+ private static final int LOGIN_USER_CODE = 38;
+ private static final int LOGOUT_USER_CODE = 39;
+
+ private final InternalTcpClient tcpClient;
+
+ UsersTcpClient(InternalTcpClient tcpClient) {
+ this.tcpClient = tcpClient;
+ }
+
+ @Override
+ public Optional<UserInfoDetails> getUser(UserId userId) {
+ var payload = toBytes(userId);
+ var response = tcpClient.send(GET_USER_CODE, payload);
+ if (response.isReadable()) {
+ return Optional.of(readUserInfoDetails(response));
+ }
+ return Optional.empty();
+ }
+
+ @Override
+ public List<UserInfo> getUsers() {
+ var response = tcpClient.send(GET_USERS_CODE);
+ List<UserInfo> users = new ArrayList<>();
+ while (response.isReadable()) {
+ users.add(BytesDeserializer.readUserInfo(response));
+ }
+ return users;
+ }
+
+ @Override
+ public UserInfoDetails createUser(String username, String password, UserStatus status, Optional<Permissions> permissions) {
+ var payload = Unpooled.buffer();
+ payload.writeBytes(nameToBytes(username));
+ payload.writeBytes(nameToBytes(password));
+ payload.writeByte(status.asCode());
+ permissions.ifPresentOrElse(perms -> {
+ payload.writeByte(1);
+ var permissionBytes = toBytes(perms);
+ payload.writeIntLE(permissionBytes.readableBytes());
+ payload.writeBytes(permissionBytes);
+ }, () -> payload.writeByte(0));
+
+ var response = tcpClient.send(CREATE_USER_CODE, payload);
+ return readUserInfoDetails(response);
+ }
+
+ @Override
+ public void deleteUser(UserId userId) {
+ var payload = toBytes(userId);
+ tcpClient.send(DELETE_USER_CODE, payload);
+ }
+
+ @Override
+ public void updateUser(UserId userId, Optional<String> usernameOptional, Optional<UserStatus> statusOptional) {
+ var payload = toBytes(userId);
+ usernameOptional.ifPresentOrElse((username) -> {
+ payload.writeByte(1);
+ payload.writeBytes(nameToBytes(username));
+ }, () -> payload.writeByte(0));
+ statusOptional.ifPresentOrElse((status) -> {
+ payload.writeByte(1);
+ payload.writeByte(status.asCode());
+ }, () -> payload.writeByte(0));
+
+ tcpClient.send(UPDATE_USER_CODE, payload);
+ }
+
+ @Override
+ public void updatePermissions(UserId userId, Optional<Permissions> permissionsOptional) {
+ var payload = toBytes(userId);
+
+ permissionsOptional.ifPresentOrElse(permissions -> {
+ payload.writeByte(1);
+ var permissionBytes = toBytes(permissions);
+ payload.writeIntLE(permissionBytes.readableBytes());
+ payload.writeBytes(permissionBytes);
+ }, () -> payload.writeByte(0));
+
+ tcpClient.send(UPDATE_PERMISSIONS_CODE, payload);
+ }
+
+ @Override
+ public void changePassword(UserId userId, String currentPassword, String newPassword) {
+ var payload = toBytes(userId);
+ payload.writeBytes(nameToBytes(currentPassword));
+ payload.writeBytes(nameToBytes(newPassword));
+
+ tcpClient.send(CHANGE_PASSWORD_CODE, payload);
+ }
+
+ @Override
+ public IdentityInfo login(String username, String password) {
+ String version = "0.6.30";
+ String context = "java-sdk";
+ var payloadSize = 2 + username.length() + password.length() + 4 + version.length() + 4 + context.length();
+ var payload = Unpooled.buffer(payloadSize);
+
+ payload.writeBytes(nameToBytes(username));
+ payload.writeBytes(nameToBytes(password));
+ payload.writeIntLE(version.length());
+ payload.writeBytes(version.getBytes());
+ payload.writeIntLE(context.length());
+ payload.writeBytes(context.getBytes());
+
+ var response = tcpClient.send(LOGIN_USER_CODE, payload);
+
+ var userId = response.readUnsignedIntLE();
+ return new IdentityInfo(userId, Optional.empty());
+ }
+
+ @Override
+ public void logout() {
+ tcpClient.send(LOGOUT_USER_CODE);
+ }
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/tcp/package-info.java b/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/tcp/package-info.java
new file mode 100644
index 00000000..cd4cb35e
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/clients/blocking/tcp/package-info.java
@@ -0,0 +1,4 @@
+@NonNullApi
+package rs.iggy.clients.blocking.tcp;
+
+import rs.iggy.NonNullApi;
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/consumergroup/Consumer.java b/foreign/java/java-sdk/src/main/java/rs/iggy/consumergroup/Consumer.java
new file mode 100644
index 00000000..24aa51d5
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/consumergroup/Consumer.java
@@ -0,0 +1,38 @@
+package rs.iggy.consumergroup;
+
+import rs.iggy.identifier.ConsumerId;
+
+public record Consumer(Kind kind, ConsumerId id) {
+
+ public static Consumer of(Long id) {
+ return new Consumer(Kind.Consumer, ConsumerId.of(id));
+ }
+
+ public static Consumer of(ConsumerId id) {
+ return new Consumer(Kind.Consumer, id);
+ }
+
+ public static Consumer group(Long id) {
+ return new Consumer(Kind.ConsumerGroup, ConsumerId.of(id));
+ }
+
+ public static Consumer group(ConsumerId id) {
+ return new Consumer(Kind.ConsumerGroup, id);
+ }
+
+ public enum Kind {
+ Consumer(1),
+ ConsumerGroup(2);
+
+ private final int code;
+
+ Kind(int code) {
+ this.code = code;
+ }
+
+ public int asCode() {
+ return code;
+ }
+
+ }
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/consumergroup/ConsumerGroup.java b/foreign/java/java-sdk/src/main/java/rs/iggy/consumergroup/ConsumerGroup.java
new file mode 100644
index 00000000..482f9afd
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/consumergroup/ConsumerGroup.java
@@ -0,0 +1,9 @@
+package rs.iggy.consumergroup;
+
+public record ConsumerGroup(
+ Long id,
+ String name,
+ Long partitionsCount,
+ Long membersCount
+) {
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/consumergroup/ConsumerGroupDetails.java b/foreign/java/java-sdk/src/main/java/rs/iggy/consumergroup/ConsumerGroupDetails.java
new file mode 100644
index 00000000..31d9a498
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/consumergroup/ConsumerGroupDetails.java
@@ -0,0 +1,19 @@
+package rs.iggy.consumergroup;
+
+import java.util.List;
+
+public record ConsumerGroupDetails(
+ Long id,
+ String name,
+ Long partitionsCount,
+ Long membersCount,
+ List<ConsumerGroupMember> members
+) {
+ public ConsumerGroupDetails(ConsumerGroup consumerGroup, List<ConsumerGroupMember> members) {
+ this(consumerGroup.id(),
+ consumerGroup.name(),
+ consumerGroup.partitionsCount(),
+ consumerGroup.membersCount(),
+ members);
+ }
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/consumergroup/ConsumerGroupMember.java b/foreign/java/java-sdk/src/main/java/rs/iggy/consumergroup/ConsumerGroupMember.java
new file mode 100644
index 00000000..66684191
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/consumergroup/ConsumerGroupMember.java
@@ -0,0 +1,10 @@
+package rs.iggy.consumergroup;
+
+import java.util.List;
+
+public record ConsumerGroupMember(
+ Long id,
+ Long partitionsCount,
+ List<Long> partitions
+) {
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/consumergroup/package-info.java b/foreign/java/java-sdk/src/main/java/rs/iggy/consumergroup/package-info.java
new file mode 100644
index 00000000..8094edca
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/consumergroup/package-info.java
@@ -0,0 +1,4 @@
+@NonNullApi
+package rs.iggy.consumergroup;
+
+import rs.iggy.NonNullApi;
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/consumeroffset/ConsumerOffsetInfo.java b/foreign/java/java-sdk/src/main/java/rs/iggy/consumeroffset/ConsumerOffsetInfo.java
new file mode 100644
index 00000000..eac4e3db
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/consumeroffset/ConsumerOffsetInfo.java
@@ -0,0 +1,10 @@
+package rs.iggy.consumeroffset;
+
+import java.math.BigInteger;
+
+public record ConsumerOffsetInfo(
+ Long partitionId,
+ BigInteger currentOffset,
+ BigInteger storedOffset
+) {
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/consumeroffset/package-info.java b/foreign/java/java-sdk/src/main/java/rs/iggy/consumeroffset/package-info.java
new file mode 100644
index 00000000..7527d514
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/consumeroffset/package-info.java
@@ -0,0 +1,4 @@
+@NonNullApi
+package rs.iggy.consumeroffset;
+
+import rs.iggy.NonNullApi;
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/identifier/ConsumerId.java b/foreign/java/java-sdk/src/main/java/rs/iggy/identifier/ConsumerId.java
new file mode 100644
index 00000000..6aedd1da
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/identifier/ConsumerId.java
@@ -0,0 +1,19 @@
+package rs.iggy.identifier;
+
+import javax.annotation.Nullable;
+
+public class ConsumerId extends Identifier {
+
+ private ConsumerId(@Nullable String name, @Nullable Long id) {
+ super(name, id);
+ }
+
+ public static ConsumerId of(String name) {
+ return new ConsumerId(name, null);
+ }
+
+ public static ConsumerId of(Long id) {
+ return new ConsumerId(null, id);
+ }
+
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/identifier/Identifier.java b/foreign/java/java-sdk/src/main/java/rs/iggy/identifier/Identifier.java
new file mode 100644
index 00000000..c24a3fb8
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/identifier/Identifier.java
@@ -0,0 +1,51 @@
+package rs.iggy.identifier;
+
+import org.apache.commons.lang3.StringUtils;
+import javax.annotation.Nullable;
+
+public abstract class Identifier {
+
+ private final String name;
+ private final Long id;
+
+ protected Identifier(@Nullable String name, @Nullable Long id) {
+ if (StringUtils.isBlank(name) && id == null) {
+ throw new IllegalArgumentException("Name and id cannot be blank");
+ }
+ if (StringUtils.isNotBlank(name) && id != null) {
+ throw new IllegalArgumentException("Name and id cannot be both present");
+ }
+ if (StringUtils.isNotBlank(name)) {
+ this.name = name;
+ this.id = null;
+ } else {
+ this.name = null;
+ this.id = id;
+ }
+ }
+
+ @Override
+ public String toString() {
+ if (StringUtils.isNotBlank(name)) {
+ return name;
+ }
+ return id.toString();
+ }
+
+ public int getKind() {
+ if (id != null) {
+ return 1;
+ }
+ return 2;
+ }
+
+ public Long getId() {
+ return id;
+ }
+
+ public String getName() {
+ return name;
+ }
+
+}
+
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/identifier/StreamId.java b/foreign/java/java-sdk/src/main/java/rs/iggy/identifier/StreamId.java
new file mode 100644
index 00000000..c8f4ce89
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/identifier/StreamId.java
@@ -0,0 +1,19 @@
+package rs.iggy.identifier;
+
+import javax.annotation.Nullable;
+
+public class StreamId extends Identifier {
+
+ private StreamId(@Nullable String name, @Nullable Long id) {
+ super(name, id);
+ }
+
+ public static StreamId of(String name) {
+ return new StreamId(name, null);
+ }
+
+ public static StreamId of(Long id) {
+ return new StreamId(null, id);
+ }
+
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/identifier/TopicId.java b/foreign/java/java-sdk/src/main/java/rs/iggy/identifier/TopicId.java
new file mode 100644
index 00000000..af8e2dca
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/identifier/TopicId.java
@@ -0,0 +1,19 @@
+package rs.iggy.identifier;
+
+import javax.annotation.Nullable;
+
+public class TopicId extends Identifier {
+
+ private TopicId(@Nullable String name, @Nullable Long id) {
+ super(name, id);
+ }
+
+ public static TopicId of(String name) {
+ return new TopicId(name, null);
+ }
+
+ public static TopicId of(Long id) {
+ return new TopicId(null, id);
+ }
+
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/identifier/UserId.java b/foreign/java/java-sdk/src/main/java/rs/iggy/identifier/UserId.java
new file mode 100644
index 00000000..da8e1a3f
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/identifier/UserId.java
@@ -0,0 +1,19 @@
+package rs.iggy.identifier;
+
+import javax.annotation.Nullable;
+
+public class UserId extends Identifier {
+
+ private UserId(@Nullable String name, @Nullable Long id) {
+ super(name, id);
+ }
+
+ public static UserId of(String name) {
+ return new UserId(name, null);
+ }
+
+ public static UserId of(Long id) {
+ return new UserId(null, id);
+ }
+
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/identifier/package-info.java b/foreign/java/java-sdk/src/main/java/rs/iggy/identifier/package-info.java
new file mode 100644
index 00000000..7a7ed98a
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/identifier/package-info.java
@@ -0,0 +1,4 @@
+@NonNullApi
+package rs.iggy.identifier;
+
+import rs.iggy.NonNullApi;
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/message/BigIntegerMessageId.java b/foreign/java/java-sdk/src/main/java/rs/iggy/message/BigIntegerMessageId.java
new file mode 100644
index 00000000..66f00646
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/message/BigIntegerMessageId.java
@@ -0,0 +1,34 @@
+package rs.iggy.message;
+
+import io.netty.buffer.ByteBuf;
+import rs.iggy.clients.blocking.tcp.BytesSerializer;
+import java.math.BigInteger;
+
+public class BigIntegerMessageId implements MessageId {
+
+ private static final BigIntegerMessageId DEFAULT_ID = new BigIntegerMessageId(BigInteger.ZERO);
+ private final BigInteger value;
+
+ public BigIntegerMessageId(BigInteger value) {
+ this.value = value;
+ }
+
+ public static BigIntegerMessageId defaultId() {
+ return DEFAULT_ID;
+ }
+
+ @Override
+ public BigInteger toBigInteger() {
+ return value;
+ }
+
+ public ByteBuf toBytes() {
+ return BytesSerializer.toBytesAsU128(value);
+ }
+
+ @Override
+ public String toString() {
+ return value.toString();
+ }
+
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/message/BytesMessageId.java b/foreign/java/java-sdk/src/main/java/rs/iggy/message/BytesMessageId.java
new file mode 100644
index 00000000..af47f0a9
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/message/BytesMessageId.java
@@ -0,0 +1,34 @@
+package rs.iggy.message;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import java.math.BigInteger;
+import java.util.Arrays;
+
+public class BytesMessageId implements MessageId {
+
+ private final byte[] value;
+
+ public BytesMessageId(byte[] value) {
+ if (value.length != 16) {
+ throw new IllegalArgumentException("Message id must have 16 bytes");
+ }
+ this.value = value;
+ }
+
+ @Override
+ public BigInteger toBigInteger() {
+ return new BigInteger(1, value);
+ }
+
+ @Override
+ public ByteBuf toBytes() {
+ return Unpooled.wrappedBuffer(value);
+ }
+
+ @Override
+ public String toString() {
+ return Arrays.toString(value);
+ }
+
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/message/HeaderKind.java b/foreign/java/java-sdk/src/main/java/rs/iggy/message/HeaderKind.java
new file mode 100644
index 00000000..a427b050
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/message/HeaderKind.java
@@ -0,0 +1,38 @@
+package rs.iggy.message;
+
+public enum HeaderKind {
+ Raw(1),
+ String(2),
+ Bool(3),
+ Int8(4),
+ Int16(5),
+ Int32(6),
+ Int64(7),
+ Int128(8),
+ Uint8(9),
+ Uint16(10),
+ Uint32(11),
+ Uint64(12),
+ Uint128(13),
+ Float32(14),
+ Float64(15);
+
+ private final int code;
+
+ HeaderKind(int code) {
+ this.code = code;
+ }
+
+ public static HeaderKind fromCode(int code) {
+ for (HeaderKind kind : values()) {
+ if (kind.code == code) {
+ return kind;
+ }
+ }
+ throw new IllegalArgumentException("Unknown header kind: " + code);
+ }
+
+ public int asCode() {
+ return code;
+ }
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/message/HeaderValue.java b/foreign/java/java-sdk/src/main/java/rs/iggy/message/HeaderValue.java
new file mode 100644
index 00000000..d5549477
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/message/HeaderValue.java
@@ -0,0 +1,4 @@
+package rs.iggy.message;
+
+public record HeaderValue(HeaderKind kind, String value) {
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/message/Message.java b/foreign/java/java-sdk/src/main/java/rs/iggy/message/Message.java
new file mode 100644
index 00000000..c94decd4
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/message/Message.java
@@ -0,0 +1,16 @@
+package rs.iggy.message;
+
+import java.util.Map;
+import java.util.Optional;
+
+public record Message(
+ MessageId id,
+ byte[] payload,
+ Optional<Map<String, HeaderValue>> headers
+) {
+
+ public static Message of(String payload) {
+ return new Message(MessageId.serverGenerated(), payload.getBytes(), Optional.empty());
+ }
+
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/message/MessageId.java b/foreign/java/java-sdk/src/main/java/rs/iggy/message/MessageId.java
new file mode 100644
index 00000000..388d1b7f
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/message/MessageId.java
@@ -0,0 +1,24 @@
+package rs.iggy.message;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonValue;
+import io.netty.buffer.ByteBuf;
+import java.math.BigInteger;
+
+public interface MessageId {
+
+ static MessageId serverGenerated() {
+ return BigIntegerMessageId.defaultId();
+ }
+
+ @JsonCreator
+ static MessageId from(BigInteger bigInteger) {
+ return new BigIntegerMessageId(bigInteger);
+ }
+
+ @JsonValue
+ BigInteger toBigInteger();
+
+ ByteBuf toBytes();
+
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/message/MessageState.java b/foreign/java/java-sdk/src/main/java/rs/iggy/message/MessageState.java
new file mode 100644
index 00000000..3a38fe6c
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/message/MessageState.java
@@ -0,0 +1,23 @@
+package rs.iggy.message;
+
+public enum MessageState {
+ Available(1),
+ Unavailable(10),
+ Poisoned(20),
+ MarkedForDeletion(30);
+
+ private final int code;
+
+ MessageState(int code) {
+ this.code = code;
+ }
+
+ public static MessageState fromCode(int code) {
+ for (MessageState state : MessageState.values()) {
+ if (state.code == code) {
+ return state;
+ }
+ }
+ throw new IllegalArgumentException("Unknown message state code: " + code);
+ }
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/message/Partitioning.java b/foreign/java/java-sdk/src/main/java/rs/iggy/message/Partitioning.java
new file mode 100644
index 00000000..32df7b22
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/message/Partitioning.java
@@ -0,0 +1,29 @@
+package rs.iggy.message;
+
+import org.apache.commons.lang3.ArrayUtils;
+import java.nio.ByteBuffer;
+
+public record Partitioning(
+ PartitioningKind kind,
+ byte[] value
+) {
+
+ public static Partitioning balanced() {
+ return new Partitioning(PartitioningKind.Balanced, new byte[]{});
+ }
+
+ public static Partitioning partitionId(Long id) {
+ ByteBuffer buffer = ByteBuffer.allocate(4);
+ buffer.putInt(id.intValue());
+ byte[] partitionId = buffer.array();
+ ArrayUtils.reverse(partitionId);
+ return new Partitioning(PartitioningKind.PartitionId, partitionId);
+ }
+
+ public static Partitioning messagesKey(String key) {
+ if (key == null || key.isBlank() || key.length() > 255) {
+ throw new IllegalArgumentException("Key must be non-empty and less than 255 characters long");
+ }
+ return new Partitioning(PartitioningKind.MessagesKey, key.getBytes());
+ }
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/message/PartitioningKind.java b/foreign/java/java-sdk/src/main/java/rs/iggy/message/PartitioningKind.java
new file mode 100644
index 00000000..aacd636f
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/message/PartitioningKind.java
@@ -0,0 +1,22 @@
+package rs.iggy.message;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+public enum PartitioningKind {
+ @JsonProperty("balanced")
+ Balanced(1),
+ @JsonProperty("partition_id")
+ PartitionId(2),
+ @JsonProperty("messages_key")
+ MessagesKey(3);
+
+ private final int code;
+
+ PartitioningKind(int code) {
+ this.code = code;
+ }
+
+ public int asCode() {
+ return code;
+ }
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/message/PolledMessage.java b/foreign/java/java-sdk/src/main/java/rs/iggy/message/PolledMessage.java
new file mode 100644
index 00000000..6becc4ea
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/message/PolledMessage.java
@@ -0,0 +1,16 @@
+package rs.iggy.message;
+
+import java.math.BigInteger;
+import java.util.Map;
+import java.util.Optional;
+
+public record PolledMessage(
+ BigInteger offset,
+ MessageState state,
+ BigInteger timestamp,
+ MessageId id,
+ Long checksum,
+ Optional<Map<String, HeaderValue>> headers,
+ byte[] payload
+) {
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/message/PolledMessages.java b/foreign/java/java-sdk/src/main/java/rs/iggy/message/PolledMessages.java
new file mode 100644
index 00000000..72263bf3
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/message/PolledMessages.java
@@ -0,0 +1,11 @@
+package rs.iggy.message;
+
+import java.math.BigInteger;
+import java.util.List;
+
+public record PolledMessages(
+ Long partitionId,
+ BigInteger currentOffset,
+ List<PolledMessage> messages
+) {
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/message/PollingKind.java b/foreign/java/java-sdk/src/main/java/rs/iggy/message/PollingKind.java
new file mode 100644
index 00000000..17c29ee7
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/message/PollingKind.java
@@ -0,0 +1,19 @@
+package rs.iggy.message;
+
+public enum PollingKind {
+ Offset(1),
+ Timestamp(2),
+ First(3),
+ Last(4),
+ Next(5);
+
+ private final int code;
+
+ PollingKind(int code) {
+ this.code = code;
+ }
+
+ public int asCode() {
+ return code;
+ }
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/message/PollingStrategy.java b/foreign/java/java-sdk/src/main/java/rs/iggy/message/PollingStrategy.java
new file mode 100644
index 00000000..87fa5112
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/message/PollingStrategy.java
@@ -0,0 +1,27 @@
+package rs.iggy.message;
+
+import java.math.BigInteger;
+
+public record PollingStrategy(PollingKind kind, BigInteger value) {
+
+ public static PollingStrategy offset(BigInteger value) {
+ return new PollingStrategy(PollingKind.Offset, value);
+ }
+
+ public static PollingStrategy timestamp(BigInteger value) {
+ return new PollingStrategy(PollingKind.Timestamp, value);
+ }
+
+ public static PollingStrategy first() {
+ return new PollingStrategy(PollingKind.First, BigInteger.ZERO);
+ }
+
+ public static PollingStrategy last() {
+ return new PollingStrategy(PollingKind.Last, BigInteger.ZERO);
+ }
+
+ public static PollingStrategy next() {
+ return new PollingStrategy(PollingKind.Next, BigInteger.ZERO);
+ }
+
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/message/UuidMessageId.java b/foreign/java/java-sdk/src/main/java/rs/iggy/message/UuidMessageId.java
new file mode 100644
index 00000000..49e0e5ca
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/message/UuidMessageId.java
@@ -0,0 +1,37 @@
+package rs.iggy.message;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import java.math.BigInteger;
+import java.nio.ByteBuffer;
+import java.util.UUID;
+
+public class UuidMessageId implements MessageId {
+
+ private final UUID value;
+
+ public UuidMessageId(UUID value) {
+ this.value = value;
+ }
+
+ @Override
+ public BigInteger toBigInteger() {
+ ByteBuffer buffer = ByteBuffer.wrap(new byte[16]);
+ buffer.putLong(value.getMostSignificantBits());
+ buffer.putLong(value.getLeastSignificantBits());
+ return new BigInteger(1, buffer.array());
+ }
+
+ public ByteBuf toBytes() {
+ var buffer = Unpooled.buffer(16, 16);
+ buffer.writeLongLE(value.getLeastSignificantBits());
+ buffer.writeLongLE(value.getMostSignificantBits());
+ return buffer;
+ }
+
+ @Override
+ public String toString() {
+ return value.toString();
+ }
+
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/message/package-info.java b/foreign/java/java-sdk/src/main/java/rs/iggy/message/package-info.java
new file mode 100644
index 00000000..286f0c8c
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/message/package-info.java
@@ -0,0 +1,4 @@
+@NonNullApi
+package rs.iggy.message;
+
+import rs.iggy.NonNullApi;
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/partition/Partition.java b/foreign/java/java-sdk/src/main/java/rs/iggy/partition/Partition.java
new file mode 100644
index 00000000..a46ade84
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/partition/Partition.java
@@ -0,0 +1,13 @@
+package rs.iggy.partition;
+
+import java.math.BigInteger;
+
+public record Partition(
+ Long id,
+ BigInteger createdAt,
+ Long segmentsCount,
+ BigInteger currentOffset,
+ String size,
+ BigInteger messagesCount
+) {
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/partition/package-info.java b/foreign/java/java-sdk/src/main/java/rs/iggy/partition/package-info.java
new file mode 100644
index 00000000..ed95b50d
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/partition/package-info.java
@@ -0,0 +1,4 @@
+@NonNullApi
+package rs.iggy.partition;
+
+import rs.iggy.NonNullApi;
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/personalaccesstoken/PersonalAccessTokenInfo.java b/foreign/java/java-sdk/src/main/java/rs/iggy/personalaccesstoken/PersonalAccessTokenInfo.java
new file mode 100644
index 00000000..87015827
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/personalaccesstoken/PersonalAccessTokenInfo.java
@@ -0,0 +1,7 @@
+package rs.iggy.personalaccesstoken;
+
+import java.math.BigInteger;
+import java.util.Optional;
+
+public record PersonalAccessTokenInfo(String name, Optional<BigInteger> expiryAt) {
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/personalaccesstoken/RawPersonalAccessToken.java b/foreign/java/java-sdk/src/main/java/rs/iggy/personalaccesstoken/RawPersonalAccessToken.java
new file mode 100644
index 00000000..7aab05a7
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/personalaccesstoken/RawPersonalAccessToken.java
@@ -0,0 +1,4 @@
+package rs.iggy.personalaccesstoken;
+
+public record RawPersonalAccessToken(String token) {
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/stream/StreamBase.java b/foreign/java/java-sdk/src/main/java/rs/iggy/stream/StreamBase.java
new file mode 100644
index 00000000..82b503aa
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/stream/StreamBase.java
@@ -0,0 +1,13 @@
+package rs.iggy.stream;
+
+import java.math.BigInteger;
+
+public record StreamBase(
+ Long id,
+ BigInteger createdAt,
+ String name,
+ String size,
+ BigInteger messagesCount,
+ Long topicsCount
+) {
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/stream/StreamDetails.java b/foreign/java/java-sdk/src/main/java/rs/iggy/stream/StreamDetails.java
new file mode 100644
index 00000000..54b14dff
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/stream/StreamDetails.java
@@ -0,0 +1,19 @@
+package rs.iggy.stream;
+
+import rs.iggy.topic.Topic;
+import java.math.BigInteger;
+import java.util.List;
+
+public record StreamDetails(
+ Long id,
+ BigInteger createdAt,
+ String name,
+ String size,
+ BigInteger messagesCount,
+ Long topicsCount,
+ List<Topic> topics
+) {
+ public StreamDetails(StreamBase base, List<Topic> topics) {
+ this(base.id(), base.createdAt(), base.name(), base.size(), base.messagesCount(), base.topicsCount(), topics);
+ }
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/stream/package-info.java b/foreign/java/java-sdk/src/main/java/rs/iggy/stream/package-info.java
new file mode 100644
index 00000000..51587d25
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/stream/package-info.java
@@ -0,0 +1,4 @@
+@NonNullApi
+package rs.iggy.stream;
+
+import rs.iggy.NonNullApi;
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/system/ClientInfo.java b/foreign/java/java-sdk/src/main/java/rs/iggy/system/ClientInfo.java
new file mode 100644
index 00000000..de5cbcce
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/system/ClientInfo.java
@@ -0,0 +1,12 @@
+package rs.iggy.system;
+
+import java.util.Optional;
+
+public record ClientInfo(
+ Long clientId,
+ Optional<Long> userId,
+ String address,
+ String transport,
+ Long consumerGroupsCount
+) {
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/system/ClientInfoDetails.java b/foreign/java/java-sdk/src/main/java/rs/iggy/system/ClientInfoDetails.java
new file mode 100644
index 00000000..9714408f
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/system/ClientInfoDetails.java
@@ -0,0 +1,18 @@
+package rs.iggy.system;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Optional;
+
+public record ClientInfoDetails(
+ Long clientId,
+ Optional<Long> userId,
+ String address,
+ String transport,
+ Long consumerGroupsCount,
+ List<ConsumerGroupInfo> consumerGroups
+) {
+ public ClientInfoDetails(ClientInfo clientInfo, ArrayList<ConsumerGroupInfo> consumerGroups) {
+ this(clientInfo.clientId(), clientInfo.userId(), clientInfo.address(), clientInfo.transport(), clientInfo.consumerGroupsCount(), consumerGroups);
+ }
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/system/ConsumerGroupInfo.java b/foreign/java/java-sdk/src/main/java/rs/iggy/system/ConsumerGroupInfo.java
new file mode 100644
index 00000000..04024205
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/system/ConsumerGroupInfo.java
@@ -0,0 +1,8 @@
+package rs.iggy.system;
+
+public record ConsumerGroupInfo(
+ Long streamId,
+ Long topicId,
+ Long consumerGroupId
+) {
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/system/Stats.java b/foreign/java/java-sdk/src/main/java/rs/iggy/system/Stats.java
new file mode 100644
index 00000000..1c10798d
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/system/Stats.java
@@ -0,0 +1,29 @@
+package rs.iggy.system;
+
+import java.math.BigInteger;
+
+public record Stats(
+ Long processId,
+ Float cpuUsage,
+ Float totalCpuUsage,
+ String memoryUsage,
+ String totalMemory,
+ String availableMemory,
+ BigInteger runTime,
+ BigInteger startTime,
+ String readBytes,
+ String writtenBytes,
+ String messagesSizeBytes,
+ Long streamsCount,
+ Long topicsCount,
+ Long partitionsCount,
+ Long segmentsCount,
+ BigInteger messagesCount,
+ Long clientsCount,
+ Long consumerGroupsCount,
+ String hostname,
+ String osName,
+ String osVersion,
+ String kernelVersion
+) {
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/system/package-info.java b/foreign/java/java-sdk/src/main/java/rs/iggy/system/package-info.java
new file mode 100644
index 00000000..36e94130
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/system/package-info.java
@@ -0,0 +1,4 @@
+@NonNullApi
+package rs.iggy.system;
+
+import rs.iggy.NonNullApi;
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/topic/CompressionAlgorithm.java b/foreign/java/java-sdk/src/main/java/rs/iggy/topic/CompressionAlgorithm.java
new file mode 100644
index 00000000..ca404ab1
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/topic/CompressionAlgorithm.java
@@ -0,0 +1,25 @@
+package rs.iggy.topic;
+
+public enum CompressionAlgorithm {
+ None(1),
+ Gzip(2);
+
+ private final Integer code;
+
+ CompressionAlgorithm(Integer code) {
+ this.code = code;
+ }
+
+ public static CompressionAlgorithm fromCode(byte code) {
+ for (CompressionAlgorithm algorithm : values()) {
+ if (algorithm.code == code) {
+ return algorithm;
+ }
+ }
+ throw new IllegalArgumentException();
+ }
+
+ public Integer asCode() {
+ return code;
+ }
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/topic/Topic.java b/foreign/java/java-sdk/src/main/java/rs/iggy/topic/Topic.java
new file mode 100644
index 00000000..b59d26ff
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/topic/Topic.java
@@ -0,0 +1,17 @@
+package rs.iggy.topic;
+
+import java.math.BigInteger;
+
+public record Topic(
+ Long id,
+ BigInteger createdAt,
+ String name,
+ String size,
+ BigInteger messageExpiry,
+ CompressionAlgorithm compressionAlgorithm,
+ BigInteger maxTopicSize,
+ Short replicationFactor,
+ BigInteger messagesCount,
+ Long partitionsCount
+) {
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/topic/TopicDetails.java b/foreign/java/java-sdk/src/main/java/rs/iggy/topic/TopicDetails.java
new file mode 100644
index 00000000..135470cb
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/topic/TopicDetails.java
@@ -0,0 +1,33 @@
+package rs.iggy.topic;
+
+import rs.iggy.partition.Partition;
+import java.math.BigInteger;
+import java.util.List;
+
+public record TopicDetails(
+ Long id,
+ BigInteger createdAt,
+ String name,
+ String size,
+ BigInteger messageExpiry,
+ CompressionAlgorithm compressionAlgorithm,
+ BigInteger maxTopicSize,
+ Short replicationFactor,
+ BigInteger messagesCount,
+ Long partitionsCount,
+ List<Partition> partitions
+) {
+ public TopicDetails(Topic topic, List<Partition> partitions) {
+ this(topic.id(),
+ topic.createdAt(),
+ topic.name(),
+ topic.size(),
+ topic.messageExpiry(),
+ topic.compressionAlgorithm(),
+ topic.maxTopicSize(),
+ topic.replicationFactor(),
+ topic.messagesCount(),
+ topic.partitionsCount(),
+ partitions);
+ }
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/topic/package-info.java b/foreign/java/java-sdk/src/main/java/rs/iggy/topic/package-info.java
new file mode 100644
index 00000000..b472f8c3
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/topic/package-info.java
@@ -0,0 +1,4 @@
+@NonNullApi
+package rs.iggy.topic;
+
+import rs.iggy.NonNullApi;
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/user/GlobalPermissions.java b/foreign/java/java-sdk/src/main/java/rs/iggy/user/GlobalPermissions.java
new file mode 100644
index 00000000..48edb005
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/user/GlobalPermissions.java
@@ -0,0 +1,15 @@
+package rs.iggy.user;
+
+public record GlobalPermissions(
+ boolean manageServers,
+ boolean readServers,
+ boolean manageUsers,
+ boolean readUsers,
+ boolean manageStreams,
+ boolean readStreams,
+ boolean manageTopics,
+ boolean readTopics,
+ boolean pollMessages,
+ boolean sendMessages
+) {
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/user/IdentityInfo.java b/foreign/java/java-sdk/src/main/java/rs/iggy/user/IdentityInfo.java
new file mode 100644
index 00000000..4e201684
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/user/IdentityInfo.java
@@ -0,0 +1,6 @@
+package rs.iggy.user;
+
+import java.util.Optional;
+
+public record IdentityInfo(Long userId, Optional<TokenInfo> accessToken) {
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/user/Permissions.java b/foreign/java/java-sdk/src/main/java/rs/iggy/user/Permissions.java
new file mode 100644
index 00000000..991a6224
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/user/Permissions.java
@@ -0,0 +1,12 @@
+package rs.iggy.user;
+
+import com.fasterxml.jackson.annotation.JsonSetter;
+import com.fasterxml.jackson.annotation.Nulls;
+import java.util.Map;
+
+public record Permissions(
+ GlobalPermissions global,
+ @JsonSetter(nulls = Nulls.AS_EMPTY)
+ Map<Long, StreamPermissions> streams
+) {
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/user/StreamPermissions.java b/foreign/java/java-sdk/src/main/java/rs/iggy/user/StreamPermissions.java
new file mode 100644
index 00000000..886a502d
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/user/StreamPermissions.java
@@ -0,0 +1,17 @@
+package rs.iggy.user;
+
+import com.fasterxml.jackson.annotation.JsonSetter;
+import com.fasterxml.jackson.annotation.Nulls;
+import java.util.Map;
+
+public record StreamPermissions(
+ boolean manageStream,
+ boolean readStream,
+ boolean manageTopics,
+ boolean readTopics,
+ boolean pollMessages,
+ boolean sendMessages,
+ @JsonSetter(nulls = Nulls.AS_EMPTY)
+ Map<Long, TopicPermissions> topics
+) {
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/user/TokenInfo.java b/foreign/java/java-sdk/src/main/java/rs/iggy/user/TokenInfo.java
new file mode 100644
index 00000000..280db972
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/user/TokenInfo.java
@@ -0,0 +1,6 @@
+package rs.iggy.user;
+
+import java.math.BigInteger;
+
+public record TokenInfo(String token, BigInteger expiry) {
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/user/TopicPermissions.java b/foreign/java/java-sdk/src/main/java/rs/iggy/user/TopicPermissions.java
new file mode 100644
index 00000000..d5f76cdb
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/user/TopicPermissions.java
@@ -0,0 +1,9 @@
+package rs.iggy.user;
+
+public record TopicPermissions(
+ boolean manageTopic,
+ boolean readTopic,
+ boolean pollMessages,
+ boolean sendMessages
+) {
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/user/UserInfo.java b/foreign/java/java-sdk/src/main/java/rs/iggy/user/UserInfo.java
new file mode 100644
index 00000000..c16f86be
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/user/UserInfo.java
@@ -0,0 +1,11 @@
+package rs.iggy.user;
+
+import java.math.BigInteger;
+
+public record UserInfo(
+ Long id,
+ BigInteger createdAt,
+ UserStatus status,
+ String username
+) {
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/user/UserInfoDetails.java b/foreign/java/java-sdk/src/main/java/rs/iggy/user/UserInfoDetails.java
new file mode 100644
index 00000000..ac444b9d
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/user/UserInfoDetails.java
@@ -0,0 +1,16 @@
+package rs.iggy.user;
+
+import java.math.BigInteger;
+import java.util.Optional;
+
+public record UserInfoDetails(
+ Long id,
+ BigInteger createdAt,
+ UserStatus status,
+ String username,
+ Optional<Permissions> permissions
+) {
+ public UserInfoDetails(UserInfo userInfo, Optional<Permissions> permissions) {
+ this(userInfo.id(), userInfo.createdAt(), userInfo.status(), userInfo.username(), permissions);
+ }
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/user/UserStatus.java b/foreign/java/java-sdk/src/main/java/rs/iggy/user/UserStatus.java
new file mode 100644
index 00000000..2c58e33d
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/user/UserStatus.java
@@ -0,0 +1,29 @@
+package rs.iggy.user;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+public enum UserStatus {
+ @JsonProperty("active")
+ Active(1),
+ @JsonProperty("inactive")
+ Inactive(2);
+
+ private final int code;
+
+ UserStatus(int code) {
+ this.code = code;
+ }
+
+ public static UserStatus fromCode(int code) {
+ for (UserStatus userStatus : UserStatus.values()) {
+ if (userStatus.code == code) {
+ return userStatus;
+ }
+ }
+ throw new IllegalArgumentException("Invalid user status: " + code);
+ }
+
+ public int asCode() {
+ return code;
+ }
+}
diff --git a/foreign/java/java-sdk/src/main/java/rs/iggy/user/package-info.java b/foreign/java/java-sdk/src/main/java/rs/iggy/user/package-info.java
new file mode 100644
index 00000000..625d3370
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/rs/iggy/user/package-info.java
@@ -0,0 +1,4 @@
+@NonNullApi
+package rs.iggy.user;
+
+import rs.iggy.NonNullApi;
diff --git a/foreign/java/java-sdk/src/test/java/rs/iggy/IggyTest.java b/foreign/java/java-sdk/src/test/java/rs/iggy/IggyTest.java
new file mode 100644
index 00000000..dd197904
--- /dev/null
+++ b/foreign/java/java-sdk/src/test/java/rs/iggy/IggyTest.java
@@ -0,0 +1,20 @@
+package rs.iggy;
+
+import org.junit.jupiter.api.Test;
+import rs.iggy.clients.blocking.http.IggyHttpClient;
+import static org.assertj.core.api.Assertions.assertThat;
+
+class IggyTest {
+
+ @Test
+ void test() {
+ var baseClient = new IggyHttpClient("http://localhost:8080");
+
+ var client = Iggy.clientBuilder()
+ .withBaseClient(baseClient)
+ .build();
+
+ assertThat(client).isNotNull();
+ }
+
+}
diff --git a/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/ConsumerGroupsClientBaseTest.java b/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/ConsumerGroupsClientBaseTest.java
new file mode 100644
index 00000000..ee715644
--- /dev/null
+++ b/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/ConsumerGroupsClientBaseTest.java
@@ -0,0 +1,90 @@
+package rs.iggy.clients.blocking;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import rs.iggy.consumergroup.ConsumerGroup;
+import rs.iggy.identifier.ConsumerId;
+import rs.iggy.identifier.StreamId;
+import rs.iggy.identifier.TopicId;
+import java.util.Optional;
+import static org.assertj.core.api.Assertions.assertThat;
+
+public abstract class ConsumerGroupsClientBaseTest extends IntegrationTest {
+
+ protected ConsumerGroupsClient consumerGroupsClient;
+
+ @BeforeEach
+ void beforeEachBase() {
+ consumerGroupsClient = client.consumerGroups();
+
+ login();
+ }
+
+ @Test
+ void shouldCreateAndDeleteConsumerGroup() {
+ // given
+ setUpStreamAndTopic();
+
+ // when
+ consumerGroupsClient.createConsumerGroup(42L,
+ 42L,
+ Optional.of(42L),
+ "consumer-group-42");
+
+ var consumerGroupById = consumerGroupsClient.getConsumerGroup(StreamId.of(42L),
+ TopicId.of(42L),
+ ConsumerId.of(42L));
+ var consumerGroupByName = consumerGroupsClient.getConsumerGroup(StreamId.of(42L),
+ TopicId.of(42L),
+ ConsumerId.of("consumer-group-42"));
+
+ // then
+ assertThat(consumerGroupById).isPresent();
+ assertThat(consumerGroupById.get().id()).isEqualTo(42L);
+ assertThat(consumerGroupById.get().name()).isEqualTo("consumer-group-42");
+ assertThat(consumerGroupById).isEqualTo(consumerGroupByName);
+
+ // when
+ consumerGroupsClient.deleteConsumerGroup(42L, 42L, 42L);
+
+ // then
+ assertThat(consumerGroupsClient.getConsumerGroups(42L, 42L)).isEmpty();
+ assertThat(consumerGroupsClient.getConsumerGroup(42L, 42L, 42L)).isEmpty();
+ }
+
+ @Test
+ void shouldDeleteConsumerGroupByName() {
+ // given
+ setUpStreamAndTopic();
+ consumerGroupsClient.createConsumerGroup(42L, 42L, Optional.of(42L), "consumer-group-42");
+ var consumerGroup = consumerGroupsClient.getConsumerGroup(42L, 42L, 42L);
+ assert consumerGroup.isPresent();
+
+ // when
+ consumerGroupsClient.deleteConsumerGroup(StreamId.of(42L), TopicId.of(42L),
+ ConsumerId.of("consumer-group-42"));
+
+ // then
+ assertThat(consumerGroupsClient.getConsumerGroups(42L, 42L)).isEmpty();
+ }
+
+ @Test
+ void shouldGetAllConsumerGroups() {
+ // given
+ setUpStreamAndTopic();
+
+ consumerGroupsClient.createConsumerGroup(42L, 42L, Optional.of(42L), "consumer-group-42");
+ consumerGroupsClient.createConsumerGroup(42L, 42L, Optional.of(43L), "consumer-group-43");
+ consumerGroupsClient.createConsumerGroup(42L, 42L, Optional.of(44L), "consumer-group-44");
+
+ // when
+ var consumerGroups = consumerGroupsClient.getConsumerGroups(42L, 42L);
+
+ // then
+ assertThat(consumerGroups).hasSize(3);
+ assertThat(consumerGroups)
+ .map(ConsumerGroup::name)
+ .containsExactlyInAnyOrder("consumer-group-42", "consumer-group-43", "consumer-group-44");
+ }
+
+}
diff --git a/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/ConsumerOffsetsClientBaseTest.java b/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/ConsumerOffsetsClientBaseTest.java
new file mode 100644
index 00000000..c8fd505a
--- /dev/null
+++ b/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/ConsumerOffsetsClientBaseTest.java
@@ -0,0 +1,50 @@
+package rs.iggy.clients.blocking;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import rs.iggy.consumergroup.Consumer;
+import rs.iggy.identifier.ConsumerId;
+import rs.iggy.identifier.StreamId;
+import rs.iggy.identifier.TopicId;
+import java.math.BigInteger;
+import java.util.Optional;
+import static org.assertj.core.api.Assertions.assertThat;
+
+public abstract class ConsumerOffsetsClientBaseTest extends IntegrationTest {
+
+ ConsumerOffsetsClient consumerOffsetsClient;
+
+ @BeforeEach
+ void beforeEachBase() {
+ consumerOffsetsClient = client.consumerOffsets();
+
+ login();
+ setUpStreamAndTopic();
+ }
+
+ @Test
+ void shouldGetConsumerOffset() {
+ // when
+ var consumer = new Consumer(Consumer.Kind.Consumer, ConsumerId.of(1223L));
+ consumerOffsetsClient.storeConsumerOffset(StreamId.of(42L),
+ TopicId.of(42L),
+ Optional.empty(),
+ consumer,
+ BigInteger.ZERO);
+ var consumerOffset = consumerOffsetsClient.getConsumerOffset(StreamId.of(42L),
+ TopicId.of(42L),
+ Optional.of(1L),
+ consumer);
+
+ var nonExistingConsumerOffset = consumerOffsetsClient.getConsumerOffset(StreamId.of(42L),
+ TopicId.of(42L),
+ Optional.of(2L),
+ new Consumer(
+ Consumer.Kind.Consumer, ConsumerId.of(123L)));
+
+ // then
+ assertThat(consumerOffset).isPresent();
+ assertThat(nonExistingConsumerOffset).isEmpty();
+ }
+
+}
diff --git a/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/IntegrationTest.java b/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/IntegrationTest.java
new file mode 100644
index 00000000..bbad5e79
--- /dev/null
+++ b/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/IntegrationTest.java
@@ -0,0 +1,53 @@
+package rs.iggy.clients.blocking;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
+import org.testcontainers.utility.DockerImageName;
+import rs.iggy.topic.CompressionAlgorithm;
+import java.math.BigInteger;
+import static java.util.Optional.empty;
+import static java.util.Optional.of;
+
+@Testcontainers
+public abstract class IntegrationTest {
+
+ public static final int HTTP_PORT = 3000;
+ public static final int TCP_PORT = 8090;
+
+ @Container
+ protected final GenericContainer<?> iggyServer = new GenericContainer<>(DockerImageName.parse("iggyrs/iggy:latest"))
+ .withExposedPorts(HTTP_PORT, TCP_PORT);
+
+ protected IggyBaseClient client;
+
+ @BeforeEach
+ void beforeEachIntegrationTest() {
+ client = getClient();
+ }
+
+ abstract protected IggyBaseClient getClient();
+
+ protected void setUpStream() {
+ client.streams().createStream(of(42L), "test-stream");
+ }
+
+ protected void setUpStreamAndTopic() {
+ setUpStream();
+ client.topics()
+ .createTopic(42L,
+ of(42L),
+ 1L,
+ CompressionAlgorithm.None,
+ BigInteger.ZERO,
+ BigInteger.ZERO,
+ empty(),
+ "test-topic");
+ }
+
+ protected void login() {
+ client.users().login("iggy", "iggy");
+ }
+
+}
diff --git a/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/MessagesClientBaseTest.java b/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/MessagesClientBaseTest.java
new file mode 100644
index 00000000..61b732b0
--- /dev/null
+++ b/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/MessagesClientBaseTest.java
@@ -0,0 +1,74 @@
+package rs.iggy.clients.blocking;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import rs.iggy.message.*;
+import java.math.BigInteger;
+import java.util.List;
+import java.util.UUID;
+import static java.util.Optional.empty;
+import static org.assertj.core.api.Assertions.assertThat;
+
+public abstract class MessagesClientBaseTest extends IntegrationTest {
+
+ protected MessagesClient messagesClient;
+
+ @BeforeEach
+ void beforeEachBase() {
+ messagesClient = client.messages();
+
+ login();
+ }
+
+ @Test
+ void shouldSendAndGetMessages() {
+ // given
+ setUpStreamAndTopic();
+
+ // when
+ String text = "message from java sdk";
+ messagesClient.sendMessages(42L, 42L, Partitioning.partitionId(1L),
+ List.of(new Message(new UuidMessageId(UUID.randomUUID()), text.getBytes(), empty())));
+
+ var polledMessages = messagesClient.pollMessages(42L, 42L, empty(), 0L,
+ new PollingStrategy(PollingKind.Last, BigInteger.TEN), 10L, false);
+
+ // then
+ assertThat(polledMessages.messages()).hasSize(1);
+ }
+
+ @Test
+ void shouldSendMessageWithBalancedPartitioning() {
+ // given
+ setUpStreamAndTopic();
+
+ // when
+ String text = "message from java sdk";
+ messagesClient.sendMessages(42L, 42L, Partitioning.balanced(),
+ List.of(new Message(new UuidMessageId(UUID.randomUUID()), text.getBytes(), empty())));
+
+ var polledMessages = messagesClient.pollMessages(42L, 42L, empty(), 0L,
+ new PollingStrategy(PollingKind.Last, BigInteger.TEN), 10L, false);
+
+ // then
+ assertThat(polledMessages.messages()).hasSize(1);
+ }
+
+ @Test
+ void shouldSendMessageWithMessageKeyPartitioning() {
+ // given
+ setUpStreamAndTopic();
+
+ // when
+ String text = "message from java sdk";
+ messagesClient.sendMessages(42L, 42L, Partitioning.messagesKey("test-key"),
+ List.of(new Message(new UuidMessageId(UUID.randomUUID()), text.getBytes(), empty())));
+
+ var polledMessages = messagesClient.pollMessages(42L, 42L, empty(), 0L,
+ new PollingStrategy(PollingKind.Last, BigInteger.TEN), 10L, false);
+
+ // then
+ assertThat(polledMessages.messages()).hasSize(1);
+ }
+
+}
diff --git a/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/PartitionsClientBaseTest.java b/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/PartitionsClientBaseTest.java
new file mode 100644
index 00000000..f21c18da
--- /dev/null
+++ b/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/PartitionsClientBaseTest.java
@@ -0,0 +1,43 @@
+package rs.iggy.clients.blocking;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import rs.iggy.topic.TopicDetails;
+import static org.assertj.core.api.Assertions.assertThat;
+
+public abstract class PartitionsClientBaseTest extends IntegrationTest {
+
+ TopicsClient topicsClient;
+ PartitionsClient partitionsClient;
+
+ @BeforeEach
+ void beforeEachBase() {
+ topicsClient = client.topics();
+ partitionsClient = client.partitions();
+
+ login();
+ setUpStreamAndTopic();
+ }
+
+
+ @Test
+ void shouldCreateAndDeletePartitions() {
+ // given
+ assert topicsClient.getTopic(42L, 42L).get().partitionsCount() == 1L;
+
+ // when
+ partitionsClient.createPartitions(42L, 42L, 10L);
+
+ // then
+ TopicDetails topic = topicsClient.getTopic(42L, 42L).get();
+ assertThat(topic.partitionsCount()).isEqualTo(11L);
+
+ // when
+ partitionsClient.deletePartitions(42L, 42L, 10L);
+
+ // then
+ topic = topicsClient.getTopic(42L, 42L).get();
+ assertThat(topic.partitionsCount()).isEqualTo(1L);
+ }
+
+}
diff --git a/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/PersonalAccessTokensBaseTest.java b/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/PersonalAccessTokensBaseTest.java
new file mode 100644
index 00000000..3df303a6
--- /dev/null
+++ b/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/PersonalAccessTokensBaseTest.java
@@ -0,0 +1,64 @@
+package rs.iggy.clients.blocking;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import rs.iggy.user.IdentityInfo;
+import java.math.BigInteger;
+import static org.assertj.core.api.Assertions.assertThat;
+
+public abstract class PersonalAccessTokensBaseTest extends IntegrationTest {
+
+ protected PersonalAccessTokensClient personalAccessTokensClient;
+
+ @BeforeEach
+ void beforeEachBase() {
+ personalAccessTokensClient = client.personalAccessTokens();
+
+ login();
+ }
+
+ @Test
+ void shouldManagePersonalAccessTokens() {
+ // when
+ var createdToken = personalAccessTokensClient.createPersonalAccessToken("new-token",
+ BigInteger.valueOf(50_000));
+
+ // then
+ assertThat(createdToken).isNotNull();
+
+ // when
+ var tokens = personalAccessTokensClient.getPersonalAccessTokens();
+
+ // then
+ assertThat(tokens).isNotNull();
+ assertThat(tokens).hasSize(1);
+
+ // when
+ personalAccessTokensClient.deletePersonalAccessToken("new-token");
+ tokens = personalAccessTokensClient.getPersonalAccessTokens();
+
+ //
+ assertThat(tokens).hasSize(0);
+ }
+
+ @Test
+ void shouldCreateAndLogInWithPersonalAccessToken() {
+ // given
+ var createdToken = personalAccessTokensClient.createPersonalAccessToken("new-token",
+ BigInteger.valueOf(50_000));
+ client.users().logout();
+
+ // when
+ IdentityInfo identityInfo = personalAccessTokensClient.loginWithPersonalAccessToken(createdToken.token());
+
+ // then
+ assertThat(identityInfo).isNotNull();
+
+ // when
+ var user = client.users().getUser(1L);
+
+ // then
+ assertThat(user).isPresent();
+ }
+
+}
diff --git a/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/StreamClientBaseTest.java b/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/StreamClientBaseTest.java
new file mode 100644
index 00000000..963b01d9
--- /dev/null
+++ b/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/StreamClientBaseTest.java
@@ -0,0 +1,70 @@
+package rs.iggy.clients.blocking;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import java.util.Optional;
+import static org.assertj.core.api.Assertions.assertThat;
+
+public abstract class StreamClientBaseTest extends IntegrationTest {
+
+ StreamsClient streamsClient;
+
+ @BeforeEach
+ void beforeEachBase() {
+ streamsClient = client.streams();
+
+ login();
+ }
+
+ @Test
+ void shouldCreateAndDeleteStream() {
+ // when
+ streamsClient.createStream(Optional.of(42L), "test-stream");
+ var streamOptional = streamsClient.getStream(42L);
+
+ // then
+ assertThat(streamOptional).isPresent();
+ var stream = streamOptional.get();
+ assertThat(stream.id()).isEqualTo(42L);
+ assertThat(stream.name()).isEqualTo("test-stream");
+
+ // when
+ var streams = streamsClient.getStreams();
+
+ // then
+ assertThat(streams).hasSize(1);
+
+ // when
+ streamsClient.deleteStream(42L);
+ streams = streamsClient.getStreams();
+
+ // then
+ assertThat(streams).isEmpty();
+ }
+
+ @Test
+ void shouldUpdateStream() {
+ // given
+ streamsClient.createStream(Optional.of(42L), "test-stream");
+
+ // when
+ streamsClient.updateStream(42L, "test-stream-new");
+
+ // then
+ var streamOptional = streamsClient.getStream(42L);
+
+ assertThat(streamOptional).isPresent();
+ var stream = streamOptional.get();
+ assertThat(stream.id()).isEqualTo(42L);
+ assertThat(stream.name()).isEqualTo("test-stream-new");
+ }
+
+ @Test
+ void shouldReturnEmptyForNonExistingStream() {
+ // when
+ var stream = streamsClient.getStream(333L);
+
+ // then
+ assertThat(stream).isEmpty();
+ }
+}
diff --git a/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/SystemClientBaseTest.java b/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/SystemClientBaseTest.java
new file mode 100644
index 00000000..4b13742e
--- /dev/null
+++ b/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/SystemClientBaseTest.java
@@ -0,0 +1,27 @@
+package rs.iggy.clients.blocking;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import static org.assertj.core.api.Assertions.assertThat;
+
+public abstract class SystemClientBaseTest extends IntegrationTest {
+
+ protected SystemClient systemClient;
+
+ @BeforeEach
+ void beforeEachBase() {
+ systemClient = client.system();
+
+ login();
+ }
+
+ @Test
+ void shouldGetStats() {
+ // when
+ var stats = systemClient.getStats();
+
+ // then
+ assertThat(stats).isNotNull();
+ }
+
+}
diff --git a/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/TopicsClientBaseTest.java b/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/TopicsClientBaseTest.java
new file mode 100644
index 00000000..6bc4dea4
--- /dev/null
+++ b/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/TopicsClientBaseTest.java
@@ -0,0 +1,88 @@
+package rs.iggy.clients.blocking;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import rs.iggy.identifier.StreamId;
+import rs.iggy.identifier.TopicId;
+import rs.iggy.topic.CompressionAlgorithm;
+import java.math.BigInteger;
+import static java.util.Optional.empty;
+import static java.util.Optional.of;
+import static org.assertj.core.api.Assertions.assertThat;
+
+public abstract class TopicsClientBaseTest extends IntegrationTest {
+
+ protected TopicsClient topicsClient;
+
+ @BeforeEach
+ void beforeEachBase() {
+ topicsClient = client.topics();
+
+ login();
+ setUpStream();
+ }
+
+ @Test
+ void shouldCreateAndDeleteTopic() {
+ // when
+ topicsClient.createTopic(42L,
+ of(42L),
+ 1L,
+ CompressionAlgorithm.None,
+ BigInteger.ZERO,
+ BigInteger.ZERO,
+ empty(),
+ "test-topic");
+ var topicOptional = topicsClient.getTopic(42L, 42L);
+
+ // then
+ assertThat(topicOptional).isPresent();
+ var topic = topicOptional.get();
+ assertThat(topic.id()).isEqualTo(42L);
+ assertThat(topic.name()).isEqualTo("test-topic");
+
+ // when
+ topicsClient.deleteTopic(42L, 42L);
+
+ // then
+ assertThat(topicsClient.getTopics(42L)).isEmpty();
+ }
+
+ @Test
+ void shouldUpdateTopic() {
+ // given
+ var topic = topicsClient.createTopic(42L,
+ of(42L),
+ 1L,
+ CompressionAlgorithm.None,
+ BigInteger.ZERO,
+ BigInteger.ZERO,
+ empty(),
+ "test-topic");
+
+
+ // when
+ topicsClient.updateTopic(StreamId.of(42L),
+ TopicId.of(topic.id()),
+ CompressionAlgorithm.None,
+ BigInteger.valueOf(5000),
+ BigInteger.ZERO,
+ empty(),
+ "new-name");
+
+ // then
+ var updatedTopic = topicsClient.getTopic(42L, 42L).get();
+ assertThat(updatedTopic.name()).isEqualTo("new-name");
+ assertThat(updatedTopic.messageExpiry()).isEqualTo(BigInteger.valueOf(5000));
+ }
+
+ @Test
+ void shouldReturnEmptyForNonExistingTopic() {
+ // when
+ var topic = topicsClient.getTopic(42L, 404L);
+
+ // then
+ assertThat(topic).isEmpty();
+ }
+
+}
diff --git a/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/UsersClientBaseTest.java b/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/UsersClientBaseTest.java
new file mode 100644
index 00000000..b8be0ef5
--- /dev/null
+++ b/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/UsersClientBaseTest.java
@@ -0,0 +1,137 @@
+package rs.iggy.clients.blocking;
+
+import org.jetbrains.annotations.NotNull;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import rs.iggy.user.*;
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+import static org.assertj.core.api.Assertions.assertThat;
+
+public abstract class UsersClientBaseTest extends IntegrationTest {
+
+ protected UsersClient usersClient;
+
+ @BeforeEach
+ void beforeEachBase() {
+ usersClient = client.users();
+ }
+
+ @Test
+ void shouldLogin() {
+ // when
+ var identityInfo = usersClient.login("iggy", "iggy");
+
+ // then
+ assertThat(identityInfo).isNotNull();
+ assertThat(identityInfo.userId()).isEqualTo(1L);
+ }
+
+ @Test
+ void shouldGetUser() {
+ // given
+ login();
+
+ // when
+ var user = usersClient.getUser(1L);
+
+ // then
+ assertThat(user).isPresent();
+ }
+
+ @Test
+ void shouldCreateAndDeleteUser() {
+ // given
+ login();
+
+ // when
+ var createdUser = usersClient.createUser("test",
+ "test",
+ UserStatus.Active,
+ Optional.of(new Permissions(createGlobalPermissions(true), Collections.emptyMap())));
+
+ // then
+ assertThat(createdUser).isNotNull();
+ assertThat(createdUser.permissions().map(Permissions::global).map(GlobalPermissions::manageServers).orElse(false)).isTrue();
+
+ // when
+ List<UserInfo> users = usersClient.getUsers();
+
+ // then
+ assertThat(users).hasSize(2);
+ assertThat(users).map(UserInfo::username).containsExactlyInAnyOrder("iggy", "test");
+
+ // when
+ usersClient.deleteUser(createdUser.id());
+
+ // then
+ users = usersClient.getUsers();
+ assertThat(users).hasSize(1);
+ }
+
+ @Test
+ void shouldUpdateUserStatus() {
+ // given
+ login();
+ UserInfoDetails user = usersClient.createUser("test", "test", UserStatus.Active, Optional.empty());
+
+ // when
+ usersClient.updateUser(user.id(), Optional.empty(), Optional.of(UserStatus.Inactive));
+
+ // then
+ List<UserInfo> users = usersClient.getUsers();
+ assertThat(users).map(UserInfo::status).contains(UserStatus.Inactive);
+ }
+
+ @Test
+ void shouldUpdateUserPermissions() {
+ // given
+ var permissions = new Permissions(createGlobalPermissions(true), Collections.emptyMap());
+ login();
+ UserInfoDetails user = usersClient.createUser("test", "test", UserStatus.Active, Optional.of(permissions));
+
+ // when
+ usersClient.updatePermissions(user.id(), Optional.of(new Permissions(createGlobalPermissions(false), Collections.emptyMap())));
+
+ // then
+ var updatedUser = usersClient.getUser(user.id());
+ assertThat(updatedUser).isPresent();
+ assertThat(updatedUser.get()
+ .permissions()
+ .map(Permissions::global)
+ .map(GlobalPermissions::manageServers)
+ .orElse(true)).isFalse();
+ }
+
+ @Test
+ void shouldChangePassword() {
+ // given
+ IdentityInfo identity = usersClient.login("iggy", "iggy");
+
+ // when
+ usersClient.changePassword(identity.userId(), "iggy", "new-pass");
+ usersClient.logout();
+ IdentityInfo newLogin = usersClient.login("iggy", "new-pass");
+
+ // then
+ assertThat(newLogin).isNotNull();
+ }
+
+ @Test
+ void shouldReturnEmptyForNonExistingUser() {
+ // given
+ login();
+
+ // when
+ var user = usersClient.getUser(404L);
+
+ // then
+ assertThat(user).isEmpty();
+ }
+
+ private static @NotNull GlobalPermissions createGlobalPermissions(boolean manageServers) {
+ return new GlobalPermissions(manageServers, false, false, false, false, false, false, false, false, false);
+ }
+
+}
diff --git a/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/http/ConsumerGroupsHttpClientTest.java b/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/http/ConsumerGroupsHttpClientTest.java
new file mode 100644
index 00000000..c7b0ba54
--- /dev/null
+++ b/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/http/ConsumerGroupsHttpClientTest.java
@@ -0,0 +1,13 @@
+package rs.iggy.clients.blocking.http;
+
+import rs.iggy.clients.blocking.ConsumerGroupsClientBaseTest;
+import rs.iggy.clients.blocking.IggyBaseClient;
+
+class ConsumerGroupsHttpClientTest extends ConsumerGroupsClientBaseTest {
+
+ @Override
+ protected IggyBaseClient getClient() {
+ return HttpClientFactory.create(iggyServer);
+ }
+
+}
diff --git a/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/http/ConsumerOffsetsHttpClientTest.java b/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/http/ConsumerOffsetsHttpClientTest.java
new file mode 100644
index 00000000..8c6371a7
--- /dev/null
+++ b/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/http/ConsumerOffsetsHttpClientTest.java
@@ -0,0 +1,13 @@
+package rs.iggy.clients.blocking.http;
+
+import rs.iggy.clients.blocking.ConsumerOffsetsClientBaseTest;
+import rs.iggy.clients.blocking.IggyBaseClient;
+
+class ConsumerOffsetsHttpClientTest extends ConsumerOffsetsClientBaseTest {
+
+ @Override
+ protected IggyBaseClient getClient() {
+ return HttpClientFactory.create(iggyServer);
+ }
+
+}
diff --git a/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/http/HttpClientFactory.java b/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/http/HttpClientFactory.java
new file mode 100644
index 00000000..1b653fac
--- /dev/null
+++ b/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/http/HttpClientFactory.java
@@ -0,0 +1,14 @@
+package rs.iggy.clients.blocking.http;
+
+import org.testcontainers.containers.GenericContainer;
+import static rs.iggy.clients.blocking.IntegrationTest.HTTP_PORT;
+
+class HttpClientFactory {
+
+ static IggyHttpClient create(GenericContainer<?> iggyServer) {
+ String address = iggyServer.getHost();
+ Integer port = iggyServer.getMappedPort(HTTP_PORT);
+ return new IggyHttpClient("http://" + address + ":" + port);
+ }
+
+}
diff --git a/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/http/MessagesHttpClientTest.java b/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/http/MessagesHttpClientTest.java
new file mode 100644
index 00000000..99d7f978
--- /dev/null
+++ b/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/http/MessagesHttpClientTest.java
@@ -0,0 +1,13 @@
+package rs.iggy.clients.blocking.http;
+
+import rs.iggy.clients.blocking.IggyBaseClient;
+import rs.iggy.clients.blocking.MessagesClientBaseTest;
+
+class MessagesHttpClientTest extends MessagesClientBaseTest {
+
+ @Override
+ protected IggyBaseClient getClient() {
+ return HttpClientFactory.create(iggyServer);
+ }
+
+}
diff --git a/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/http/PartitionsHttpClientTest.java b/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/http/PartitionsHttpClientTest.java
new file mode 100644
index 00000000..8601f156
--- /dev/null
+++ b/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/http/PartitionsHttpClientTest.java
@@ -0,0 +1,13 @@
+package rs.iggy.clients.blocking.http;
+
+import rs.iggy.clients.blocking.IggyBaseClient;
+import rs.iggy.clients.blocking.PartitionsClientBaseTest;
+
+class PartitionsHttpClientTest extends PartitionsClientBaseTest {
+
+ @Override
+ protected IggyBaseClient getClient() {
+ return HttpClientFactory.create(iggyServer);
+ }
+
+}
diff --git a/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/http/PersonalAccessTokensHttpClientTest.java b/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/http/PersonalAccessTokensHttpClientTest.java
new file mode 100644
index 00000000..3159fee5
--- /dev/null
+++ b/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/http/PersonalAccessTokensHttpClientTest.java
@@ -0,0 +1,13 @@
+package rs.iggy.clients.blocking.http;
+
+import rs.iggy.clients.blocking.IggyBaseClient;
+import rs.iggy.clients.blocking.PersonalAccessTokensBaseTest;
+
+class PersonalAccessTokensHttpClientTest extends PersonalAccessTokensBaseTest {
+
+ @Override
+ protected IggyBaseClient getClient() {
+ return HttpClientFactory.create(iggyServer);
+ }
+
+}
diff --git a/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/http/StreamHttpClientTest.java b/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/http/StreamHttpClientTest.java
new file mode 100644
index 00000000..4f9560a4
--- /dev/null
+++ b/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/http/StreamHttpClientTest.java
@@ -0,0 +1,13 @@
+package rs.iggy.clients.blocking.http;
+
+import rs.iggy.clients.blocking.IggyBaseClient;
+import rs.iggy.clients.blocking.StreamClientBaseTest;
+
+class StreamHttpClientTest extends StreamClientBaseTest {
+
+ @Override
+ protected IggyBaseClient getClient() {
+ return HttpClientFactory.create(iggyServer);
+ }
+
+}
diff --git a/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/http/SystemHttpClientTest.java b/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/http/SystemHttpClientTest.java
new file mode 100644
index 00000000..b0b29f3e
--- /dev/null
+++ b/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/http/SystemHttpClientTest.java
@@ -0,0 +1,13 @@
+package rs.iggy.clients.blocking.http;
+
+import rs.iggy.clients.blocking.IggyBaseClient;
+import rs.iggy.clients.blocking.SystemClientBaseTest;
+
+class SystemHttpClientTest extends SystemClientBaseTest {
+
+ @Override
+ protected IggyBaseClient getClient() {
+ return HttpClientFactory.create(iggyServer);
+ }
+
+}
diff --git a/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/http/TopicsHttpClientTest.java b/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/http/TopicsHttpClientTest.java
new file mode 100644
index 00000000..9261122d
--- /dev/null
+++ b/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/http/TopicsHttpClientTest.java
@@ -0,0 +1,13 @@
+package rs.iggy.clients.blocking.http;
+
+import rs.iggy.clients.blocking.IggyBaseClient;
+import rs.iggy.clients.blocking.TopicsClientBaseTest;
+
+class TopicsHttpClientTest extends TopicsClientBaseTest {
+
+ @Override
+ protected IggyBaseClient getClient() {
+ return HttpClientFactory.create(iggyServer);
+ }
+
+}
diff --git a/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/http/UsersHttpClientTest.java b/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/http/UsersHttpClientTest.java
new file mode 100644
index 00000000..f51abe4b
--- /dev/null
+++ b/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/http/UsersHttpClientTest.java
@@ -0,0 +1,27 @@
+package rs.iggy.clients.blocking.http;
+
+import org.junit.jupiter.api.Test;
+import rs.iggy.clients.blocking.IggyBaseClient;
+import rs.iggy.clients.blocking.UsersClientBaseTest;
+import static org.assertj.core.api.Assertions.assertThat;
+
+class UsersHttpClientTest extends UsersClientBaseTest {
+
+ @Override
+ protected IggyBaseClient getClient() {
+ return HttpClientFactory.create(iggyServer);
+ }
+
+
+ @Test
+ void shouldGetTokens() {
+ // when
+ var identityInfo = usersClient.login("iggy", "iggy");
+
+ // then
+ assertThat(identityInfo).isNotNull();
+ assertThat(identityInfo.accessToken()).isPresent();
+ assertThat(identityInfo.accessToken().get().token()).isNotBlank();
+ }
+
+}
diff --git a/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/tcp/BytesSerializerTest.java b/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/tcp/BytesSerializerTest.java
new file mode 100644
index 00000000..f6d9a31f
--- /dev/null
+++ b/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/tcp/BytesSerializerTest.java
@@ -0,0 +1,101 @@
+package rs.iggy.clients.blocking.tcp;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import org.junit.jupiter.api.Nested;
+import org.junit.jupiter.api.Test;
+import java.math.BigInteger;
+import java.util.Arrays;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+class BytesSerializerTest {
+
+ @Nested
+ class U64 {
+
+ @Test
+ void shouldSerializeMaxValue() {
+ // given
+ long maxLong = 0xFFFF_FFFF_FFFF_FFFFL;
+ var maxU64 = Long.toUnsignedString(maxLong);
+ var value = new BigInteger(maxU64);
+
+ // when
+ ByteBuf bytesAsU64 = BytesSerializer.toBytesAsU64(value);
+
+ // then
+ assertThat(bytesAsU64).isEqualByComparingTo(Unpooled.copyLong(maxLong));
+ }
+
+ @Test
+ void shouldSerializeZero() {
+ // given
+ var value = BigInteger.ZERO;
+ var zeroAsU64 = Unpooled.buffer(8);
+ zeroAsU64.writeZero(8);
+
+ // when
+ ByteBuf bytesAsU64 = BytesSerializer.toBytesAsU64(value);
+
+ // then
+ assertThat(bytesAsU64).isEqualByComparingTo(zeroAsU64);
+ }
+
+ @Test
+ void shouldFailForValueBelowZero() {
+ // given
+ var value = BigInteger.valueOf(-1);
+
+ // when & then
+ assertThatThrownBy(() -> BytesSerializer.toBytesAsU64(value))
+ .isInstanceOf(IllegalArgumentException.class);
+ }
+
+ }
+
+ @Nested
+ class U128 {
+
+ @Test
+ void shouldSerializeMaxValue() {
+ // given
+ byte[] maxU128 = new byte[17];
+ Arrays.fill(maxU128, 1, 17, (byte) 0xFF);
+
+ var value = new BigInteger(maxU128);
+
+ // when
+ ByteBuf bytesAsU128 = BytesSerializer.toBytesAsU128(value);
+
+ // then
+ assertThat(bytesAsU128).isEqualByComparingTo(Unpooled.wrappedBuffer(maxU128, 1, 16));
+ }
+
+
+ @Test
+ void shouldSerializeZero() {
+ // given
+ var value = BigInteger.ZERO;
+ var zeroAsU128 = Unpooled.buffer(8);
+ zeroAsU128.writeZero(16);
+
+ // when
+ ByteBuf bytesAsU128 = BytesSerializer.toBytesAsU128(value);
+
+ // then
+ assertThat(bytesAsU128).isEqualByComparingTo(zeroAsU128);
+ }
+
+ @Test
+ void shouldFailForValueBelowZero() {
+ // given
+ var value = BigInteger.valueOf(-1);
+
+ // when & then
+ assertThatThrownBy(() -> BytesSerializer.toBytesAsU128(value))
+ .isInstanceOf(IllegalArgumentException.class);
+ }
+ }
+
+}
diff --git a/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/tcp/ConsumerGroupsTcpClientTest.java b/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/tcp/ConsumerGroupsTcpClientTest.java
new file mode 100644
index 00000000..9439feb2
--- /dev/null
+++ b/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/tcp/ConsumerGroupsTcpClientTest.java
@@ -0,0 +1,40 @@
+package rs.iggy.clients.blocking.tcp;
+
+import org.junit.jupiter.api.Test;
+import rs.iggy.clients.blocking.ConsumerGroupsClientBaseTest;
+import rs.iggy.clients.blocking.IggyBaseClient;
+import java.util.Optional;
+import static org.assertj.core.api.Assertions.assertThat;
+
+class ConsumerGroupsTcpClientTest extends ConsumerGroupsClientBaseTest {
+
+ @Override
+ protected IggyBaseClient getClient() {
+ return TcpClientFactory.create(iggyServer);
+ }
+
+ @Test
+ void shouldJoinAndLeaveConsumerGroup() {
+ // given
+ setUpStreamAndTopic();
+ var group = consumerGroupsClient.createConsumerGroup(42L,
+ 42L,
+ Optional.of(42L),
+ "consumer-group-42");
+
+ // when
+ consumerGroupsClient.joinConsumerGroup(42L, 42L, group.id());
+
+ // then
+ group = consumerGroupsClient.getConsumerGroup(42L, 42L, group.id()).get();
+ assertThat(group.membersCount()).isEqualTo(1);
+
+ // when
+ consumerGroupsClient.leaveConsumerGroup(42L, 42L, group.id());
+
+ // then
+ group = consumerGroupsClient.getConsumerGroup(42L, 42L, group.id()).get();
+ assertThat(group.membersCount()).isEqualTo(0);
+ }
+
+}
diff --git a/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/tcp/ConsumerOffsetsTcpClientTest.java b/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/tcp/ConsumerOffsetsTcpClientTest.java
new file mode 100644
index 00000000..1996b54f
--- /dev/null
+++ b/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/tcp/ConsumerOffsetsTcpClientTest.java
@@ -0,0 +1,13 @@
+package rs.iggy.clients.blocking.tcp;
+
+import rs.iggy.clients.blocking.ConsumerOffsetsClientBaseTest;
+import rs.iggy.clients.blocking.IggyBaseClient;
+
+class ConsumerOffsetsTcpClientTest extends ConsumerOffsetsClientBaseTest {
+
+ @Override
+ protected IggyBaseClient getClient() {
+ return TcpClientFactory.create(iggyServer);
+ }
+
+}
diff --git a/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/tcp/MessagesTcpClientTest.java b/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/tcp/MessagesTcpClientTest.java
new file mode 100644
index 00000000..8b9377d9
--- /dev/null
+++ b/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/tcp/MessagesTcpClientTest.java
@@ -0,0 +1,13 @@
+package rs.iggy.clients.blocking.tcp;
+
+import rs.iggy.clients.blocking.IggyBaseClient;
+import rs.iggy.clients.blocking.MessagesClientBaseTest;
+
+class MessagesTcpClientTest extends MessagesClientBaseTest {
+
+ @Override
+ protected IggyBaseClient getClient() {
+ return TcpClientFactory.create(iggyServer);
+ }
+
+}
diff --git a/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/tcp/PartitionsTcpClientTest.java b/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/tcp/PartitionsTcpClientTest.java
new file mode 100644
index 00000000..b6c5d694
--- /dev/null
+++ b/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/tcp/PartitionsTcpClientTest.java
@@ -0,0 +1,13 @@
+package rs.iggy.clients.blocking.tcp;
+
+import rs.iggy.clients.blocking.IggyBaseClient;
+import rs.iggy.clients.blocking.PartitionsClientBaseTest;
+
+class PartitionsTcpClientTest extends PartitionsClientBaseTest {
+
+ @Override
+ protected IggyBaseClient getClient() {
+ return TcpClientFactory.create(iggyServer);
+ }
+
+}
diff --git a/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/tcp/PersonalAccessTokensTcpClientTest.java b/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/tcp/PersonalAccessTokensTcpClientTest.java
new file mode 100644
index 00000000..36fbce40
--- /dev/null
+++ b/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/tcp/PersonalAccessTokensTcpClientTest.java
@@ -0,0 +1,13 @@
+package rs.iggy.clients.blocking.tcp;
+
+import rs.iggy.clients.blocking.IggyBaseClient;
+import rs.iggy.clients.blocking.PersonalAccessTokensBaseTest;
+
+class PersonalAccessTokensTcpClientTest extends PersonalAccessTokensBaseTest {
+
+ @Override
+ protected IggyBaseClient getClient() {
+ return TcpClientFactory.create(iggyServer);
+ }
+
+}
diff --git a/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/tcp/StreamTcpClientTest.java b/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/tcp/StreamTcpClientTest.java
new file mode 100644
index 00000000..45c0117f
--- /dev/null
+++ b/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/tcp/StreamTcpClientTest.java
@@ -0,0 +1,13 @@
+package rs.iggy.clients.blocking.tcp;
+
+import rs.iggy.clients.blocking.IggyBaseClient;
+import rs.iggy.clients.blocking.StreamClientBaseTest;
+
+class StreamTcpClientTest extends StreamClientBaseTest {
+
+ @Override
+ protected IggyBaseClient getClient() {
+ return TcpClientFactory.create(iggyServer);
+ }
+
+}
diff --git a/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/tcp/SystemTcpClientTest.java b/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/tcp/SystemTcpClientTest.java
new file mode 100644
index 00000000..49ecb2a8
--- /dev/null
+++ b/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/tcp/SystemTcpClientTest.java
@@ -0,0 +1,42 @@
+package rs.iggy.clients.blocking.tcp;
+
+import org.junit.jupiter.api.Test;
+import rs.iggy.clients.blocking.IggyBaseClient;
+import rs.iggy.clients.blocking.SystemClientBaseTest;
+import rs.iggy.system.ClientInfo;
+import rs.iggy.system.ClientInfoDetails;
+import java.util.List;
+import static org.assertj.core.api.Assertions.assertThat;
+
+class SystemTcpClientTest extends SystemClientBaseTest {
+
+ @Override
+ protected IggyBaseClient getClient() {
+ return TcpClientFactory.create(iggyServer);
+ }
+
+ @Test
+ void shouldGetMeAndClient() {
+ // when
+ ClientInfoDetails me = systemClient.getMe();
+
+ // then
+ assertThat(me).isNotNull();
+
+ // when
+ var clientInfo = systemClient.getClient(me.clientId());
+
+ // then
+ assertThat(clientInfo).isNotNull();
+ }
+
+ @Test
+ void shouldGetClients() {
+ // when
+ List<ClientInfo> clients = systemClient.getClients();
+
+ // then
+ assertThat(clients).isNotNull();
+ assertThat(clients.size()).isEqualTo(1);
+ }
+}
diff --git a/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/tcp/TcpClientFactory.java b/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/tcp/TcpClientFactory.java
new file mode 100644
index 00000000..54fec51b
--- /dev/null
+++ b/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/tcp/TcpClientFactory.java
@@ -0,0 +1,14 @@
+package rs.iggy.clients.blocking.tcp;
+
+import org.testcontainers.containers.GenericContainer;
+import static rs.iggy.clients.blocking.IntegrationTest.TCP_PORT;
+
+class TcpClientFactory {
+
+ static IggyTcpClient create(GenericContainer<?> iggyServer) {
+ String address = iggyServer.getHost();
+ Integer port = iggyServer.getMappedPort(TCP_PORT);
+ return new IggyTcpClient(address, port);
+ }
+
+}
diff --git a/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/tcp/TopicsTcpClientTest.java b/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/tcp/TopicsTcpClientTest.java
new file mode 100644
index 00000000..b10d2463
--- /dev/null
+++ b/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/tcp/TopicsTcpClientTest.java
@@ -0,0 +1,13 @@
+package rs.iggy.clients.blocking.tcp;
+
+import rs.iggy.clients.blocking.IggyBaseClient;
+import rs.iggy.clients.blocking.TopicsClientBaseTest;
+
+class TopicsTcpClientTest extends TopicsClientBaseTest {
+
+ @Override
+ protected IggyBaseClient getClient() {
+ return TcpClientFactory.create(iggyServer);
+ }
+
+}
diff --git a/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/tcp/UsersTcpClientTest.java b/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/tcp/UsersTcpClientTest.java
new file mode 100644
index 00000000..75ff9a29
--- /dev/null
+++ b/foreign/java/java-sdk/src/test/java/rs/iggy/clients/blocking/tcp/UsersTcpClientTest.java
@@ -0,0 +1,22 @@
+package rs.iggy.clients.blocking.tcp;
+
+import org.junit.jupiter.api.BeforeEach;
+import rs.iggy.clients.blocking.IggyBaseClient;
+import rs.iggy.clients.blocking.UsersClient;
+import rs.iggy.clients.blocking.UsersClientBaseTest;
+
+class UsersTcpClientTest extends UsersClientBaseTest {
+
+ protected UsersClient usersClient;
+
+ @Override
+ protected IggyBaseClient getClient() {
+ return TcpClientFactory.create(iggyServer);
+ }
+
+ @BeforeEach
+ void beforeEach() {
+ usersClient = client.users();
+ }
+
+}
diff --git a/foreign/java/java-sdk/src/test/resources/logback-test.xml b/foreign/java/java-sdk/src/test/resources/logback-test.xml
new file mode 100644
index 00000000..6e38fcd1
--- /dev/null
+++ b/foreign/java/java-sdk/src/test/resources/logback-test.xml
@@ -0,0 +1,19 @@
+<?xml version="1.0" encoding="UTF-8" ?>
+<!DOCTYPE configuration>
+
+<configuration>
+ <import class="ch.qos.logback.classic.encoder.PatternLayoutEncoder"/>
+ <import class="ch.qos.logback.core.ConsoleAppender"/>
+
+ <appender name="STDOUT" class="ConsoleAppender">
+ <encoder class="PatternLayoutEncoder">
+ <pattern>%d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} -%kvp- %msg%n</pattern>
+ </encoder>
+ </appender>
+
+ <logger name="rs.iggy" level="DEBUG"/>
+
+ <root level="INFO">
+ <appender-ref ref="STDOUT"/>
+ </root>
+</configuration>
diff --git a/foreign/java/settings.gradle.kts b/foreign/java/settings.gradle.kts
new file mode 100644
index 00000000..b37f9e78
--- /dev/null
+++ b/foreign/java/settings.gradle.kts
@@ -0,0 +1,8 @@
+rootProject.name = "iggy-java-client"
+
+include("java-sdk")
+
+include("simple-consumer-example")
+project(":simple-consumer-example").projectDir = file("examples/simple-consumer")
+include("simple-producer-example")
+project(":simple-producer-example").projectDir = file("examples/simple-producer")