feat: Added subscribeTopic method to TopicClient

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

@@ -4,6 +4,8 @@
 import com.openelements.hiero.base.data.Account;
 import org.jspecify.annotations.NonNull;
 
+import java.util.Set;
+
 /**
  * Context for a specific Hiero connection to a network.
  */
@@ -25,4 +27,7 @@ public interface HieroContext {
      */
     @NonNull
     Client getClient();
+
+    @NonNull
+    Set<String> getMirrorNodeEndPoint();
 }

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

@@ -1,10 +1,14 @@
 package com.openelements.hiero.base;
 
 import com.hedera.hashgraph.sdk.PrivateKey;
+import com.hedera.hashgraph.sdk.SubscriptionHandle;
 import com.hedera.hashgraph.sdk.TopicId;
+import com.hedera.hashgraph.sdk.TopicMessage;
 import org.jspecify.annotations.NonNull;
 
+import java.time.Instant;
 import java.util.Objects;
+import java.util.function.Consumer;
 
 /**
  * Interface for interacting with a Hiero network. This interface provides methods for interacting with Hiero Topic,
@@ -317,4 +321,106 @@ default void submitMessage(@NonNull String topicId, @NonNull String submitKey, @
         Objects.requireNonNull(message, "message cannot be null");
         submitMessage(TopicId.fromString(topicId), PrivateKey.fromString(submitKey), message);
     };
+
+    /**
+     * Subscribe to a Topic
+     *
+     * @param topicId the topicId of topic
+     * @return SubscriptionHandle for the Topic
+     * @throws HieroException if Topic could not be subscribed
+     */
+    SubscriptionHandle subscribeTopic(@NonNull TopicId topicId, @NonNull Consumer<TopicMessage> subscription)
+            throws HieroException;
+
+    /**
+     * Subscribe to a Topic
+     *
+     * @param topicId the topicId of topic
+     * @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 {
+        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);
+    }
+
+    /**
+     * Subscribe to a Topic
+     *
+     * @param topicId the topicId of topic
+     * @param startTime time to start subscribing to a topic
+     * @param endTime time to stop subscribing to a topic
+     * @return SubscriptionHandle for the Topic
+     * @throws HieroException if Topic could not be subscribed
+     */
+    SubscriptionHandle subscribeTopic(@NonNull TopicId topicId, @NonNull Consumer<TopicMessage> subscription,
+                                      @NonNull Instant startTime, @NonNull Instant endTime) throws HieroException;
+
+    /**
+     * Subscribe to a Topic
+     *
+     * @param topicId the topicId of topic
+     * @param startTime time to start subscribing to a topic
+     * @param endTime time to stop subscribing to a topic
+     * @return SubscriptionHandle for the Topic
+     * @throws HieroException if Topic could not be subscribed
+     */
+    default SubscriptionHandle subscribeTopic(@NonNull String topicId, @NonNull Consumer<TopicMessage> subscription,
+                                              @NonNull Instant startTime, @NonNull Instant endTime) throws HieroException {
+        return subscribeTopic(TopicId.fromString(topicId), subscription, startTime, endTime);
+    }
+
+    /**
+     * Subscribe to a Topic
+     *
+     * @param topicId the topicId of topic
+     * @param startTime time to start subscribing to a topic
+     * @param endTime time to stop subscribing to a 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,
+                                      @NonNull Instant startTime, @NonNull Instant endTime, long limit)
+            throws HieroException;
+
+    /**
+     * Subscribe to a Topic
+     *
+     * @param topicId the topicId of topic
+     * @param startTime time to start subscribing to a topic
+     * @param endTime time to stop subscribing to a 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,
+                                              @NonNull Instant startTime, @NonNull Instant endTime, long limit)
+            throws HieroException {
+        return subscribeTopic(TopicId.fromString(topicId), subscription, startTime, endTime, limit);
+    }
 }

hiero-enterprise-base/src/main/java/com/openelements/hiero/base/config/HieroConfig.java

@@ -41,6 +41,9 @@ public interface HieroConfig {
     @NonNull
     Set<String> getMirrorNodeAddresses();
 
+    @NonNull
+    Set<String> getConsensusServiceAddress();
+
     /**
      * Returns the consensus nodes.
      *
@@ -88,6 +91,11 @@ default HieroContext createHieroContext() {
             public @NonNull Client getClient() {
                 return client;
             }
+
+            @Override
+            public @NonNull Set<String> getMirrorNodeEndPoint() {
+                return getMirrorNodeAddresses();
+            }
         };
     }
 
@@ -102,7 +110,7 @@ default Client createClient() {
             final Map<String, AccountId> nodes = getConsensusNodes().stream()
                     .collect(Collectors.toMap(n -> n.getAddress(), n -> n.getAccountId()));
             final Client client = Client.forNetwork(nodes);
-            final List<String> mirrorNodeAddresses = getMirrorNodeAddresses().stream().collect(Collectors.toList());
+            final List<String> mirrorNodeAddresses = getConsensusServiceAddress().stream().collect(Collectors.toList());
             client.setMirrorNetwork(mirrorNodeAddresses);
             client.setOperator(getOperatorAccount().accountId(), getOperatorAccount().privateKey());
             getRequestTimeout().ifPresent(client::setRequestTimeout);

hiero-enterprise-base/src/main/java/com/openelements/hiero/base/config/NetworkSettings.java

@@ -40,6 +40,14 @@ public interface NetworkSettings {
     @NonNull
     Set<String> getMirrorNodeAddresses();
 
+    /**
+     * Returns the consensus service address.
+     *
+     * @return the consensus service addresses
+     */
+    @NonNull
+    Set<String> getConsensusServiceAddress();
+
     /**
      * Returns the consensus nodes.
      *

hiero-enterprise-base/src/main/java/com/openelements/hiero/base/config/hedera/HederaMainnetSettings.java

@@ -31,6 +31,9 @@ public final class HederaMainnetSettings implements NetworkSettings {
         return Set.of("https://mainnet.mirrornode.hedera.com:443");
     }
 
+    @Override
+    public @NonNull Set<String> getConsensusServiceAddress() {return Set.of("mainnet.mirrornode.hedera.com:443");}
+
     @Override
     public @NonNull Set<ConsensusNode> getConsensusNodes() {
         return Set.of(new ConsensusNode("35.186.191.247", "50211", "0.0.4"));

hiero-enterprise-base/src/main/java/com/openelements/hiero/base/config/hedera/HederaTestnetSettings.java

@@ -31,6 +31,9 @@ public final class HederaTestnetSettings implements NetworkSettings {
         return Set.of("https://testnet.mirrornode.hedera.com:443");
     }
 
+    @Override
+    public @NonNull Set<String> getConsensusServiceAddress() {return Set.of("testnet.mirrornode.hedera.com:443");}
+
     @Override
     public @NonNull Set<ConsensusNode> getConsensusNodes() {
         return Set.of(new ConsensusNode("0.testnet.hedera.com", "50211", "0.0.3"));

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

@@ -17,6 +17,7 @@ public class EnvBasedHieroConfig implements HieroConfig {
     private final PrivateKey operatorPrivateKey;
     private final String mirrorNodeAddress;
 
+    private final String consensusServiceAddress;
     private final String consensusNodeIp;
     private final String consensusNodePort;
     private final String consensusNodeAccount;
@@ -37,6 +38,8 @@ public EnvBasedHieroConfig() {
                 .orElseThrow(() -> new IllegalStateException("HEDERA_OPERATOR_PRIVATE_KEY is not set"));
         mirrorNodeAddress = getEnv("HEDERA_MIRROR_NODE_ADDRESS")
                 .orElse(null);
+        consensusServiceAddress = getEnv("HEDERA_CONSENSUS_SERVICE_ADDRESS")
+                .orElseThrow(() -> new IllegalStateException("HEDERA_CONSENSUS_SERVICE_ADDRESS is not set"));
         consensusNodeIp = getEnv("HEDERA_CONSENSUS_NODE_IP")
                 .orElseThrow(() -> new IllegalStateException("HEDERA_CONSENSUS_NODE_IP is not set"));
         consensusNodePort = getEnv("HEDERA_CONSENSUS_NODE_PORT")
@@ -71,6 +74,11 @@ private Optional<String> getEnv(String key) {
         return Set.of(mirrorNodeAddress);
     }
 
+    @Override
+    public @NonNull Set<String> getConsensusServiceAddress() {
+        return Set.of(consensusServiceAddress);
+    }
+
     @Override
     public @NonNull Set<ConsensusNode> getConsensusNodes() {
         ConsensusNode node = new ConsensusNode(

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

@@ -37,6 +37,11 @@ public NetworkSettingsBasedHieroConfig(@NonNull final Account operatorAccount, @
         return networkSetting.getMirrorNodeAddresses();
     }
 
+    @Override
+    public @NonNull Set<String> getConsensusServiceAddress() {
+        return networkSetting.getConsensusServiceAddress();
+    }
+
     @Override
     public @NonNull Set<ConsensusNode> getConsensusNodes() {
         return networkSetting.getConsensusNodes();

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

@@ -434,7 +434,7 @@ public TopicMessageResult executeTopicMessageQuery(TopicMessageRequest request)
                 query.setLimit(request.limit());
             }
             final SubscriptionHandle subscribe = query.subscribe(hieroContext.getClient(), request.subscription());
-            return new TopicMessageResult();
+            return new TopicMessageResult(subscribe);
         } catch (final Exception e) {
             throw new HieroException("Failed to execute query message transaction", e);
         }

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

@@ -1,15 +1,25 @@
 package com.openelements.hiero.base.implementation;
 
 import com.hedera.hashgraph.sdk.PrivateKey;
+import com.hedera.hashgraph.sdk.SubscriptionHandle;
 import com.hedera.hashgraph.sdk.TopicId;
+import com.hedera.hashgraph.sdk.TopicMessage;
 import com.openelements.hiero.base.HieroException;
 import com.openelements.hiero.base.TopicClient;
 import com.openelements.hiero.base.data.Account;
 import com.openelements.hiero.base.protocol.ProtocolLayerClient;
-import com.openelements.hiero.base.protocol.data.*;
+import com.openelements.hiero.base.protocol.data.TopicCreateRequest;
+import com.openelements.hiero.base.protocol.data.TopicCreateResult;
+import com.openelements.hiero.base.protocol.data.TopicUpdateRequest;
+import com.openelements.hiero.base.protocol.data.TopicDeleteRequest;
+import com.openelements.hiero.base.protocol.data.TopicSubmitMessageRequest;
+import com.openelements.hiero.base.protocol.data.TopicMessageRequest;
+import com.openelements.hiero.base.protocol.data.TopicMessageResult;
 import org.jspecify.annotations.NonNull;
 
+import java.time.Instant;
 import java.util.Objects;
+import java.util.function.Consumer;
 
 public class TopicClientImpl implements TopicClient {
     private final ProtocolLayerClient client;
@@ -107,7 +117,7 @@ public void updateTopic(@NonNull TopicId topicId, @NonNull PrivateKey updatedAdm
         Objects.requireNonNull(topicId, "topicId must not be null");
         Objects.requireNonNull(submitKey, "submitKey must not be null");
         Objects.requireNonNull(memo, "memo must not be null");
-       updateTopic(topicId, operationalAccount.privateKey(), updatedAdminKey, submitKey, memo);
+        updateTopic(topicId, operationalAccount.privateKey(), updatedAdminKey, submitKey, memo);
     }
 
     @Override
@@ -202,4 +212,72 @@ public void submitMessage(@NonNull TopicId topicId, @NonNull PrivateKey submitKe
         TopicSubmitMessageRequest request = TopicSubmitMessageRequest.of(topicId, submitKey, message);
         client.executeTopicMessageSubmitTransaction(request);
     }
+
+    @Override
+    public SubscriptionHandle subscribeTopic(@NonNull TopicId topicId, @NonNull Consumer<TopicMessage> subscription)
+            throws HieroException {
+        Objects.requireNonNull(topicId, "topicId must not be null");
+        Objects.requireNonNull(subscription, "subscription must not be null");
+        TopicMessageRequest request = TopicMessageRequest.of(topicId, subscription);
+        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();
+    }
+
+    @Override
+    public SubscriptionHandle subscribeTopic(@NonNull TopicId topicId, @NonNull Consumer<TopicMessage> subscription,
+                                             Instant startTime, Instant endTime) throws HieroException {
+        Objects.requireNonNull(topicId, "topicId must not be null");
+        Objects.requireNonNull(subscription, "subscription must not be null");
+        Objects.requireNonNull(startTime, "startTime must not be null");
+        Objects.requireNonNull(endTime, "endTime must not be null");
+
+        if (startTime.isBefore(Instant.now())) {
+            throw  new IllegalArgumentException("startTime must be greater than currentTime");
+        }
+        if (endTime.isBefore(startTime)) {
+            throw  new IllegalArgumentException("endTime must be greater than startTime");
+        }
+
+        TopicMessageRequest request = TopicMessageRequest.of(topicId, subscription, startTime, endTime);
+        TopicMessageResult result = client.executeTopicMessageQuery(request);
+        return result.subscriptionHandle();
+    }
+
+    @Override
+    public SubscriptionHandle subscribeTopic(@NonNull TopicId topicId, @NonNull Consumer<TopicMessage> subscription,
+                                             @NonNull Instant startTime, @NonNull Instant endTime, long limit)
+            throws HieroException {
+        Objects.requireNonNull(topicId, "topicId must not be null");
+        Objects.requireNonNull(subscription, "subscription must not be null");
+        Objects.requireNonNull(startTime, "startTime must not be null");
+        Objects.requireNonNull(endTime, "endTime must not be null");
+
+        if (startTime.isBefore(Instant.now())) {
+            throw  new IllegalArgumentException("startTime must be greater than currentTime");
+        }
+        if (endTime.isBefore(startTime)) {
+            throw  new IllegalArgumentException("endTime must be greater than startTime");
+        }
+        if (limit == 0) {
+            throw new IllegalArgumentException("limit must be greater than 0");
+        }
+
+        TopicMessageRequest request = TopicMessageRequest.of(topicId, subscription, startTime, endTime, 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,22 @@ 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);
+    }
+
+    @NonNull
+    public static TopicMessageRequest of(@NonNull TopicId topicId, @NonNull Consumer<TopicMessage> subscription,
+                                         @NonNull Instant startTime, @NonNull Instant endTime) {
+        return new TopicMessageRequest(topicId, subscription, startTime, endTime, NO_LIMIT, null, null);
+    }
+
+    @NonNull
+    public static TopicMessageRequest of(@NonNull TopicId topicId, @NonNull Consumer<TopicMessage> subscription,
+                                         @NonNull Instant startTime, @NonNull Instant endTime, long limit) {
+        return new TopicMessageRequest(topicId, subscription, startTime, endTime, limit, null, null);
+    }
 }