FABJ-383 Garbage collection orderer peer connection.

PS
06 Minor logging updates.
07 Logging changes.

Change-Id: I1e6ed709b43b987ad573da15af63a36b492c8850
Signed-off-by: rickr <cr22rc@gmail.com>

Author: cr22rc

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

@@ -17,7 +17,6 @@
 import java.util.concurrent.TimeUnit;
 
 import com.google.common.util.concurrent.ListenableFuture;
-import io.grpc.ConnectivityState;
 import io.grpc.ManagedChannel;
 import io.grpc.ManagedChannelBuilder;
 import org.apache.commons.logging.Log;
@@ -28,30 +27,50 @@
 import org.hyperledger.fabric.protos.peer.FabricProposal;
 import org.hyperledger.fabric.protos.peer.FabricProposalResponse;
 import org.hyperledger.fabric.sdk.exception.PeerException;
+import org.hyperledger.fabric.sdk.helper.Config;
+
+import static java.lang.String.format;
 
 /**
  * Sample client code that makes gRPC calls to the server.
  */
 class EndorserClient {
+    private static final Config config = Config.getConfig();
     private static final Log logger = LogFactory.getLog(EndorserClient.class);
+    private static final boolean IS_TRACE_LEVEL = logger.isTraceEnabled();
 
+    //    private final String channelName;
+//    private final String name;
+//    private final String url;
     private ManagedChannel managedChannel;
     private EndorserGrpc.EndorserFutureStub futureStub;
-    DiscoveryGrpc.DiscoveryFutureStub discoveryFutureStub;
+    private DiscoveryGrpc.DiscoveryFutureStub discoveryFutureStub;
     private boolean shutdown = false;
+    private final String toString;
 
     /**
      * Construct client for accessing Peer server using the existing channel.
      *
      * @param channelBuilder The ChannelBuilder to build the endorser client
      */
-    EndorserClient(ManagedChannelBuilder<?> channelBuilder) {
+    EndorserClient(String channelName, String name, String url, ManagedChannelBuilder<?> channelBuilder) {
         managedChannel = channelBuilder.build();
         futureStub = EndorserGrpc.newFutureStub(managedChannel);
         discoveryFutureStub = DiscoveryGrpc.newFutureStub(managedChannel);
+        toString = "EndorserClient{" + "id: " + config.getNextID() + ", channel: " + channelName + ", name:" + name + ", url: " + url + "}";
+        logger.trace("Created " + toString());
+
+    }
+
+    @Override
+    public String toString() {
+        return toString;
     }
 
     synchronized void shutdown(boolean force) {
+        if (IS_TRACE_LEVEL) {
+            logger.trace(format("%s shutdown called force: %b, shutdown: %b, managedChannel: %s", toString(), force, shutdown, "" + managedChannel));
+        }
         if (shutdown) {
             return;
         }
@@ -60,49 +79,76 @@ synchronized void shutdown(boolean force) {
         // let all referenced resource finalize
         managedChannel = null;
         discoveryFutureStub = null;
-
         futureStub = null;
 
         if (lchannel == null) {
             return;
         }
         if (force) {
-            lchannel.shutdownNow();
+
+            try {
+                lchannel.shutdownNow();
+            } catch (Exception e) {
+                logger.warn(e);
+            }
+
         } else {
             boolean isTerminated = false;
 
             try {
                 isTerminated = lchannel.shutdown().awaitTermination(3, TimeUnit.SECONDS);
             } catch (Exception e) {
-                logger.debug(e); //best effort
+                logger.debug(toString(), e); //best effort
             }
             if (!isTerminated) {
-                lchannel.shutdownNow();
+                try {
+                    lchannel.shutdownNow();
+                } catch (Exception e) {
+                    logger.warn(toString(), e);
+                }
             }
         }
     }
 
     public ListenableFuture<FabricProposalResponse.ProposalResponse> sendProposalAsync(FabricProposal.SignedProposal proposal) throws PeerException {
         if (shutdown) {
-            throw new PeerException("Shutdown");
+            throw new PeerException("Shutdown " + toString());
         }
         return futureStub.processProposal(proposal);
     }
 
     public ListenableFuture<Protocol.Response> sendDiscoveryRequestAsync(Protocol.SignedRequest signedRequest) throws PeerException {
         if (shutdown) {
-            throw new PeerException("Shutdown");
+            throw new PeerException("Shutdown " + toString());
         }
         return discoveryFutureStub.discover(signedRequest);
     }
 
     boolean isChannelActive() {
         ManagedChannel lchannel = managedChannel;
-        return lchannel != null && !lchannel.isShutdown() && !lchannel.isTerminated() && ConnectivityState.READY.equals(lchannel.getState(true));
+        if (null == lchannel) {
+
+            logger.trace(toString() + "Grpc channel needs creation.");
+
+            return false;
+        }
+
+        final boolean isTerminated = lchannel.isTerminated();
+        final boolean isShutdown = lchannel.isShutdown();
+        final boolean ret = !lchannel.isShutdown() && !isTerminated; // && ConnectivityState.READY.equals(lchannel.getState(true));
+        if (IS_TRACE_LEVEL) {
+            logger.trace(format("%s grpc channel isActive: %b, isShutdown: %b, isTerminated: %b, state: %s ", toString(), ret, isShutdown, isTerminated, "" + lchannel.getState(false)));
+        }
+
+        return ret;
     }
 
     @Override
     public void finalize() {
+        if (!shutdown) {
+            logger.warn(toString() + " finalized not shutdown is Active" + isChannelActive());
+        }
+
         shutdown(true);
     }
 }

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

@@ -14,6 +14,8 @@
 
 package org.hyperledger.fabric.sdk;
 
+import java.io.IOException;
+import java.io.ObjectInputStream;
 import java.io.Serializable;
 import java.util.ArrayList;
 import java.util.Properties;
@@ -52,8 +54,11 @@
 
 public class EventHub implements Serializable {
     private static final long serialVersionUID = 2882609588201108148L;
-    private static final Log logger = LogFactory.getLog(EventHub.class);
     private static final Config config = Config.getConfig();
+    private transient String id = config.getNextID();
+    private static final Log logger = LogFactory.getLog(EventHub.class);
+    private static final boolean IS_TRACE_LEVEL = logger.isTraceEnabled();
+
     private static final long EVENTHUB_CONNECTION_WAIT_TIME = config.getEventHubConnectionWaitTime();
     private static final long EVENTHUB_RECONNECTION_WARNING_RATE = config.getEventHubReconnectionWarningRate();
 
@@ -78,6 +83,7 @@ public class EventHub implements Serializable {
     private transient long reconnectCount;
     private transient long lastBlockNumber;
     private transient BlockEvent lastBlockEvent;
+    private String channelName;
 
     /**
      * Get disconnected time.
@@ -99,6 +105,22 @@ public boolean isConnected() {
         return connected;
     }
 
+    String getStatus() {
+
+        StringBuilder sb = new StringBuilder(1000);
+        sb.append(toString()).append(", connected: ").append(connected);
+        ManagedChannel lmanagedChannel = managedChannel;
+        if (lmanagedChannel == null) {
+            sb.append("managedChannel: null");
+        } else {
+            sb.append(", isShutdown: ").append(lmanagedChannel.isShutdown());
+            sb.append(", isTerminated: ").append(lmanagedChannel.isTerminated());
+            sb.append(", state: ").append("" + lmanagedChannel.getState(false));
+        }
+
+        return sb.toString();
+    }
+
     /**
      * Get last connect time.
      *
@@ -136,6 +158,7 @@ public long getLastConnectedAttempt() {
         this.name = name;
         this.executorService = executorService;
         this.properties = properties == null ? null : (Properties) properties.clone(); //keep our own copy.
+        logger.debug("Created " + toString());
     }
 
     /**
@@ -186,7 +209,7 @@ synchronized boolean connect(final TransactionContext transactionContext, final
 
         final CountDownLatch finishLatch = new CountDownLatch(1);
 
-        logger.debug(format("EventHub %s is connecting.", name));
+        logger.debug(format("%s is connecting.", toString()));
 
         lastConnectedAttempt = System.currentTimeMillis();
 
@@ -203,12 +226,14 @@ synchronized boolean connect(final TransactionContext transactionContext, final
             @Override
             public void onNext(PeerEvents.Event event) {
 
-                logger.debug(format("EventHub %s got  event type: %s", EventHub.this.name, event.getEventCase().name()));
+                logger.debug(format("%s got  event type: %s", EventHub.this.toString(), event.getEventCase().name()));
 
                 if (event.getEventCase() == PeerEvents.Event.EventCase.BLOCK) {
                     try {
 
                         BlockEvent blockEvent = new BlockEvent(EventHub.this, event);
+
+                        logger.trace(format("%s got block number: %d", EventHub.this.toString(), blockEvent.getBlockNumber()));
                         setLastBlockSeen(blockEvent);
 
                         eventQue.addBEvent(blockEvent);  //add to channel queue
@@ -220,14 +245,17 @@ public void onNext(PeerEvents.Event event) {
                 } else if (event.getEventCase() == PeerEvents.Event.EventCase.REGISTER) {
 
                     if (reconnectCount > 1) {
-                        logger.info(format("Eventhub %s has reconnecting after %d attempts", name, reconnectCount));
+                        logger.info(format("%s has reconnecting after %d attempts", EventHub.this.toString(), reconnectCount));
                     }
 
                     connected = true;
                     connectedTime = System.currentTimeMillis();
                     reconnectCount = 0L;
 
                     finishLatch.countDown();
+                } else {
+                    logger.error(format("%s got a unexpected block type: %s",
+                            EventHub.this.toString(), event.getEventCase().name()));
                 }
             }
 
@@ -271,7 +299,7 @@ public void onError(Throwable t) {
                     try {
                         reconnect();
                     } catch (Exception e) {
-                        logger.warn(format("Eventhub %s Failed shutdown msg:  %s", EventHub.this.name, e.getMessage()));
+                        logger.warn(format("%s Failed shutdown msg:  %s", EventHub.this.toString(), e.getMessage()));
                     }
 
                 }
@@ -300,17 +328,17 @@ public void onCompleted() {
 
             if (!reconnection && !finishLatch.await(EVENTHUB_CONNECTION_WAIT_TIME, TimeUnit.MILLISECONDS)) {
 
-                logger.warn(format("EventHub %s failed to connect in %s ms.", name, EVENTHUB_CONNECTION_WAIT_TIME));
+                logger.warn(format("%s failed to connect in %s ms.", toString(), EVENTHUB_CONNECTION_WAIT_TIME));
 
             } else {
-                logger.trace(format("Eventhub %s Done waiting for reply!", name));
+                logger.trace(format("%s done waiting for reply!", toString()));
             }
 
         } catch (InterruptedException e) {
             logger.error(e);
         }
 
-        logger.debug(format("Eventhub %s connect is done with connect status: %b ", name, connected));
+        logger.debug(format("%s connect is done with connect status: %b ", toString(), connected));
 
         if (connected) {
             eventStream = eventStreamLocal;
@@ -383,16 +411,19 @@ void setEventQue(Channel.ChannelEventQue eventQue) {
 
     @Override
     public String toString() {
-        return "EventHub:" + getName();
+        return "EventHub{" + "id: " + id + ", name: " + getName() + ", channelName: " + channelName + ", url: " + getUrl() + "}";
     }
 
-    public void shutdown() {
+    public synchronized void shutdown() {
+        if (shutdown) {
+            return;
+        }
+        logger.trace(toString() + " being shutdown.");
         shutdown = true;
         lastBlockEvent = null;
         lastBlockNumber = 0;
         connected = false;
         disconnectedHandler = null;
-        channel = null;
         eventStream = null;
         final ManagedChannel lmanagedChannel = managedChannel;
         managedChannel = null;
@@ -406,20 +437,23 @@ void setChannel(Channel channel) throws InvalidArgumentException {
             throw new InvalidArgumentException("setChannel Channel can not be null");
         }
 
-        if (null != this.channel) {
+        if (null != channelName) {
             throw new InvalidArgumentException(format("Can not add event hub  %s to channel %s because it already belongs to channel %s.",
-                    name, channel.getName(), this.channel.getName()));
+                    name, channel.getName(), channelName));
         }
-
-        this.channel = channel;
+        logger.debug(toString() + " set to channel: " + channel);
+        this.channelName = channel.getName();
     }
 
-    synchronized void setLastBlockSeen(BlockEvent lastBlockSeen) {
+    private synchronized void setLastBlockSeen(BlockEvent lastBlockSeen) {
         long newLastBlockNumber = lastBlockSeen.getBlockNumber();
         // overkill but make sure.
         if (lastBlockNumber < newLastBlockNumber) {
             lastBlockNumber = newLastBlockNumber;
             this.lastBlockEvent = lastBlockSeen;
+            if (IS_TRACE_LEVEL) {
+                logger.trace(toString() + " last block seen: " + lastBlockNumber);
+            }
         }
     }
 
@@ -446,7 +480,7 @@ public interface EventHubDisconnected {
         @Override
         public synchronized void disconnected(final EventHub eventHub) {
             if (reconnectCount == 1) {
-                logger.warn(format("Channel %s detected disconnect on event hub %s (%s)", channel.getName(), eventHub.toString(), url));
+                logger.warn(format("%s detected disconnect.", eventHub.toString()));
             }
 
             executorService.execute(() -> {
@@ -455,7 +489,7 @@ public synchronized void disconnected(final EventHub eventHub) {
                     Thread.sleep(500);
 
                     if (transactionContext == null) {
-                        logger.warn("Eventhub reconnect failed with no user context");
+                        logger.warn(EventHub.this.toString() + " reconnect failed with no user context");
                         return;
                     }
 
@@ -485,4 +519,16 @@ public EventHubDisconnected setEventHubDisconnectedHandler(EventHubDisconnected
         return ret;
     }
 
+    private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException {
+        in.defaultReadObject();
+        id = config.getNextID();
+    }
+
+    public void finalize() throws Throwable {
+        logger.trace(format("%s finalized", toString()));
+        shutdown();
+        super.finalize();
+
+    }
+
 }

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

@@ -51,7 +51,7 @@
 import static org.hyperledger.fabric.sdk.User.userContextCheck;
 
 public class HFClient {
-    private static final Config config = Config.getConfig();
+    private static final Config config = Config.getConfig(); // never remove this! config needs to load first.
 
     private CryptoSuite cryptoSuite;
     protected final ExecutorService executorService;
@@ -66,7 +66,6 @@ public class HFClient {
         }
     }
 
-
     ExecutorService getExecutorService() {
         return executorService;
     }