Event relayer as a messaging layer for Java sdk

.circleci/config.yml

@@ -427,8 +427,8 @@ jobs:
             echo; git -P reflog --date=iso
             go version
       - run:
-          name: Build production code
-          command: make build
+          name: Build production code and test code
+          command: make buildall
           working_directory: ~/project/services/relayer
 
   relayer-test:

applications/20acts/javatest/src/intTest/java/net/consensys/gpact/applications/twentyacts/helpers/TwentyActsExampleSystemManager.java

@@ -196,7 +196,7 @@ private void addBcTxRootSign(
         creds,
         bc,
         ((TxRootTransferManagerGroup) messagingManagerGroup).getTxRootContractAddress(bc.bcId));
-    crossControlManagerGroup.addBlockchainAndDeployContracts(
-        creds, bc, txRootTransferGroup.getVerifier(bc.bcId));
+    crossControlManagerGroup.addBlockchainAndDeployContracts(creds, bc);
+    crossControlManagerGroup.setMessageVerifier(bc.bcId, txRootTransferGroup.getVerifier(bc.bcId));
   }
 }

applications/20acts/javatest/src/main/java/net/consensys/gpact/applications/twentyacts/crosscontrol/TwentyActsManagerGroup.java

@@ -33,10 +33,7 @@ public class TwentyActsManagerGroup implements CrossControlManagerGroup {
   private Map<BlockchainId, BcHolder> blockchains = new HashMap<>();
 
   @Override
-  public void addBlockchainAndDeployContracts(
-      Credentials creds,
-      BlockchainConfig bcInfo,
-      MessagingVerificationInterface messageVerification)
+  public void addBlockchainAndDeployContracts(Credentials creds, BlockchainConfig bcInfo)
       throws Exception {
     BlockchainId blockchainId = bcInfo.bcId;
     if (this.blockchains.containsKey(blockchainId)) {
@@ -48,18 +45,13 @@ public void addBlockchainAndDeployContracts(
     holder.cbc = new TwentyActsManager(creds, bcInfo);
     holder.cbc.deployCbcContract();
     holder.cbcContractAddress = holder.cbc.getCbcContractAddress();
-    holder.ver = messageVerification;
 
     this.blockchains.put(blockchainId, holder);
   }
 
   @Override
   public void addBlockchainAndLoadCbcContract(
-      Credentials creds,
-      BlockchainConfig bcInfo,
-      String cbcAddress,
-      MessagingVerificationInterface messageVerification)
-      throws Exception {
+      Credentials creds, BlockchainConfig bcInfo, String cbcAddress) throws Exception {
     BlockchainId blockchainId = bcInfo.bcId;
     if (this.blockchains.containsKey(blockchainId)) {
       return;
@@ -70,11 +62,17 @@ public void addBlockchainAndLoadCbcContract(
 
     holder.cbc.loadCbcContract(cbcAddress);
     holder.cbcContractAddress = cbcAddress;
-    holder.ver = messageVerification;
 
     this.blockchains.put(blockchainId, holder);
   }
 
+  @Override
+  public void setMessageVerifier(
+      final BlockchainId bcId, final MessagingVerificationInterface messageVerification) {
+    BcHolder holder = this.blockchains.get(bcId);
+    holder.ver = messageVerification;
+  }
+
   @Override
   public CrosschainCallResult executeCrosschainCall(
       String executionEngine, CallExecutionTree callTree, long timeout)

examples/gpact/erc20bridge/java/src/main/java/net/consensys/gpact/examples/gpact/erc20bridge/Erc20User.java

@@ -74,9 +74,11 @@ public void createCbcManager(
     this.crossControlManagerGroup =
         CrosschainProtocols.getFunctionCallInstance(CrosschainProtocols.GPACT).get();
     this.crossControlManagerGroup.addBlockchainAndLoadCbcContract(
-        this.creds, bcInfoA, cbcContractAddressOnBcA, msgVerA);
+        this.creds, bcInfoA, cbcContractAddressOnBcA);
+    this.crossControlManagerGroup.setMessageVerifier(bcInfoA.bcId, msgVerA);
     this.crossControlManagerGroup.addBlockchainAndLoadCbcContract(
-        this.creds, bcInfoB, cbcContractAddressOnBcB, msgVerB);
+        this.creds, bcInfoB, cbcContractAddressOnBcB);
+    this.crossControlManagerGroup.setMessageVerifier(bcInfoB.bcId, msgVerB);
 
     this.bcInfoA = bcInfoA;
     this.bcInfoB = bcInfoB;

examples/gpact/hotel-train/java/src/main/java/net/consensys/gpact/examples/gpact/hoteltrain/EntityTravelAgency.java

@@ -97,11 +97,14 @@ public void createCbcManager(
     this.crossControlManagerGroup =
         CrosschainProtocols.getFunctionCallInstance(CrosschainProtocols.GPACT).get();
     this.crossControlManagerGroup.addBlockchainAndLoadCbcContract(
-        this.credentials, bcInfoTravel, cbcContractAddressOnBcTravel, msgVerTravel);
+        this.credentials, bcInfoTravel, cbcContractAddressOnBcTravel);
+    this.crossControlManagerGroup.setMessageVerifier(bcInfoTravel.bcId, msgVerTravel);
     this.crossControlManagerGroup.addBlockchainAndLoadCbcContract(
-        this.credentials, bcInfoHotel, cbcContractAddressOnBcHotel, msgVerHotel);
+        this.credentials, bcInfoHotel, cbcContractAddressOnBcHotel);
+    this.crossControlManagerGroup.setMessageVerifier(bcInfoHotel.bcId, msgVerHotel);
     this.crossControlManagerGroup.addBlockchainAndLoadCbcContract(
-        this.credentials, bcInfoTrain, cbcContractAddressOnBcTrain, msgVerTrain);
+        this.credentials, bcInfoTrain, cbcContractAddressOnBcTrain);
+    this.crossControlManagerGroup.setMessageVerifier(bcInfoTrain.bcId, msgVerTrain);
   }
 
   public BigInteger book(final int dateInt, GpactExampleSystemManager exampleManager)
@@ -149,7 +152,7 @@ public BigInteger book(final int dateInt, GpactExampleSystemManager exampleManag
 
   public void grantAllowance(EntityBase entity, int amount, String spender) throws Exception {
     TransactionReceiptProcessor txrProcessor =
-        new PollingTransactionReceiptProcessor(entity.web3j, this.pollingInterval, RETRY);
+        new PollingTransactionReceiptProcessor(entity.web3j, this.pollingIntervalMs, RETRY);
     FastTxManager atm =
         TxManagerCache.getOrCreate(
             entity.web3j, this.credentials, entity.getBlockchainId().asLong(), txrProcessor);

examples/helpers/src/main/java/net/consensys/gpact/helpers/AbstractExampleTest.java

@@ -16,20 +16,28 @@ public abstract class AbstractExampleTest {
   public static final String MSG_STORE_FROM_DISPATCHER = "msgstore:8080";
   public static final String MSG_STORE_FROM_USER = "127.0.0.1:8080";
 
+  public enum MessagingType {
+    EVENT_SIGNING,
+    TRANSACTION_RECEIPT_SIGNING,
+    EVENT_RELAY
+  }
+
   protected String createPropertiesFile(
       boolean useDirectSigning, boolean serialExecution, boolean oneBlockchain) throws IOException {
+    MessagingType msgType =
+        useDirectSigning ? MessagingType.EVENT_SIGNING : MessagingType.TRANSACTION_RECEIPT_SIGNING;
+    return createPropertiesFile(msgType, serialExecution, oneBlockchain);
+  }
+
+  protected String createPropertiesFile(
+      MessagingType msgType, boolean serialExecution, boolean oneBlockchain) throws IOException {
     File file = File.createTempFile("temp", null);
     //    file.deleteOnExit();
 
     Properties props = new Properties();
 
     props.setProperty("RELAYER_URI", "http://127.0.0.1:9625");
-
-    if (useDirectSigning) {
-      props.setProperty("CONSENSUS_METHODOLOGY", "EVENT_SIGNING");
-    } else {
-      props.setProperty("CONSENSUS_METHODOLOGY", "TRANSACTION_RECEIPT_SIGNING");
-    }
+    props.setProperty("CONSENSUS_METHODOLOGY", msgType.name());
 
     if (serialExecution) {
       props.setProperty("EXECUTION_ENGINE", "SERIAL");

examples/helpers/src/main/java/net/consensys/gpact/helpers/BaseExampleSystemManager.java

@@ -16,26 +16,21 @@
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Set;
 import net.consensys.gpact.CrosschainProtocols;
-import net.consensys.gpact.common.AnIdentity;
-import net.consensys.gpact.common.BlockchainConfig;
-import net.consensys.gpact.common.BlockchainId;
-import net.consensys.gpact.common.StatsHolder;
+import net.consensys.gpact.common.*;
 import net.consensys.gpact.functioncall.CrossControlManager;
 import net.consensys.gpact.functioncall.CrossControlManagerGroup;
 import net.consensys.gpact.messaging.MessagingManagerGroup;
 import net.consensys.gpact.messaging.common.attestorrelayer.AttestorRelayer;
 import net.consensys.gpact.messaging.eventattest.AttestorSignerGroup;
+import net.consensys.gpact.messaging.eventrelay.EventRelayGroup;
 import net.consensys.gpact.messaging.txrootrelay.TxRootRelayerGroup;
 import net.consensys.gpact.messaging.txrootrelay.TxRootTransferGroup;
 import net.consensys.gpact.messaging.txrootrelay.TxRootTransferManagerGroup;
-import org.apache.logging.log4j.LogManager;
-import org.apache.logging.log4j.Logger;
 import org.web3j.crypto.Credentials;
 
 public abstract class BaseExampleSystemManager {
-  static final Logger LOG = LogManager.getLogger(BaseExampleSystemManager.class);
-
   private final String propsFileName;
 
   protected BlockchainConfig root;
@@ -50,11 +45,6 @@ public BaseExampleSystemManager(String propertiesFileName) {
   }
 
   public void standardExampleConfig(int numberOfBlockchains) throws Exception {
-    standardExampleConfig(numberOfBlockchains, true);
-  }
-
-  public void standardExampleConfig(int numberOfBlockchains, boolean fetchFromStore)
-      throws Exception {
     // Less than two blockchains doesn't make sense for crosschain.
     // The test infrasturcture only supports three blockchains at present.
     if (!(numberOfBlockchains == 2 || numberOfBlockchains == 3)) {
@@ -72,13 +62,15 @@ public void standardExampleConfig(int numberOfBlockchains, boolean fetchFromStor
     StatsHolder.log(consensusMethodology.name());
     String relayerUri = propsLoader.getRelayerUri();
 
-    loadFunctionPayerProperties(propsLoader);
+    loadFunctionLayerProperties(propsLoader);
 
     // To keep the example simple, just have one signer for all blockchains.
     AnIdentity globalSigner = AnIdentity.createNewRandomIdentity();
 
     this.crossControlManagerGroup = getFunctionCallInstance();
 
+    List<AttestorRelayer.Source> sources = new ArrayList<>();
+
     // Set-up GPACT contracts: Deploy Crosschain Control and Registrar contracts on
     // each blockchain.
     switch (consensusMethodology) {
@@ -87,8 +79,6 @@ public void standardExampleConfig(int numberOfBlockchains, boolean fetchFromStor
         this.messagingManagerGroup =
             CrosschainProtocols.getMessagingInstance(CrosschainProtocols.ATTESTOR).get();
 
-        List<AttestorRelayer.Source> sources = new ArrayList<>();
-
         addBcAttestorSign(
             messagingManagerGroup,
             crossControlManagerGroup,
@@ -113,19 +103,51 @@ public void standardExampleConfig(int numberOfBlockchains, boolean fetchFromStor
               this.bc3,
               sources);
         }
+
         attestorSignerGroup.configureRelayer(
             globalSigner,
             relayerUri,
             sources,
             this.root.dispatcherUri,
             this.root.msgStoreUrlFromDispatcher,
             this.root.msgStoreUrlFromUser);
+        // ensure attestors send events from the given sources to the message store
+        attestorSignerGroup.registerRouteToMessageStore(relayerUri, sources);
+        break;
+      case EVENT_RELAY:
+        EventRelayGroup eventRelayGroup = new EventRelayGroup();
+        this.messagingManagerGroup =
+            CrosschainProtocols.getMessagingInstance(CrosschainProtocols.EVENTRELAY).get();
+
+        addBcEventRelay(
+            messagingManagerGroup,
+            crossControlManagerGroup,
+            eventRelayGroup,
+            creds,
+            this.root,
+            sources);
 
-        if (fetchFromStore) {
-          // ensure attestors send events from the given sources to the message store
-          attestorSignerGroup.registerRouteToMessageStore(relayerUri, sources);
+        addBcEventRelay(
+            messagingManagerGroup,
+            crossControlManagerGroup,
+            eventRelayGroup,
+            creds,
+            this.bc2,
+            sources);
+        if (numberOfBlockchains == 3) {
+          addBcEventRelay(
+              messagingManagerGroup,
+              crossControlManagerGroup,
+              eventRelayGroup,
+              creds,
+              this.bc3,
+              sources);
         }
 
+        List<AttestorRelayer.Dest> targets = setupDispatcherTargets();
+
+        eventRelayGroup.configureRelayer(
+            globalSigner, relayerUri, sources, this.root.dispatcherUri, targets);
         break;
       case TRANSACTION_RECEIPT_SIGNING:
         TxRootRelayerGroup relayerGroup = new TxRootRelayerGroup();
@@ -168,7 +190,7 @@ public void standardExampleConfig(int numberOfBlockchains, boolean fetchFromStor
     setupCrosschainTrust(crossControlManagerGroup, messagingManagerGroup);
   }
 
-  protected abstract void loadFunctionPayerProperties(PropertiesLoader propsLoader);
+  protected abstract void loadFunctionLayerProperties(PropertiesLoader propsLoader);
 
   protected abstract CrossControlManagerGroup getFunctionCallInstance() throws Exception;
 
@@ -200,11 +222,33 @@ private void addBcAttestorSign(
       BlockchainConfig bc,
       List<AttestorRelayer.Source> sources)
       throws Exception {
+    crossControlManagerGroup.addBlockchainAndDeployContracts(creds, bc);
+    String crosschainControlAddr = crossControlManagerGroup.getCbcAddress(bc.bcId);
     messagingManagerGroup.addBlockchainAndDeployContracts(creds, bc);
     attestorSignerGroup.addBlockchain(bc.bcId, bc.msgStoreUrlFromUser);
-    crossControlManagerGroup.addBlockchainAndDeployContracts(
-        creds, bc, attestorSignerGroup.getVerifier(bc.bcId));
+    crossControlManagerGroup.setMessageVerifier(bc.bcId, attestorSignerGroup.getVerifier(bc.bcId));
+    sources.add(
+        new AttestorRelayer.Source(
+            bc.bcId,
+            crosschainControlAddr,
+            getFunctionCallImplName(),
+            bc.observerUri,
+            bc.blockchainNodeWsUri));
+  }
+
+  private void addBcEventRelay(
+      MessagingManagerGroup messagingManagerGroup,
+      CrossControlManagerGroup crossControlManagerGroup,
+      EventRelayGroup eventRelayGroup,
+      Credentials creds,
+      BlockchainConfig bc,
+      List<AttestorRelayer.Source> sources)
+      throws Exception {
+    crossControlManagerGroup.addBlockchainAndDeployContracts(creds, bc);
     String crosschainControlAddr = crossControlManagerGroup.getCbcAddress(bc.bcId);
+    messagingManagerGroup.addBlockchainAndDeployContracts(creds, bc, crosschainControlAddr);
+    eventRelayGroup.addBlockchain(bc.bcId);
+    crossControlManagerGroup.setMessageVerifier(bc.bcId, eventRelayGroup.getVerifier(bc.bcId));
     sources.add(
         new AttestorRelayer.Source(
             bc.bcId,
@@ -228,8 +272,26 @@ private void addBcTxRootSign(
         creds,
         bc,
         ((TxRootTransferManagerGroup) messagingManagerGroup).getTxRootContractAddress(bc.bcId));
-    crossControlManagerGroup.addBlockchainAndDeployContracts(
-        creds, bc, txRootTransferGroup.getVerifier(bc.bcId));
+    crossControlManagerGroup.addBlockchainAndDeployContracts(creds, bc);
+    crossControlManagerGroup.setMessageVerifier(bc.bcId, txRootTransferGroup.getVerifier(bc.bcId));
+  }
+
+  private List<AttestorRelayer.Dest> setupDispatcherTargets() throws Exception {
+    String txPKeyStr = "59a182023c8bdb02c838288635ac54809528381a3197758fd17a31b166fab85e";
+    byte[] txPKey = FormatConversion.hexStringToByteArray(txPKeyStr);
+    List<AttestorRelayer.Dest> targets = new ArrayList<>();
+    Set<BlockchainId> blockchainIds = this.messagingManagerGroup.getSupportedBlockchains();
+    for (BlockchainId sourceBcId : blockchainIds) {
+      for (BlockchainId targetBcId : blockchainIds) {
+        String targetBcVerifierAddr = this.messagingManagerGroup.getVerifierAddress(targetBcId);
+        String targetBcWsUri = this.messagingManagerGroup.getWsUri(targetBcId);
+        AttestorRelayer.Dest dest =
+            new AttestorRelayer.Dest(
+                sourceBcId, targetBcId, targetBcWsUri, txPKey, targetBcVerifierAddr);
+        targets.add(dest);
+      }
+    }
+    return targets;
   }
 
   public BlockchainConfig getRootBcInfo() {

examples/helpers/src/main/java/net/consensys/gpact/helpers/CrossBlockchainConsensusType.java

@@ -16,6 +16,7 @@
 
 public enum CrossBlockchainConsensusType {
   EVENT_SIGNING,
+  EVENT_RELAY,
   TRANSACTION_RECEIPT_SIGNING,
   COORDINATION_CHAIN_TX_RECEIPT_SIGNING
 }

examples/helpers/src/main/java/net/consensys/gpact/helpers/GpactExampleSystemManager.java

@@ -29,7 +29,7 @@ public GpactExampleSystemManager(String propertiesFileName) {
     super(propertiesFileName);
   }
 
-  protected void loadFunctionPayerProperties(PropertiesLoader propsLoader) {
+  protected void loadFunctionLayerProperties(PropertiesLoader propsLoader) {
     this.executionEngineType = propsLoader.getExecutionEnngine();
     StatsHolder.log(executionEngineType.name());
   }

examples/helpers/src/main/java/net/consensys/gpact/helpers/SfcExampleSystemManager.java

@@ -26,7 +26,7 @@ public SfcExampleSystemManager(String propertiesFileName) {
     super(propertiesFileName);
   }
 
-  protected void loadFunctionPayerProperties(PropertiesLoader propsLoader) {}
+  protected void loadFunctionLayerProperties(PropertiesLoader propsLoader) {}
 
   protected CrossControlManagerGroup getFunctionCallInstance() throws Exception {
     return CrosschainProtocols.getFunctionCallInstance(CrosschainProtocols.SFC).get();

examples/sfc/erc20bridge/java/src/intTest/java/net/consensys/gpact/examples/sfc/erc20bridge/SfcCrosschainTokenBridgeTest.java

@@ -16,4 +16,16 @@ public void directSignMultipleBlockchainMassC() throws Exception {
     String tempPropsFile = createPropertiesFile(true, false, false);
     ERC20TokenBridgeExample.main(new String[] {tempPropsFile}, true);
   }
+
+  @Test
+  public void eventRelayMultipleBlockchainMinting() throws Exception {
+    String tempPropsFile = createPropertiesFile(MessagingType.EVENT_RELAY, false, false);
+    ERC20TokenBridgeExample.main(new String[] {tempPropsFile}, false);
+  }
+
+  @Test
+  public void eventRelayMultipleBlockchainMassC() throws Exception {
+    String tempPropsFile = createPropertiesFile(MessagingType.EVENT_RELAY, false, false);
+    ERC20TokenBridgeExample.main(new String[] {tempPropsFile}, true);
+  }
 }