FAB-6200 Java serialize channels.

PS2 Check for shutdown channel.
PS3 Minor cleanup no need for thread local.
PS4 Remove unnecessary added code.
PS5 Remove added commented code.
PS6 Minor testcode change.
PS7 Refactor cleaner to serialize on client itself.
PS8 Refactor samplestore to use above.
PS9 Typo fix.
PS10 ""
PS11 make sure stream closed.
PS12 Make sure client is updated with channel.
PS13 Move checks on both channels.
PS14 sync on channels
PS15 Do query peer channels on both again.

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

Author: cr22rc

README.md

@@ -9,6 +9,8 @@ The SDK acts on behave of a particular User which is defined by the embedding ap
 
 Note, the SDK does ***not*** provide a means of persistence
   for the application defined channels and user artifacts on the client. This is left for the embedding application to best manage.
+  Channels may be serialized via Java serialization in the context of a client.
+  Channels deserialized are not in an initialized state.
 
 The SDK also provides a client for Hyperledger's certificate authority.  The SDK is however not dependent on this
 particular implementation of a certificate authority. Other Certificate authority's maybe used by implementing the

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

@@ -14,6 +14,15 @@
 
 package org.hyperledger.fabric.sdk;
 
+import java.io.ByteArrayOutputStream;
+import java.io.File;
+import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
+import java.io.Serializable;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.nio.file.StandardOpenOption;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
@@ -30,7 +39,6 @@
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
-import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Future;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.TimeUnit;
@@ -112,7 +120,8 @@
  * The class representing a channel with which the client SDK interacts.
  * <p>
  */
-public class Channel {
+public class Channel implements Serializable {
+    private static final long serialVersionUID = -3266164166893832538L;
     private static final Log logger = LogFactory.getLog(Channel.class);
     private static final boolean IS_DEBUG_LEVEL = logger.isDebugEnabled();
     private static final boolean IS_TRACE_LEVEL = logger.isTraceEnabled();
@@ -129,7 +138,7 @@ public class Channel {
     private final String name;
 
     // The peers on this channel to which the client can connect
-    private final Collection<Peer> peers = new Vector<>();
+    final Collection<Peer> peers = new Vector<>();
 
     // Temporary variables to control how long to wait for deploy and invoke to complete before
     // emitting events.  This will be removed when the SDK is able to receive events from the
@@ -141,10 +150,26 @@ public class Channel {
 
     // The crypto primitives object
     //   private CryptoSuite cryptoSuite;
-    private final Collection<Orderer> orderers = new LinkedList<>();
-    HFClient client;
-    private boolean initialized = false;
-    private boolean shutdown = false;
+    final Collection<Orderer> orderers = new LinkedList<>();
+    transient HFClient client;
+    private transient boolean initialized = false;
+    private transient boolean shutdown = false;
+
+    private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException {
+
+        in.defaultReadObject();
+        initialized = false;
+        shutdown = false;
+        msps = new HashMap<>();
+        txListeners = new LinkedHashMap<>();
+        channelEventQue = new ChannelEventQue();
+        blockListeners = new LinkedHashMap<>();
+
+        for (EventHub eventHub : getEventHubs()) {
+            eventHub.setEventQue(channelEventQue);
+        }
+
+    }
 
     /**
      * Get all Event Hubs on this channel.
@@ -155,9 +180,8 @@ public Collection<EventHub> getEventHubs() {
         return Collections.unmodifiableCollection(eventHubs);
     }
 
-    private final Collection<EventHub> eventHubs = new LinkedList<>();
-    private ExecutorService executorService;
-    private Block genesisBlock;
+    final Collection<EventHub> eventHubs = new LinkedList<>();
+    private transient Block genesisBlock;
     private final boolean systemChannel;
 
     private Channel(String name, HFClient hfClient, Orderer orderer, ChannelConfiguration channelConfiguration, byte[][] signers) throws InvalidArgumentException, TransactionException {
@@ -461,8 +485,6 @@ private Channel(String name, HFClient client, final boolean systemChannel) throw
         }
         this.name = name;
         this.client = client;
-        this.executorService = client.getExecutorService();
-
         logger.debug(format("Creating channel: %s, client context %s", isSystemChannel() ? "SYSTEM_CHANNEL" : name, client.getUserContext().getName()));
 
     }
@@ -731,10 +753,10 @@ public Channel initialize() throws InvalidArgumentException, TransactionExceptio
      * @throws InvalidArgumentException
      * @throws CryptoException
      */
-    private void loadCACertificates() throws InvalidArgumentException, CryptoException {
+    protected void loadCACertificates() throws InvalidArgumentException, CryptoException {
         logger.debug(format("Channel %s loadCACertificates", name));
 
-        if (msps == null) {
+        if (msps == null || msps.isEmpty()) {
             throw new InvalidArgumentException("Unable to load CA certificates. Channel " + name + " does not have any MSPs.");
         }
 
@@ -822,7 +844,7 @@ private Block getGenesisBlock(Orderer orderer) throws TransactionException {
         return genesisBlock;
     }
 
-    private Map<String, MSP> msps = new HashMap<>();
+    private transient Map<String, MSP> msps = new HashMap<>();
 
     boolean isSystemChannel() {
         return systemChannel;
@@ -2585,7 +2607,7 @@ public boolean unRegisterBlockListener(String handle) throws InvalidArgumentExce
      * A queue each eventing hub will write events to.
      */
 
-    private final ChannelEventQue channelEventQue = new ChannelEventQue();
+    private transient ChannelEventQue channelEventQue = new ChannelEventQue();
 
     class ChannelEventQue {
 
@@ -2659,15 +2681,15 @@ BlockEvent getNextEvent() throws EventHubException {
      * Runs processing events from event hubs.
      */
 
-    Thread eventQueueThread = null;
+    transient Thread eventQueueThread = null;
 
     private void startEventQue() {
 
         if (eventQueueThread != null) {
             return;
         }
 
-        executorService.execute(() -> {
+        client.getExecutorService().execute(() -> {
             eventQueueThread = Thread.currentThread();
 
             while (!shutdown) {
@@ -2700,7 +2722,7 @@ private void startEventQue() {
 
                     for (BL l : blcopy) {
                         try {
-                            executorService.execute(() -> l.listener.received(blockEvent));
+                            client.getExecutorService().execute(() -> l.listener.received(blockEvent));
                         } catch (Throwable e) { //Don't let one register stop rest.
                             logger.error("Error trying to call block listener on channel " + blockEvent.getChannelId(), e);
                         }
@@ -2717,7 +2739,7 @@ private void startEventQue() {
 
     private static final String BLOCK_LISTENER_TAG = "BLOCK_LISTENER_HANDLE";
 
-    private final LinkedHashMap<String, BL> blockListeners = new LinkedHashMap<>();
+    private transient LinkedHashMap<String, BL> blockListeners = new LinkedHashMap<>();
 
     class BL {
 
@@ -2791,7 +2813,7 @@ private String registerTransactionListenerProcessor() throws InvalidArgumentExce
         });
     }
 
-    private final LinkedHashMap<String, LinkedList<TL>> txListeners = new LinkedHashMap<>();
+    private transient LinkedHashMap<String, LinkedList<TL>> txListeners = new LinkedHashMap<>();
 
     private class TL {
         final String txID;
@@ -2841,9 +2863,9 @@ void fire(BlockEvent.TransactionEvent transactionEvent) {
             }
 
             if (transactionEvent.isValid()) {
-                executorService.execute(() -> future.complete(transactionEvent));
+                client.getExecutorService().execute(() -> future.complete(transactionEvent));
             } else {
-                executorService.execute(() -> future.completeExceptionally(
+                client.getExecutorService().execute(() -> future.completeExceptionally(
                         new TransactionEventException(format("Received invalid transaction event. Transaction ID %s status %s",
                                 transactionEvent.getTransactionID(),
                                 transactionEvent.getValidationCode()),
@@ -2905,7 +2927,7 @@ boolean isMatch(ChaincodeEvent chaincodeEvent) {
 
         void fire(BlockEvent blockEvent, ChaincodeEvent ce) {
 
-            executorService.execute(() -> chaincodeEventListener.received(handle, blockEvent, ce));
+            client.getExecutorService().execute(() -> chaincodeEventListener.received(handle, blockEvent, ce));
 
         }
     }
@@ -2949,7 +2971,7 @@ public String registerChaincodeEventListener(Pattern chaincodeId, Pattern eventN
 
     }
 
-    private String blh = null;
+    private transient String blh = null;
 
     /**
      * Unregister an existing chaincode event listener.
@@ -3063,16 +3085,22 @@ public synchronized void shutdown(boolean force) {
 
         initialized = false;
         shutdown = true;
+        if (chainCodeListeners != null) {
+            chainCodeListeners.clear();
 
-        executorService = null;
+        }
 
-        chainCodeListeners.clear();
+        if (blockListeners != null) {
+            blockListeners.clear();
+        }
 
-        blockListeners.clear();
+        if (client != null) {
+            client.removeChannel(this);
+        }
 
-        client.removeChannel(this);
+        client = null;
 
-        for (EventHub eh : getEventHubs()) {
+        for (EventHub eh : eventHubs) {
 
             try {
                 eh.shutdown();
@@ -3098,12 +3126,66 @@ public synchronized void shutdown(boolean force) {
 
         orderers.clear();
 
-        if (eventQueueThread != null) {
-            eventQueueThread.interrupt();
+        if (null != eventQueueThread) {
+
+            if (eventQueueThread != null) {
+                eventQueueThread.interrupt();
+            }
+            eventQueueThread = null;
+        }
+    }
+
+    /**
+     * Serialize channel to a file using Java serialization.
+     * Deserialized channel will NOT be in an initialized state.
+     *
+     * @param file file
+     * @throws IOException
+     * @throws InvalidArgumentException
+     */
+
+    public void serializeChannel(File file) throws IOException, InvalidArgumentException {
+
+        if (null == file) {
+            throw new InvalidArgumentException("File parameter may not be null");
+        }
+
+        Files.write(Paths.get(file.getAbsolutePath()), serializeChannel(),
+                StandardOpenOption.CREATE, StandardOpenOption.TRUNCATE_EXISTING, StandardOpenOption.WRITE);
+
+    }
+
+    /**
+     * Serialize channel to a byte array using Java serialization.
+     * Deserialized channel will NOT be in an initialized state.
+     *
+     * @throws InvalidArgumentException
+     * @throws IOException
+     */
+    public byte[] serializeChannel() throws IOException, InvalidArgumentException {
+
+        if (isShutdown()) {
+            throw new InvalidArgumentException(format("Channel %s has been shutdown.", getName()));
+        }
+
+        ObjectOutputStream out = null;
+
+        try {
+            ByteArrayOutputStream bai = new ByteArrayOutputStream();
+            out = new ObjectOutputStream(bai);
+            out.writeObject(this);
+            out.flush();
+            return bai.toByteArray();
+        } finally {
+            if (null != out) {
+                try {
+                    out.close();
+                } catch (IOException e) {
+                    logger.error(e); // best effort.
+                }
+            }
         }
-        eventQueueThread = null;
 
-        client = null;
     }
 
     @Override

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

@@ -14,6 +14,7 @@
 
 package org.hyperledger.fabric.sdk;
 
+import java.io.Serializable;
 import java.util.ArrayList;
 import java.util.Properties;
 import java.util.concurrent.CountDownLatch;
@@ -46,27 +47,29 @@
  * Feeds Channel event queues with events
  */
 
-public class EventHub {
+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 static final long EVENTHUB_CONNECTION_WAIT_TIME = config.getEventHubConnectionWaitTime();
-    private final ExecutorService executorService;
+
+    private final transient ExecutorService executorService;
 
     private final String url;
     private final String name;
     private final Properties properties;
-    private ManagedChannel managedChannel;
-    private boolean connected = false;
-    private EventsGrpc.EventsStub events;
-    private StreamObserver<PeerEvents.SignedEvent> sender;
+    private transient ManagedChannel managedChannel;
+    private transient boolean connected = false;
+    private transient EventsGrpc.EventsStub events;
+    private transient StreamObserver<PeerEvents.SignedEvent> sender;
     /**
      * Event queue for all events from eventhubs in the channel
      */
-    private Channel.ChannelEventQue eventQue;
-    private long connectedTime = 0L; // 0 := never connected
-    private boolean shutdown = false;
+    private transient Channel.ChannelEventQue eventQue;
+    private transient long connectedTime = 0L; // 0 := never connected
+    private transient boolean shutdown = false;
     private Channel channel;
-    private TransactionContext transactionContext;
+    private transient TransactionContext transactionContext;
 
     /**
      * Get disconnected time.
@@ -170,7 +173,7 @@ boolean connect() throws EventHubException {
 
     }
 
-    private StreamObserver<PeerEvents.Event> eventStream = null; // Saved here to avoid potential garbage collection
+    private transient StreamObserver<PeerEvents.Event> eventStream = null; // Saved here to avoid potential garbage collection
 
     synchronized boolean connect(final TransactionContext transactionContext) throws EventHubException {
         if (connected) {
@@ -290,8 +293,6 @@ public void onCompleted() {
             logger.error(e);
         }
 
-
-
         if (!threw.isEmpty()) {
             eventStream = null;
             connected = false;
@@ -393,7 +394,7 @@ public interface EventHubDisconnected {
      * Default reconnect event hub implementation.  Applications are free to replace
      */
 
-    protected EventHubDisconnected disconnectedHandler = new EventHub.EventHubDisconnected() {
+    protected transient EventHubDisconnected disconnectedHandler = new EventHub.EventHubDisconnected() {
         @Override
         public synchronized void disconnected(final EventHub eventHub) throws EventHubException {
             logger.info(format("Detected disconnect %s", eventHub.toString()));