[FEAT] N2N TxSubmission

core/src/main/java/com/bloxbean/cardano/yaci/core/network/NodeClient.java

@@ -14,9 +14,6 @@
 import lombok.extern.slf4j.Slf4j;
 
 import java.net.SocketAddress;
-import java.util.Arrays;
-import java.util.List;
-import java.util.stream.Collectors;
 
 @Slf4j
 public abstract class NodeClient {
@@ -69,10 +66,7 @@ public void start() {
 
             configureChannel(b);
 
-            List<Agent> allAgents = Arrays.stream(agents).collect(Collectors.toList());
-            allAgents.add(0, handshakeAgent);
-
-            b.handler(new ChannelInitializer<Channel>() {
+            b.handler(new ChannelInitializer<>() {
 
                 @Override
                 public void initChannel(Channel ch)
@@ -84,8 +78,7 @@ public void initChannel(Channel ch)
                 }
             });
 
-            SocketAddress socketAddress = null;
-            socketAddress = createSocketAddress();
+            SocketAddress socketAddress = createSocketAddress();
 
             session = new Session(socketAddress, b, handshakeAgent, agents);
             session.setSessionListener(sessionListener);

core/src/main/java/com/bloxbean/cardano/yaci/core/protocol/txsubmission/TxSubmissionAgent.java

@@ -1,26 +1,33 @@
 package com.bloxbean.cardano.yaci.core.protocol.txsubmission;
 
+import com.bloxbean.cardano.yaci.core.common.TxBodyType;
 import com.bloxbean.cardano.yaci.core.protocol.Agent;
 import com.bloxbean.cardano.yaci.core.protocol.Message;
 import com.bloxbean.cardano.yaci.core.protocol.txsubmission.messges.*;
+import com.bloxbean.cardano.yaci.core.protocol.txsubmission.model.TxSubmissionRequest;
 import lombok.extern.slf4j.Slf4j;
 
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.ConcurrentLinkedQueue;
 
 @Slf4j
-public class TxSubmissionAgent extends Agent {
-    private final Map<String, byte[]> txs;
-    private final List<String> reqTxIds;
-    private final List<String> reqNonBlockingTxIds;
+public class TxSubmissionAgent extends Agent<TxSubmissionListener> {
+    // txs should be stored in a thread-safe, ordered (tx dependency/chaining) data structure.
+    private final ConcurrentLinkedQueue<TxSubmissionRequest> txs;
+    /**
+     * Is the queue of TX received from client
+     */
+    private final ConcurrentLinkedQueue<String> pendingTxIds;
+    /**
+     * It's the temporary list of TX ids requested from Server
+     */
+    private final ConcurrentLinkedQueue<String> requestedTxIds;
 
     public TxSubmissionAgent() {
         this.currenState = TxSubmissionState.Init;
-        txs = new HashMap<>();
-        reqTxIds = new ArrayList<>();
-        reqNonBlockingTxIds = new ArrayList<>();
+        this.txs = new ConcurrentLinkedQueue<>();
+        this.pendingTxIds = new ConcurrentLinkedQueue<>();
+        this.requestedTxIds = new ConcurrentLinkedQueue<>();
     }
 
     @Override
@@ -34,8 +41,7 @@ public Message buildNextMessage() {
             case Init:
                 return new Init();
             case TxIdsNonBlocking:
-                return getReplyTxIds();
-            case  TxIdsBlocking:
+            case TxIdsBlocking:
                 return getReplyTxIds();
             case Txs:
                 return getReplyTxs();
@@ -44,54 +50,139 @@ public Message buildNextMessage() {
         }
     }
 
+    private Optional<TxSubmissionRequest> findTxIdAndHash(String id) {
+        return txs.stream().filter(txSubmissionRequest -> txSubmissionRequest.getTxHash().equals(id)).findAny();
+    }
+
+    private Optional<TxSubmissionRequest> removeTxIdAndHash(String id) {
+        var txIdAndHashOpt = txs.stream().filter(txSubmissionRequest -> txSubmissionRequest.getTxHash().equals(id)).findAny();
+        txIdAndHashOpt.ifPresent(txs::remove);
+        return txIdAndHashOpt;
+    }
+
     private ReplyTxIds getReplyTxIds() {
-        if (txs != null && txs.size() > 0) {
+        if (!pendingTxIds.isEmpty()) {
             ReplyTxIds replyTxIds = new ReplyTxIds();
-            txs.forEach((id, txBytes) -> {
-                replyTxIds.addTxId(id, txBytes.length);
-            });
+            // Not limiting how many txs to add, as pendingTxIds should be already capped to num of req txs
+            pendingTxIds
+                    .stream()
+                    .flatMap(id -> findTxIdAndHash(id).stream())
+                    .forEach(txSubmissionRequest -> replyTxIds.addTxId(txSubmissionRequest.getTxHash(), txSubmissionRequest.getTxnBytes().length));
+            if (log.isDebugEnabled())
+                log.debug("TxIds: {}", replyTxIds.getTxIdAndSizeMap().size());
             return replyTxIds;
-        } else
-            return new ReplyTxIds();
+        }
+        return new ReplyTxIds();
     }
 
     private ReplyTxs getReplyTxs() {
-        if (reqTxIds.isEmpty())
+        if (requestedTxIds.isEmpty())
             return new ReplyTxs();
 
         ReplyTxs replyTxs = new ReplyTxs();
-        for (String txId: reqTxIds) {
-            byte[] tx = txs.get(txId);
-            replyTxs.addTx(tx);
-        }
+        requestedTxIds.forEach(txId -> removeTxIdAndHash(txId).ifPresent(txSubmissionRequest -> replyTxs.addTx(txSubmissionRequest.getTxnBytes())));
 
+        // Ids of requested TXs don't seem to be acked from server.
+        // Removing them right away now.
+        requestedTxIds.forEach(pendingTxIds::remove);
+        if (log.isDebugEnabled())
+            log.debug("Txs: {}", replyTxs.getTxns().size());
         return replyTxs;
     }
 
     @Override
     public void processResponse(Message message) {
         if (message == null) return;
-        if (message instanceof RequestTxIds) {
-            if (((RequestTxIds) message).isBlocking()) {
-                handleRequestTxIdsBlocking((RequestTxIds) message);
+
+        if (message instanceof Init) {
+            log.warn("init");
+        } else if (message instanceof RequestTxIds) {
+            var requestTxIds = (RequestTxIds) message;
+            if (requestTxIds.isBlocking()) {
+                if (log.isDebugEnabled())
+                    log.debug("RequestTxIds - Blocking, ack: {}, req: {}", requestTxIds.getAckTxIds(), requestTxIds.getReqTxIds());
+                handleRequestTxIdsBlocking(requestTxIds);
             } else {
-                handleRequestTxIdsNonBlocking((RequestTxIds) message);
+                if (log.isDebugEnabled())
+                    log.debug("RequestTxIds - NonBlocking, ack: {}, req: {}", requestTxIds.getAckTxIds(), requestTxIds.getReqTxIds());
+                handleRequestTxIdsNonBlocking(requestTxIds);
             }
         } else if (message instanceof RequestTxs) {
             handleRequestTxs((RequestTxs) message);
         }
     }
 
     private void handleRequestTxs(RequestTxs requestTxs) {
-        log.info("RequestTxs >>" + requestTxs);
+        requestedTxIds.clear();
+        requestedTxIds.addAll(requestTxs.getTxIds());
+        getAgentListeners().forEach(listener -> listener.handleRequestTxs(requestTxs));
     }
 
     private void handleRequestTxIdsNonBlocking(RequestTxIds requestTxIds) {
-        log.info("RequestTxIdsNonBlocking >> " + requestTxIds);
+        // process ack
+        removeAcknowledgedTxs(requestTxIds.getAckTxIds());
+        addTxToQueue(requestTxIds.getReqTxIds());
+        getAgentListeners().forEach(listener -> listener.handleRequestTxIdsNonBlocking(requestTxIds));
     }
 
     private void handleRequestTxIdsBlocking(RequestTxIds requestTxIds) {
-        log.info("RequestTxIdsBlocking >> " + requestTxIds);
+        // process ack
+        removeAcknowledgedTxs(requestTxIds.getAckTxIds());
+        addTxToQueue(requestTxIds.getReqTxIds());
+        getAgentListeners().forEach(listener -> listener.handleRequestTxIdsBlocking(requestTxIds));
+    }
+
+    private void addTxToQueue(int numTxToAdd) {
+        // pendingTxIds size can't exceed numTxToAdd
+        var txToAdd = numTxToAdd - pendingTxIds.size();
+        if (!txs.isEmpty()) {
+            txs.stream()
+                    .map(TxSubmissionRequest::getTxHash)
+                    .filter(txHash -> !pendingTxIds.contains(txHash))
+                    .limit(txToAdd)
+                    .forEach(pendingTxIds::add);
+        } else {
+            if (log.isDebugEnabled())
+                log.debug("Nothing to do, txs is empty");
+        }
+    }
+
+    private void addTxToQueue(String txHash) {
+        if (!pendingTxIds.contains(txHash)) {
+            pendingTxIds.add(txHash);
+        }
+    }
+
+    private void removeAcknowledgedTxs(int numAcknowledgedTransactions) {
+        if (numAcknowledgedTransactions > 0) {
+            var numTxToRemove = Math.min(numAcknowledgedTransactions, pendingTxIds.size());
+            for (int i = 0; i < numTxToRemove; i++) {
+                var txHash = pendingTxIds.poll();
+                if (txHash != null) {
+                    // remove from map
+                    removeTxIdAndHash(txHash);
+                    // removed from queue
+                    pendingTxIds.remove(txHash);
+                }
+            }
+
+        }
+
+    }
+
+    public void enqueueTransaction(String txHash, byte[] txBytes, TxBodyType txBodyType) {
+        if (txs.stream().anyMatch(txSubmissionRequest -> txSubmissionRequest.getTxHash().equals(txHash))) {
+            return;
+        }
+        txs.add(TxSubmissionRequest.builder().txHash(txHash).txnBytes(txBytes).txBodyType(txBodyType).build());
+        if (TxSubmissionState.TxIdsBlocking.equals(currenState)) {
+            addTxToQueue(txHash);
+            this.sendNextMessage();
+        }
+    }
+
+    public boolean hasPendingTx() {
+        return !pendingTxIds.isEmpty();
     }
 
     @Override
@@ -101,6 +192,9 @@ public boolean isDone() {
 
     @Override
     public void reset() {
+        txs.clear();
+        pendingTxIds.clear();
+        requestedTxIds.clear();
         this.currenState = TxSubmissionState.Init;
     }
 }

core/src/main/java/com/bloxbean/cardano/yaci/core/protocol/txsubmission/TxSubmissionListener.java

@@ -0,0 +1,13 @@
+package com.bloxbean.cardano.yaci.core.protocol.txsubmission;
+
+import com.bloxbean.cardano.yaci.core.protocol.AgentListener;
+import com.bloxbean.cardano.yaci.core.protocol.txsubmission.messges.RequestTxIds;
+import com.bloxbean.cardano.yaci.core.protocol.txsubmission.messges.RequestTxs;
+
+public interface TxSubmissionListener extends AgentListener {
+     void handleRequestTxs(RequestTxs requestTxs);
+
+     void handleRequestTxIdsNonBlocking(RequestTxIds requestTxIds);
+
+     void handleRequestTxIdsBlocking(RequestTxIds requestTxIds);
+}

core/src/main/java/com/bloxbean/cardano/yaci/core/protocol/txsubmission/messges/ReplyTxIds.java

@@ -1,22 +1,26 @@
 package com.bloxbean.cardano.yaci.core.protocol.txsubmission.messges;
 
 import com.bloxbean.cardano.yaci.core.protocol.Message;
+import com.bloxbean.cardano.yaci.core.protocol.localstate.api.Era;
 import com.bloxbean.cardano.yaci.core.protocol.txsubmission.serializers.TxSubmissionMessagesSerializers;
-import lombok.AllArgsConstructor;
-import lombok.Builder;
 import lombok.Getter;
-import lombok.NoArgsConstructor;
+import lombok.RequiredArgsConstructor;
 
 import java.util.HashMap;
 import java.util.Map;
 
 @Getter
-@AllArgsConstructor
-@NoArgsConstructor
-@Builder
+@RequiredArgsConstructor
 public class ReplyTxIds implements Message {
+
+    private final Era era;
+
     private Map<String, Integer> txIdAndSizeMap;
 
+    public ReplyTxIds() {
+        this(Era.Babbage);
+    }
+
     public void addTxId(String id, int size) {
         if (txIdAndSizeMap == null)
             txIdAndSizeMap = new HashMap<>();

core/src/main/java/com/bloxbean/cardano/yaci/core/protocol/txsubmission/messges/ReplyTxs.java

@@ -1,22 +1,26 @@
 package com.bloxbean.cardano.yaci.core.protocol.txsubmission.messges;
 
 import com.bloxbean.cardano.yaci.core.protocol.Message;
+import com.bloxbean.cardano.yaci.core.protocol.localstate.api.Era;
 import com.bloxbean.cardano.yaci.core.protocol.txsubmission.serializers.TxSubmissionMessagesSerializers;
-import lombok.AllArgsConstructor;
-import lombok.Builder;
 import lombok.Getter;
-import lombok.NoArgsConstructor;
+import lombok.RequiredArgsConstructor;
 
 import java.util.ArrayList;
 import java.util.List;
 
 @Getter
-@AllArgsConstructor
-@NoArgsConstructor
-@Builder
+@RequiredArgsConstructor
 public class ReplyTxs implements Message {
+
+    private final Era era;
+
     private List<byte[]> txns;
 
+    public ReplyTxs() {
+        this(Era.Babbage);
+    }
+
     public void addTx(byte[] tx) {
         if (txns == null)
             txns = new ArrayList<>();

core/src/main/java/com/bloxbean/cardano/yaci/core/protocol/txsubmission/model/TxSubmissionRequest.java

@@ -0,0 +1,40 @@
+package com.bloxbean.cardano.yaci.core.protocol.txsubmission.model;
+
+import com.bloxbean.cardano.client.transaction.util.TransactionUtil;
+import com.bloxbean.cardano.yaci.core.common.TxBodyType;
+import com.bloxbean.cardano.yaci.core.util.HexUtil;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Getter;
+
+@AllArgsConstructor
+@Getter
+@Builder
+public class TxSubmissionRequest {
+
+    private final TxBodyType txBodyType;
+    private final byte[] txnBytes;
+    private final String txHash;
+
+    public TxSubmissionRequest(byte[] txnBytes) {
+       this(TxBodyType.BABBAGE, txnBytes);
+    }
+
+    public TxSubmissionRequest(TxBodyType txBodyType, byte[] txnBytes) {
+        if (txnBytes == null)
+            throw new RuntimeException("TxBytes can't be null");
+
+        this.txBodyType = txBodyType;
+        this.txnBytes = txnBytes;
+        this.txHash = TransactionUtil.getTxHash(txnBytes);
+    }
+
+    @Override
+    public String toString() {
+        return "TxSubmissionRequest{" +
+                "txBodyType=" + txBodyType +
+                ", txnBytes=" + (txnBytes != null? HexUtil.encodeHexString(txnBytes) : "") +
+                ", txnHash=" + txHash +
+                '}';
+    }
+}