Rename message classes to align with server implementation

iggy-rs · Oct 23, 2024 · 2c9b5d9 · 2c9b5d9
1 parent 583abcb
commit 2c9b5d9
Show file tree

Hide file tree

Showing 12 changed files with 40 additions and 40 deletions.
diff --git a/examples/simple-consumer/src/main/java/rs/iggy/SimpleConsumer.java b/examples/simple-consumer/src/main/java/rs/iggy/SimpleConsumer.java
@@ -8,7 +8,7 @@
 import rs.iggy.identifier.ConsumerId;
 import rs.iggy.identifier.StreamId;
 import rs.iggy.identifier.TopicId;
-import rs.iggy.message.Message;
+import rs.iggy.message.PolledMessage;
 import rs.iggy.message.PolledMessages;
 import rs.iggy.message.PollingKind;
 import rs.iggy.message.PollingStrategy;
@@ -40,7 +40,7 @@ public static void main(String[] args) {
         createConsumerGroup(client);
         client.consumerGroups().joinConsumerGroup(STREAM_ID, TOPIC_ID, GROUP_ID);
 
-        List<Message> messages = new ArrayList<>();
+        List<PolledMessage> messages = new ArrayList<>();
         while (messages.size() < 1000) {
             PolledMessages polledMessages = client.messages()
                     .pollMessages(STREAM_ID,

diff --git a/examples/simple-producer/src/main/java/rs/iggy/SimpleProducer.java b/examples/simple-producer/src/main/java/rs/iggy/SimpleProducer.java
@@ -6,7 +6,7 @@
 import rs.iggy.identifier.StreamId;
 import rs.iggy.identifier.TopicId;
 import rs.iggy.message.BigIntegerMessageId;
-import rs.iggy.message.MessageToSend;
+import rs.iggy.message.Message;
 import rs.iggy.message.Partitioning;
 import rs.iggy.message.PartitioningKind;
 import rs.iggy.stream.StreamDetails;
@@ -34,7 +34,7 @@ public static void main(String[] args) {
         int counter = 0;
         while (counter++ < 1000) {
             var text = "message from simple producer " + counter;
-            var message = new MessageToSend(new BigIntegerMessageId(BigInteger.ZERO), text.getBytes(), empty());
+            var message = new Message(new BigIntegerMessageId(BigInteger.ZERO), text.getBytes(), empty());
             client.messages()
                     .sendMessages(STREAM_ID,
                             TOPIC_ID,

diff --git a/java-sdk/src/main/java/rs/iggy/clients/blocking/MessagesClient.java b/java-sdk/src/main/java/rs/iggy/clients/blocking/MessagesClient.java
@@ -3,7 +3,7 @@
 import rs.iggy.consumergroup.Consumer;
 import rs.iggy.identifier.StreamId;
 import rs.iggy.identifier.TopicId;
-import rs.iggy.message.MessageToSend;
+import rs.iggy.message.Message;
 import rs.iggy.message.Partitioning;
 import rs.iggy.message.PolledMessages;
 import rs.iggy.message.PollingStrategy;
@@ -19,10 +19,10 @@ default PolledMessages pollMessages(Long streamId, Long topicId, Optional<Long>
 
     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<MessageToSend> messages) {
+    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<MessageToSend> messages);
+    void sendMessages(StreamId streamId, TopicId topicId, Partitioning partitioning, List<Message> messages);
 
 }
diff --git a/java-sdk/src/main/java/rs/iggy/clients/blocking/http/MessagesHttpClient.java b/java-sdk/src/main/java/rs/iggy/clients/blocking/http/MessagesHttpClient.java
@@ -5,7 +5,7 @@
 import rs.iggy.consumergroup.Consumer;
 import rs.iggy.identifier.StreamId;
 import rs.iggy.identifier.TopicId;
-import rs.iggy.message.MessageToSend;
+import rs.iggy.message.Message;
 import rs.iggy.message.Partitioning;
 import rs.iggy.message.PolledMessages;
 import rs.iggy.message.PollingStrategy;
@@ -33,7 +33,7 @@ public PolledMessages pollMessages(StreamId streamId, TopicId topicId, Optional<
     }
 
     @Override
-    public void sendMessages(StreamId streamId, TopicId topicId, Partitioning partitioning, List<MessageToSend> messages) {
+    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);
     }
@@ -42,6 +42,6 @@ private static String path(StreamId streamId, TopicId topicId) {
         return "/streams/" + streamId + "/topics/" + topicId + "/messages";
     }
 
-    private record SendMessages(Partitioning partitioning, List<MessageToSend> messages) {
+    private record SendMessages(Partitioning partitioning, List<Message> messages) {
     }
 }
diff --git a/java-sdk/src/main/java/rs/iggy/clients/blocking/tcp/BytesDeserializer.java b/java-sdk/src/main/java/rs/iggy/clients/blocking/tcp/BytesDeserializer.java
@@ -138,14 +138,14 @@ public static PolledMessages readPolledMessages(ByteBuf response) {
         var partitionId = response.readUnsignedIntLE();
         var currentOffset = readU64AsBigInteger(response);
         var _messagesCount = response.readUnsignedIntLE();
-        var messages = new ArrayList<Message>();
+        var messages = new ArrayList<PolledMessage>();
         while (response.isReadable()) {
-            messages.add(readMessage(response));
+            messages.add(readPolledMessage(response));
         }
         return new PolledMessages(partitionId, currentOffset, messages);
     }
 
-    static Message readMessage(ByteBuf response) {
+    static PolledMessage readPolledMessage(ByteBuf response) {
         var offset = readU64AsBigInteger(response);
         var stateCode = response.readByte();
         var state = MessageState.fromCode(stateCode);
@@ -171,7 +171,7 @@ static Message readMessage(ByteBuf response) {
         var payloadLength = response.readUnsignedIntLE();
         var payload = newByteArray(payloadLength);
         response.readBytes(payload);
-        return new Message(offset, state, timestamp, id, checksum, headers, payload);
+        return new PolledMessage(offset, state, timestamp, id, checksum, headers, payload);
     }
 
     static Stats readStats(ByteBuf response) {

diff --git a/java-sdk/src/main/java/rs/iggy/clients/blocking/tcp/BytesSerializer.java b/java-sdk/src/main/java/rs/iggy/clients/blocking/tcp/BytesSerializer.java
@@ -6,7 +6,7 @@
 import rs.iggy.consumergroup.Consumer;
 import rs.iggy.identifier.Identifier;
 import rs.iggy.message.HeaderValue;
-import rs.iggy.message.MessageToSend;
+import rs.iggy.message.Message;
 import rs.iggy.message.Partitioning;
 import rs.iggy.message.PollingStrategy;
 import rs.iggy.user.GlobalPermissions;
@@ -61,7 +61,7 @@ static ByteBuf toBytes(Partitioning partitioning) {
         return buffer;
     }
 
-    static ByteBuf toBytes(MessageToSend message) {
+    static ByteBuf toBytes(Message message) {
         var buffer = Unpooled.buffer();
         buffer.writeBytes(message.id().toBytes());
         message.headers().ifPresentOrElse((headers) -> {

diff --git a/java-sdk/src/main/java/rs/iggy/clients/blocking/tcp/MessagesTcpClient.java b/java-sdk/src/main/java/rs/iggy/clients/blocking/tcp/MessagesTcpClient.java
@@ -4,7 +4,7 @@
 import rs.iggy.consumergroup.Consumer;
 import rs.iggy.identifier.StreamId;
 import rs.iggy.identifier.TopicId;
-import rs.iggy.message.MessageToSend;
+import rs.iggy.message.Message;
 import rs.iggy.message.Partitioning;
 import rs.iggy.message.PolledMessages;
 import rs.iggy.message.PollingStrategy;
@@ -39,7 +39,7 @@ public PolledMessages pollMessages(StreamId streamId, TopicId topicId, Optional<
     }
 
     @Override
-    public void sendMessages(StreamId streamId, TopicId topicId, Partitioning partitioning, List<MessageToSend> messages) {
+    public void sendMessages(StreamId streamId, TopicId topicId, Partitioning partitioning, List<Message> messages) {
         var payload = toBytes(streamId);
         payload.writeBytes(toBytes(topicId));
         payload.writeBytes(toBytes(partitioning));

diff --git a/java-sdk/src/main/java/rs/iggy/message/Message.java b/java-sdk/src/main/java/rs/iggy/message/Message.java
@@ -1,16 +1,11 @@
 package rs.iggy.message;
 
-import java.math.BigInteger;
 import java.util.Map;
 import java.util.Optional;
 
 public record Message(
-        BigInteger offset,
-        MessageState state,
-        BigInteger timestamp,
         MessageId id,
-        Long checksum,
-        Optional<Map<String, HeaderValue>> headers,
-        byte[] payload
+        byte[] payload,
+        Optional<Map<String, HeaderValue>> headers
 ) {
 }
diff --git a/java-sdk/src/main/java/rs/iggy/message/MessageToSend.java b/java-sdk/src/main/java/rs/iggy/message/MessageToSend.java
diff --git a/java-sdk/src/main/java/rs/iggy/message/PolledMessage.java b/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/java-sdk/src/main/java/rs/iggy/message/PolledMessages.java b/java-sdk/src/main/java/rs/iggy/message/PolledMessages.java
@@ -6,6 +6,6 @@
 public record PolledMessages(
         Long partitionId,
         BigInteger currentOffset,
-        List<Message> messages
+        List<PolledMessage> messages
 ) {
 }
diff --git a/java-sdk/src/test/java/rs/iggy/clients/blocking/MessagesClientBaseTest.java b/java-sdk/src/test/java/rs/iggy/clients/blocking/MessagesClientBaseTest.java
@@ -28,7 +28,7 @@ void shouldSendAndGetMessages() {
         // when
         String text = "message from java sdk";
         messagesClient.sendMessages(42L, 42L, Partitioning.partitionId(1L),
-                List.of(new MessageToSend(new UuidMessageId(UUID.randomUUID()), text.getBytes(), empty())));
+                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);
@@ -45,7 +45,7 @@ void shouldSendMessageWithBalancedPartitioning() {
         // when
         String text = "message from java sdk";
         messagesClient.sendMessages(42L, 42L, Partitioning.balanced(),
-                List.of(new MessageToSend(new UuidMessageId(UUID.randomUUID()), text.getBytes(), empty())));
+                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);
@@ -62,7 +62,7 @@ void shouldSendMessageWithMessageKeyPartitioning() {
         // when
         String text = "message from java sdk";
         messagesClient.sendMessages(42L, 42L, Partitioning.messagesKey("test-key"),
-                List.of(new MessageToSend(new UuidMessageId(UUID.randomUUID()), text.getBytes(), empty())));
+                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);