FAB-6244 CompletableFuture can leak in Channel

Updated with code review comments.

Change-Id: I5c741afc76d9517fd2d1d7eb57fa2b04a7d367d8
Signed-off-by: sgdev <sharadthedev@gmail.com>

Author: sgdev

src/main/java/org/hyperledger/fabric/sdk/Channel.java

@@ -2435,25 +2435,7 @@ public CompletableFuture<TransactionEvent> sendTransaction(Collection<ProposalRe
                     String emsg = format("Channel %s unsuccessful sendTransaction to orderer", name);
                     if (resp != null) {
 
-                        StringBuilder respdata = new StringBuilder(400);
-
-                        Status status = resp.getStatus();
-                        if (null != status) {
-                            respdata.append(status.name());
-                            respdata.append("-");
-                            respdata.append(status.getNumber());
-                        }
-
-                        String info = resp.getInfo();
-                        if (null != info && !info.isEmpty()) {
-                            if (respdata.length() > 0) {
-                                respdata.append(", ");
-                            }
-
-                            respdata.append("Additional information: ").append(info);
-
-                        }
-                        emsg = format("Channel %s unsuccessful sendTransaction to orderer.  %s", name, respdata.toString());
+                        emsg = format("Channel %s unsuccessful sendTransaction to orderer.  %s", name, getRespData(resp));
                     }
 
                     logger.error(emsg, e);
@@ -2466,28 +2448,12 @@ public CompletableFuture<TransactionEvent> sendTransaction(Collection<ProposalRe
                 logger.debug(format("Channel %s successful sent to Orderer transaction id: %s", name, proposalTransactionID));
                 return sret;
             } else {
-                StringBuilder respdata = new StringBuilder(400);
-                if (resp != null) {
-                    Status status = resp.getStatus();
-                    if (null != status) {
-                        respdata.append(status.name());
-                        respdata.append("-");
-                        respdata.append(status.getNumber());
-                    }
-
-                    String info = resp.getInfo();
-                    if (null != info && !info.isEmpty()) {
-                        if (respdata.length() > 0) {
-                            respdata.append(", ");
-                        }
 
-                        respdata.append("Additional information: ").append(info);
+                String emsg = format("Channel %s failed to place transaction %s on Orderer. Cause: UNSUCCESSFUL. %s",
+                        name, proposalTransactionID, getRespData(resp));
 
-                    }
+                unregisterTxListener(proposalTransactionID);
 
-                }
-                String emsg = format("Channel %s failed to place transaction %s on Orderer. Cause: UNSUCCESSFUL. %s",
-                        name, proposalTransactionID, respdata.toString());
                 CompletableFuture<TransactionEvent> ret = new CompletableFuture<>();
                 ret.completeExceptionally(new Exception(emsg));
                 return ret;
@@ -2502,6 +2468,38 @@ public CompletableFuture<TransactionEvent> sendTransaction(Collection<ProposalRe
 
     }
 
+    /**
+     * Build response details
+     * @param resp
+     * @return
+     */
+    private String getRespData(BroadcastResponse resp) {
+
+        StringBuilder respdata = new StringBuilder(400);
+        if (resp != null) {
+            Status status = resp.getStatus();
+            if (null != status) {
+                respdata.append(status.name());
+                respdata.append("-");
+                respdata.append(status.getNumber());
+            }
+
+            String info = resp.getInfo();
+            if (null != info && !info.isEmpty()) {
+                if (respdata.length() > 0) {
+                    respdata.append(", ");
+                }
+
+                respdata.append("Additional information: ").append(info);
+
+            }
+
+        }
+
+        return respdata.toString();
+
+    }
+
     private Envelope createTransactionEnvelope(Payload transactionPayload, User user) throws CryptoException {
 
         return Envelope.newBuilder()
@@ -2892,6 +2890,20 @@ private CompletableFuture<TransactionEvent> registerTxListener(String txid) {
 
     }
 
+    /**
+     * Unregister a transactionId
+     *
+     * @param txid
+     */
+    private void unregisterTxListener(String txid) {
+
+        synchronized (txListeners) {
+
+            txListeners.remove(txid);
+        }
+
+    }
+
     ////////////////////////////////////////////////////////////////////////
     ////////////////  Chaincode Events..  //////////////////////////////////