added method to subscribe topic

manishdait · manishdait · commit ad0ee016be2e · 2025-07-01T19:30:25.000+05:30
Signed-off-by: Manish Dait &lt;daitmanish88@gmail.com&gt;
diff --git a/hiero-enterprise-base/src/main/java/com/openelements/hiero/base/HieroContext.java b/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();
 }
diff --git a/hiero-enterprise-base/src/main/java/com/openelements/hiero/base/TopicClient.java b/hiero-enterprise-base/src/main/java/com/openelements/hiero/base/TopicClient.java
@@ -1,10 +1,13 @@
 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.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 +320,24 @@ 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);
+    }
 }
diff --git a/hiero-enterprise-base/src/main/java/com/openelements/hiero/base/config/HieroConfig.java b/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);
diff --git a/hiero-enterprise-base/src/main/java/com/openelements/hiero/base/config/NetworkSettings.java b/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.
      *
diff --git a/hiero-enterprise-base/src/main/java/com/openelements/hiero/base/config/hedera/HederaMainnetSettings.java b/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"));
diff --git a/hiero-enterprise-base/src/main/java/com/openelements/hiero/base/config/hedera/HederaTestnetSettings.java b/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"));
diff --git a/hiero-enterprise-base/src/main/java/com/openelements/hiero/base/implementation/ProtocolLayerClientImpl.java b/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);
         }
diff --git a/hiero-enterprise-base/src/main/java/com/openelements/hiero/base/implementation/TopicClientImpl.java b/hiero-enterprise-base/src/main/java/com/openelements/hiero/base/implementation/TopicClientImpl.java
@@ -1,15 +1,24 @@
 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.util.Objects;
+import java.util.function.Consumer;
 
 public class TopicClientImpl implements TopicClient {
     private final ProtocolLayerClient client;
@@ -107,7 +116,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 +211,14 @@ 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();
+    }
 }
diff --git a/hiero-enterprise-base/src/main/java/com/openelements/hiero/base/protocol/data/TopicMessageResult.java b/hiero-enterprise-base/src/main/java/com/openelements/hiero/base/protocol/data/TopicMessageResult.java
@@ -1,4 +1,14 @@
 package com.openelements.hiero.base.protocol.data;
 
-public record TopicMessageResult() {
+import com.hedera.hashgraph.sdk.Status;
+import com.hedera.hashgraph.sdk.SubscriptionHandle;
+import com.hedera.hashgraph.sdk.TransactionId;
+import org.jspecify.annotations.NonNull;
+
+import java.util.Objects;
+
+public record TopicMessageResult(@NonNull SubscriptionHandle subscriptionHandle) {
+    public TopicMessageResult {
+        Objects.requireNonNull(subscriptionHandle, "subscriptionHandle must not be null");
+    }
 }
diff --git a/hiero-enterprise-base/src/test/java/com/openelements/hiero/base/test/ProtocolLayerClientTests.java b/hiero-enterprise-base/src/test/java/com/openelements/hiero/base/test/ProtocolLayerClientTests.java
@@ -11,6 +11,8 @@
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
+import java.util.Set;
+
 public class ProtocolLayerClientTests {
 
     @Test
@@ -31,6 +33,11 @@ void testNullParams() {
             public @NonNull Client getClient() {
                 return null;
             }
+
+            @Override
+            public @NonNull Set<String> getMirrorNodeEndPoint() {
+                return null;
+            }
         };
         final ProtocolLayerClient client = new ProtocolLayerClientImpl(context);
 
diff --git a/hiero-enterprise-base/src/test/java/com/openelements/hiero/base/test/config/HieroTestContext.java b/hiero-enterprise-base/src/test/java/com/openelements/hiero/base/test/config/HieroTestContext.java
@@ -11,6 +11,8 @@
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Optional;
+import java.util.Set;
+
 import org.jspecify.annotations.NonNull;
 import org.slf4j.Logger;
 
@@ -22,6 +24,8 @@ public class HieroTestContext implements HieroContext {
 
     private final Client client;
 
+    private final Set<String> mirronNodeEnpoint;
+
     public HieroTestContext() {
         final Dotenv dotenv = Dotenv.configure().ignoreIfMissing().load();
 
@@ -57,6 +61,7 @@ public HieroTestContext() {
         final Map<String, AccountId> nodes = new HashMap<>();
         networkSettings.getConsensusNodes()
                 .forEach(consensusNode -> nodes.put(consensusNode.getAddress(), consensusNode.getAccountId()));
+        mirronNodeEnpoint = networkSettings.getConsensusServiceAddress();
         client = Client.forNetwork(nodes);
         if (!networkSettings.getMirrorNodeAddresses().isEmpty()) {
             try {
@@ -76,4 +81,9 @@ public HieroTestContext() {
     public Client getClient() {
         return client;
     }
+
+    @Override
+    public @NonNull Set<String> getMirrorNodeEndPoint() {
+        return mirronNodeEnpoint;
+    }
 }
diff --git a/hiero-enterprise-base/src/test/java/com/openelements/hiero/base/test/config/SoloActionNetworkSettings.java b/hiero-enterprise-base/src/test/java/com/openelements/hiero/base/test/config/SoloActionNetworkSettings.java
@@ -24,6 +24,9 @@ public class SoloActionNetworkSettings implements NetworkSettings {
         return Set.of("http://localhost:8080");
     }
 
+    @Override
+    public @NonNull Set<String> getConsensusServiceAddress() {return Set.of("http://localhost:8080");}
+
     @Override
     public @NonNull Set<ConsensusNode> getConsensusNodes() {
         return Set.of(new ConsensusNode("127.0.0.1", "50211", "0.0.3"));
diff --git a/hiero-enterprise-microprofile/src/main/java/com/openelements/hiero/microprofile/implementation/HieroConfigImpl.java b/hiero-enterprise-microprofile/src/main/java/com/openelements/hiero/microprofile/implementation/HieroConfigImpl.java
@@ -27,6 +27,8 @@ public class HieroConfigImpl implements HieroConfig {
 
     private final Set<String> mirrorNodeAddresses;
 
+    private final Set<String> consensusServiceAddress;
+
     private final Set<ConsensusNode> consensusNodes;
 
     private final Long chainId;
@@ -51,12 +53,14 @@ public HieroConfigImpl(@NonNull final HieroOperatorConfiguration configuration,
             final NetworkSettings settings = networkSettings.get();
             networkName = settings.getNetworkName().orElse(networkConfiguration.getName().orElse(null));
             mirrorNodeAddresses = Collections.unmodifiableSet(settings.getMirrorNodeAddresses());
+            consensusServiceAddress = Collections.unmodifiableSet(settings.getConsensusServiceAddress());
             consensusNodes = Collections.unmodifiableSet(settings.getConsensusNodes());
             chainId = settings.chainId().orElse(null);
             relayUrl = settings.relayUrl().orElse(null);
         } else {
             networkName = networkConfiguration.getName().orElse(null);
             mirrorNodeAddresses = networkConfiguration.getMirrornode().map(Set::of).orElse(Set.of());
+            consensusServiceAddress = Set.of();
             consensusNodes = Collections.unmodifiableSet(networkConfiguration.getNodes());
             chainId = null;
             relayUrl = null;
@@ -83,6 +87,11 @@ public Optional<Duration> getRequestTimeout() {
         return mirrorNodeAddresses;
     }
 
+    @Override
+    public @NonNull Set<String> getConsensusServiceAddress() {
+        return consensusServiceAddress;
+    }
+
     @Override
     public @NonNull Set<ConsensusNode> getConsensusNodes() {
         return consensusNodes;
diff --git a/hiero-enterprise-spring/src/main/java/com/openelements/hiero/spring/implementation/HieroAutoConfiguration.java b/hiero-enterprise-spring/src/main/java/com/openelements/hiero/spring/implementation/HieroAutoConfiguration.java
@@ -34,6 +34,8 @@
 import java.net.URI;
 import java.net.URL;
 import java.util.List;
+import java.util.stream.Collectors;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -110,7 +112,7 @@ TopicClient topicClient(final ProtocolLayerClient protocolLayerClient, HieroCont
             havingValue = "true", matchIfMissing = true)
     MirrorNodeClient mirrorNodeClient(final HieroContext hieroContext) {
         final String mirrorNodeEndpoint;
-        final List<String> mirrorNetwork = hieroContext.getClient().getMirrorNetwork();
+        final List<String> mirrorNetwork = hieroContext.getMirrorNodeEndPoint().stream().toList();
         if (mirrorNetwork.isEmpty()) {
             throw new IllegalArgumentException("Mirror node endpoint must be set");
         }
diff --git a/hiero-enterprise-spring/src/main/java/com/openelements/hiero/spring/implementation/HieroConfigImpl.java b/hiero-enterprise-spring/src/main/java/com/openelements/hiero/spring/implementation/HieroConfigImpl.java
@@ -23,6 +23,8 @@ public class HieroConfigImpl implements HieroConfig {
 
     private final Set<String> mirrorNodeAddresses;
 
+    private final Set<String> consensusServiceAddress;
+
     private final Set<ConsensusNode> consensusNodes;
 
     private final Long chainId;
@@ -50,6 +52,7 @@ public HieroConfigImpl(@NonNull final HieroProperties properties) {
             final NetworkSettings settings = networkSettings.get();
             networkName = settings.getNetworkName().orElse(properties.getNetwork().getName());
             mirrorNodeAddresses = Collections.unmodifiableSet(settings.getMirrorNodeAddresses());
+            consensusServiceAddress = Collections.unmodifiableSet(settings.getConsensusServiceAddress());
             consensusNodes = Collections.unmodifiableSet(settings.getConsensusNodes());
             chainId = settings.chainId().orElse(null);
             relayUrl = settings.relayUrl().orElse(null);
@@ -61,6 +64,7 @@ public HieroConfigImpl(@NonNull final HieroProperties properties) {
             } else {
                 mirrorNodeAddresses = Set.of();
             }
+            consensusServiceAddress = Set.of();
             final List<HieroNode> nodes = properties.getNetwork().getNodes();
             if (nodes == null || nodes.isEmpty()) {
                 consensusNodes = Set.of();
@@ -106,6 +110,11 @@ public Set<String> getMirrorNodeAddresses() {
         return mirrorNodeAddresses;
     }
 
+    @Override
+    public @NonNull Set<String> getConsensusServiceAddress() {
+        return consensusServiceAddress;
+    }
+
     @Override
     public Set<ConsensusNode> getConsensusNodes() {
         return consensusNodes;
diff --git a/hiero-enterprise-spring/src/test/java/com/openelements/hiero/spring/test/TopicClientTest.java b/hiero-enterprise-spring/src/test/java/com/openelements/hiero/spring/test/TopicClientTest.java
diff --git a/hiero-enterprise-test/src/main/java/com/openelements/hiero/test/SoloActionNetworkSettings.java b/hiero-enterprise-test/src/main/java/com/openelements/hiero/test/SoloActionNetworkSettings.java

Original file line number	Diff line number	Diff line change
`@@ -31,6 +31,9 @@ public final class HederaMainnetSettings implements NetworkSettings {`
`31`	`31`	`return Set.of("https://mainnet.mirrornode.hedera.com:443");`
`32`	`32`	`}`
`33`	`33`
	`34`	`+ @Override`
	`35`	`+ public @NonNull Set<String> getConsensusServiceAddress() {return Set.of("mainnet.mirrornode.hedera.com:443");}`
	`36`	`+`
`34`	`37`	`@Override`
`35`	`38`	`public @NonNull Set<ConsensusNode> getConsensusNodes() {`
`36`	`39`	`return Set.of(new ConsensusNode("35.186.191.247", "50211", "0.0.4"));`
Original file line number	Diff line number	Diff line change
`@@ -434,7 +434,7 @@ public TopicMessageResult executeTopicMessageQuery(TopicMessageRequest request)`
`434`	`434`	`query.setLimit(request.limit());`
`435`	`435`	`}`
`436`	`436`	`final SubscriptionHandle subscribe = query.subscribe(hieroContext.getClient(), request.subscription());`
`437`		`- return new TopicMessageResult();`
	`437`	`+ return new TopicMessageResult(subscribe);`
`438`	`438`	`} catch (final Exception e) {`
`439`	`439`	`throw new HieroException("Failed to execute query message transaction", e);`
`440`	`440`	`}`