added method to subscribe topic for limited message

Signed-off-by: Manish Dait <[email protected]>

Author: manishdait

hiero-enterprise-base/src/main/java/com/openelements/hiero/base/TopicClient.java

@@ -328,7 +328,8 @@ default void submitMessage(@NonNull String topicId, @NonNull String submitKey, @
      * @return SubscriptionHandle for the Topic
      * @throws HieroException if Topic could not be subscribed
      */
-    SubscriptionHandle subscribeTopic(@NonNull TopicId topicId, @NonNull Consumer<TopicMessage> subscription) throws HieroException;
+    SubscriptionHandle subscribeTopic(@NonNull TopicId topicId, @NonNull Consumer<TopicMessage> subscription)
+            throws HieroException;
 
     /**
      * Subscribe to a Topic
@@ -337,7 +338,32 @@ default void submitMessage(@NonNull String topicId, @NonNull String submitKey, @
      * @return SubscriptionHandle for the Topic
      * @throws HieroException if Topic could not be subscribed
      */
-    default SubscriptionHandle subscribeTopic(@NonNull String topicId, @NonNull Consumer<TopicMessage> subscription) throws HieroException {
+    default SubscriptionHandle subscribeTopic(@NonNull String topicId, @NonNull Consumer<TopicMessage> subscription)
+            throws HieroException {
         return subscribeTopic(TopicId.fromString(topicId), subscription);
     }
+
+    /**
+     * Subscribe to a Topic
+     *
+     * @param topicId the topicId of topic
+     * @param limit the number of message to return
+     * @return SubscriptionHandle for the Topic
+     * @throws HieroException if Topic could not be subscribed
+     */
+    SubscriptionHandle subscribeTopic(@NonNull TopicId topicId, @NonNull Consumer<TopicMessage> subscription,
+                                      long limit) throws HieroException;
+
+    /**
+     * Subscribe to a Topic
+     *
+     * @param topicId the topicId of topic
+     * @param limit the number of message to return
+     * @return SubscriptionHandle for the Topic
+     * @throws HieroException if Topic could not be subscribed
+     */
+    default SubscriptionHandle subscribeTopic(@NonNull String topicId, @NonNull Consumer<TopicMessage> subscription,
+                                              long limit) throws HieroException {
+        return subscribeTopic(TopicId.fromString(topicId), subscription, limit);
+    }
 }

hiero-enterprise-base/src/main/java/com/openelements/hiero/base/implementation/TopicClientImpl.java

@@ -221,4 +221,18 @@ public SubscriptionHandle subscribeTopic(@NonNull TopicId topicId, @NonNull Cons
         TopicMessageResult result = client.executeTopicMessageQuery(request);
         return result.subscriptionHandle();
     }
+
+    @Override
+    public SubscriptionHandle subscribeTopic(@NonNull TopicId topicId, @NonNull Consumer<TopicMessage> subscription,
+                                             long limit) throws HieroException {
+        Objects.requireNonNull(topicId, "topicId must not be null");
+        Objects.requireNonNull(subscription, "subscription must not be null");
+        if (limit == 0) {
+            throw new IllegalArgumentException("limit must be greater than 0");
+        }
+
+        TopicMessageRequest request = TopicMessageRequest.of(topicId, subscription, limit);
+        TopicMessageResult result = client.executeTopicMessageQuery(request);
+        return result.subscriptionHandle();
+    }
 }

hiero-enterprise-base/src/main/java/com/openelements/hiero/base/protocol/data/TopicMessageRequest.java

@@ -26,4 +26,10 @@ public record TopicMessageRequest(@NonNull TopicId topicId, @NonNull Consumer<To
     public static TopicMessageRequest of(@NonNull TopicId topicId, @NonNull Consumer<TopicMessage> subscription) {
         return new TopicMessageRequest(topicId, subscription, null, null, NO_LIMIT, null, null);
     }
+
+    @NonNull
+    public static TopicMessageRequest of(@NonNull TopicId topicId, @NonNull Consumer<TopicMessage> subscription,
+                                         @NonNull long limit) {
+        return new TopicMessageRequest(topicId, subscription, null, null,limit, null, null);
+    }
 }

hiero-enterprise-spring/src/test/java/com/openelements/hiero/spring/test/TopicClientTest.java

@@ -12,7 +12,6 @@
 import org.springframework.boot.test.context.SpringBootTest;
 
 import java.util.ArrayList;
-import java.util.Arrays;
 import java.util.List;
 
 @SpringBootTest(classes = HieroTestConfig.class)
@@ -210,19 +209,58 @@ void testSubmitMessageThrowExceptionFroInvalidId() {
 
     @Test
     void testSubscribeTopic() throws HieroException {
+        final String msg = "Hello Hiero";
         final List<String> messages = new ArrayList<>();
         final TopicId topicId = topicClient.createTopic();
         hieroTestUtils.waitForMirrorNodeRecords();
 
-        final SubscriptionHandle handle = topicClient.subscribeTopic(topicId, (message) -> {
-            messages.add(message.toString());
-            System.out.println(Arrays.toString(message.contents));
+        final SubscriptionHandle handler = topicClient.subscribeTopic(topicId, (message) -> {
+            messages.add(new String(message.contents));
         });
 
-        topicClient.submitMessage(topicId, "Hello Hiero");
+        topicClient.submitMessage(topicId, msg);
         hieroTestUtils.waitForMirrorNodeRecords();
 
-        Assertions.assertNotNull(handle);
+        Assertions.assertNotNull(handler);
         Assertions.assertEquals(1, messages.size());
+        Assertions.assertEquals(msg,messages.getFirst());
+    }
+
+    @Test
+    void testSubscribeTopicWithLimit() throws HieroException {
+        final String msg = "Hello Hiero";
+        final long limit = 1;
+
+        final List<String> messages = new ArrayList<>();
+        final TopicId topicId = topicClient.createTopic();
+        hieroTestUtils.waitForMirrorNodeRecords();
+
+        final SubscriptionHandle handler = topicClient.subscribeTopic(topicId, (message) -> {
+            messages.add(new String(message.contents));
+        }, limit);
+
+        topicClient.submitMessage(topicId, msg);
+        hieroTestUtils.waitForMirrorNodeRecords();
+
+        topicClient.submitMessage(topicId, msg);
+        hieroTestUtils.waitForMirrorNodeRecords();
+
+        Assertions.assertNotNull(handler);
+        Assertions.assertEquals(limit, messages.size());
+    }
+
+    @Test
+    void testSubscribeTopicWithInvalidLimit() throws HieroException {
+        final String msg = "limit must be greater than 0";
+        final long limit = 0;
+
+        final TopicId topicId = topicClient.createTopic();
+        hieroTestUtils.waitForMirrorNodeRecords();
+
+        final IllegalArgumentException e = Assertions.assertThrows(IllegalArgumentException.class, () -> topicClient.subscribeTopic(
+                topicId, (message) -> {/**/}, limit
+        ));
+
+        Assertions.assertEquals(msg, e.getMessage());
     }
 }