Support multiple outbound topics for publishing blockchain events

Signed-off-by: Abhay Kishore <[email protected]>
hyperledger-labs · Aug 16, 2024 · 41a78ac · 41a78ac
1 parent 41078ec
commit 41a78ac
Show file tree

Hide file tree

Showing 7 changed files with 100 additions and 33 deletions.
diff --git a/src/main/java/hlf/java/rest/client/config/FabricProperties.java b/src/main/java/hlf/java/rest/client/config/FabricProperties.java
@@ -68,14 +68,21 @@ public static class Events {
     // preferred for providing Chaincode details for Event subscription
     private List<String> chaincode;
     private boolean standardCCEventEnabled;
-    private List<String> block;
+    private List<BlockDetails> blockDetails;
     private List<ChaincodeDetails> chaincodeDetails;
   }
 
+  @Data
+  public static class BlockDetails {
+    private String channelName;
+    private List<String> listenerTopics;
+  }
+
   @Data
   public static class ChaincodeDetails {
     private String channelName;
     private String chaincodeId;
+    private List<String> listenerTopics;
   }
 
   /**

diff --git a/src/main/java/hlf/java/rest/client/listener/FabricEventListener.java b/src/main/java/hlf/java/rest/client/listener/FabricEventListener.java
@@ -47,13 +47,15 @@ public void onRefresh(RefreshScopeRefreshedEvent event) {
     startEventListener();
   }
 
-  public void startEventListener() {
+  private void startEventListener() {
 
     try {
-      List<String> blockChannelNames = fabricProperties.getEvents().getBlock();
-      if (!CollectionUtils.isEmpty(blockChannelNames)) {
+      List<FabricProperties.BlockDetails> blockDetailsList =
+          fabricProperties.getEvents().getBlockDetails();
+      if (!CollectionUtils.isEmpty(blockDetailsList)) {
 
-        for (String channelName : blockChannelNames) {
+        for (FabricProperties.BlockDetails blockDetails : blockDetailsList) {
+          String channelName = blockDetails.getChannelName();
           log.info("channel names {}", channelName);
           Network network = gateway.getNetwork(channelName);
 

diff --git a/src/main/java/hlf/java/rest/client/service/EventPublishService.java b/src/main/java/hlf/java/rest/client/service/EventPublishService.java
@@ -25,9 +25,8 @@ boolean sendMessage(
    * @param eventName String chaincode event-name
    * @param channelName String Name of the channel where the event was generated.
    * @param messageKey associated key for the payload.
-   * @return status boolean status of msg sent
    */
-  boolean publishChaincodeEvents(
+  void publishChaincodeEvents(
       final String payload,
       String chaincodeName,
       String fabricTxId,
@@ -42,9 +41,8 @@ boolean publishChaincodeEvents(
    * @param channelName String Name of the channel where the event was generated.
    * @param functionName String Name of the function name.
    * @param isPrivateDataPresent boolean flag to check if privateData present in payload
-   * @return status boolean status of msg sent
    */
-  boolean publishBlockEvents(
+  void publishBlockEvents(
       final String payload,
       String fabricTxId,
       String channelName,

diff --git a/src/main/java/hlf/java/rest/client/service/impl/EventPublishServiceImpl.java b/src/main/java/hlf/java/rest/client/service/impl/EventPublishServiceImpl.java
@@ -4,6 +4,7 @@
 import hlf.java.rest.client.config.KafkaProperties;
 import hlf.java.rest.client.service.EventPublishService;
 import hlf.java.rest.client.util.FabricClientConstants;
+import java.util.Optional;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.common.header.internals.RecordHeader;
@@ -82,18 +83,50 @@ public void onFailure(Throwable ex) {
   }
 
   @Override
-  public boolean publishChaincodeEvents(
+  public void publishChaincodeEvents(
       String payload,
       String chaincodeName,
       String fabricTxId,
       String eventName,
       String channelName,
       String messageKey) {
-    boolean status = true;
 
+    Optional<FabricProperties.ChaincodeDetails> optionalChaincodeDetails =
+        fabricProperties.getEvents().getChaincodeDetails().stream()
+            .filter(
+                chaincodeDetail ->
+                    chaincodeDetail.getChannelName().equals(channelName)
+                        && chaincodeDetail.getChaincodeId().equals(chaincodeName))
+            .findAny();
+
+    if (!optionalChaincodeDetails.isPresent()) {
+      sendMessage(
+          kafkaProperties.getEventListener().getTopic(),
+          payload,
+          chaincodeName,
+          fabricTxId,
+          eventName,
+          channelName,
+          messageKey);
+      return;
+    }
+
+    for (String topic : optionalChaincodeDetails.get().getListenerTopics()) {
+      sendMessage(topic, payload, chaincodeName, fabricTxId, eventName, channelName, messageKey);
+    }
+  }
+
+  private void sendMessage(
+      String topic,
+      String payload,
+      String chaincodeName,
+      String fabricTxId,
+      String eventName,
+      String channelName,
+      String messageKey) {
     try {
       ProducerRecord<String, String> producerRecord =
-          new ProducerRecord<>(kafkaProperties.getEventListener().getTopic(), messageKey, payload);
+          new ProducerRecord<>(topic, messageKey, payload);
 
       producerRecord
           .headers()
@@ -118,9 +151,7 @@ public boolean publishChaincodeEvents(
                   FabricClientConstants.FABRIC_EVENT_TYPE,
                   FabricClientConstants.FABRIC_EVENT_TYPE_CHAINCODE.getBytes()));
 
-      log.info(
-          "Publishing Chaincode event to outbound topic {}",
-          kafkaProperties.getEventListener().getTopic());
+      log.info("Publishing Chaincode event to outbound topic {}", topic);
 
       ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(producerRecord);
 
@@ -146,30 +177,60 @@ public void onFailure(Throwable ex) {
           });
 
     } catch (Exception ex) {
-      status = false;
       log.error("Error sending message - " + ex.getMessage());
     }
-
-    return status;
   }
 
   @Override
-  public boolean publishBlockEvents(
+  public void publishBlockEvents(
       String payload,
       String fabricTxId,
       String channelName,
       String chaincodeName,
       String functionName,
       Boolean isPrivateDataPresent) {
-    boolean status = true;
 
+    Optional<FabricProperties.BlockDetails> optionalBlockDetails =
+        fabricProperties.getEvents().getBlockDetails().stream()
+            .filter(blockDetails -> blockDetails.getChannelName().equals(channelName))
+            .findAny();
+
+    if (!optionalBlockDetails.isPresent()) {
+      sendMessage(
+          kafkaProperties.getEventListener().getTopic(),
+          payload,
+          fabricTxId,
+          channelName,
+          chaincodeName,
+          functionName,
+          isPrivateDataPresent);
+      return;
+    }
+
+    for (String topic : optionalBlockDetails.get().getListenerTopics()) {
+      sendMessage(
+          topic,
+          payload,
+          fabricTxId,
+          channelName,
+          chaincodeName,
+          functionName,
+          isPrivateDataPresent);
+    }
+  }
+
+  private void sendMessage(
+      String topic,
+      String payload,
+      String fabricTxId,
+      String channelName,
+      String chaincodeName,
+      String functionName,
+      Boolean isPrivateDataPresent) {
     try {
 
       ProducerRecord<String, String> producerRecord =
-          new ProducerRecord<>(
-              kafkaProperties.getEventListener().getTopic(),
-              String.valueOf(payload.hashCode()),
-              payload);
+          new ProducerRecord<>(topic, String.valueOf(payload.hashCode()), payload);
 
       producerRecord
           .headers()
@@ -204,9 +265,7 @@ public boolean publishBlockEvents(
                   FabricClientConstants.IS_PRIVATE_DATA_PRESENT,
                   isPrivateDataPresent.toString().getBytes()));
 
-      log.info(
-          "Publishing Block event to outbound topic {}",
-          kafkaProperties.getEventListener().getTopic());
+      log.info("Publishing Block event to outbound topic {}", topic);
 
       ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(producerRecord);
 
@@ -232,10 +291,7 @@ public void onFailure(Throwable ex) {
           });
 
     } catch (Exception ex) {
-      status = false;
       log.error("Error sending message - " + ex.getMessage());
     }
-
-    return status;
   }
 }
diff --git a/src/test/java/hlf/java/rest/client/integration/ConfigHandlerControllerIntegrationTest.java b/src/test/java/hlf/java/rest/client/integration/ConfigHandlerControllerIntegrationTest.java
@@ -81,7 +81,7 @@ private static class TestConfiguration {
     static String FABRIC_PROPERTIES_CLIENT =
         "FabricProperties.Client(rest=FabricProperties.Client.Rest(apikey=expected-key))";
     static String FABRIC_PROPERTIES_EVENTS =
-        "FabricProperties.Events(enable=true, chaincode=[chaincode12, chaincode2], standardCCEventEnabled=false, block=[block111, block2], chaincodeDetails=null)";
+        "FabricProperties.Events(enable=true, chaincode=[chaincode12, chaincode2], standardCCEventEnabled=false, blockDetails=[FabricProperties.BlockDetails(channelName=block111, listenerTopics=[topic-1])], chaincodeDetails=null)";
     static String KAFKA_PROPERTIES_PRODUCER =
         "Producer{brokerHost='localhost:8087', topic='hlf-offchain-topic1', saslJaasConfig='null'}";
     static String KAFKA_CONSUMER_PROPERTIES =

diff --git a/src/test/resources/application.yml b/src/test/resources/application.yml
@@ -67,7 +67,9 @@ fabric:
   events:
     enable: false
     chaincode: channel1
-    block: channel1
+    blockDetails:
+      - channelName: channel1
+        listenerTopics: topic-1
   client:
     rest:
       apikey: abc
diff --git a/src/test/resources/integration/sample-application.yml b/src/test/resources/integration/sample-application.yml
@@ -38,7 +38,9 @@ fabric:
     enable: true
     standardCCEventEnabled: false
     chaincode: chaincode12, chaincode2
-    block: block111, block2
+    blockDetails:
+      - channelName: block111
+        listenerTopics: topic-1
 kafka:
   integration-points:
     -