Implement joining and leaving consumer group

iggy-rs · Oct 11, 2024 · 836f5ed · 836f5ed
1 parent 5f8949b
commit 836f5ed
Show file tree

Hide file tree

Showing 3 changed files with 37 additions and 3 deletions.
diff --git a/README.md b/README.md
@@ -14,7 +14,6 @@ Official Java client SDK for [Iggy.rs](https://iggy.rs) message streaming.
 ## To do
 
 - Transport protocols
-    - TCP
     - QUIC
 - Publish maven package
 - Add examples

diff --git a/src/main/java/rs/iggy/clients/blocking/tcp/ConsumerGroupsTcpClient.java b/src/main/java/rs/iggy/clients/blocking/tcp/ConsumerGroupsTcpClient.java
@@ -77,12 +77,20 @@ public void deleteConsumerGroup(StreamId streamId, TopicId topicId, ConsumerId g
 
     @Override
     public void joinConsumerGroup(StreamId streamId, TopicId topicId, ConsumerId groupId) {
-        throw new UnsupportedOperationException();
+        var payload = toBytes(streamId);
+        payload.writeBytes(toBytes(topicId));
+        payload.writeBytes(toBytes(groupId));
+
+        connection.send(JOIN_CONSUMER_GROUP_CODE, payload);
     }
 
     @Override
     public void leaveConsumerGroup(StreamId streamId, TopicId topicId, ConsumerId groupId) {
-        throw new UnsupportedOperationException();
+        var payload = toBytes(streamId);
+        payload.writeBytes(toBytes(topicId));
+        payload.writeBytes(toBytes(groupId));
+
+        connection.send(LEAVE_CONSUMER_GROUP_CODE, payload);
     }
 
 }
diff --git a/src/test/java/rs/iggy/clients/blocking/tcp/ConsumerGroupsTcpClientTest.java b/src/test/java/rs/iggy/clients/blocking/tcp/ConsumerGroupsTcpClientTest.java
@@ -1,7 +1,10 @@
 package rs.iggy.clients.blocking.tcp;
 
+import org.junit.jupiter.api.Test;
 import rs.iggy.clients.blocking.ConsumerGroupsClientBaseTest;
 import rs.iggy.clients.blocking.IggyClient;
+import java.util.Optional;
+import static org.assertj.core.api.Assertions.assertThat;
 
 class ConsumerGroupsTcpClientTest extends ConsumerGroupsClientBaseTest {
 
@@ -10,4 +13,28 @@ protected IggyClient 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());
+        assertThat(group.membersCount()).isEqualTo(1);
+
+        // when
+        consumerGroupsClient.leaveConsumerGroup(42L, 42L, group.id());
+
+        // then
+        group = consumerGroupsClient.getConsumerGroup(42L, 42L, group.id());
+        assertThat(group.membersCount()).isEqualTo(0);
+    }
+
 }