         data-list (:data clj-block)]
     (assert (= (count tx-status-codes) (count data-list)))
     ;; 1. Block Event on all channels
-    (async/put! (:block-event-chan event-hub) clj-block)
+    (async/put! (deref (:block-event-chan event-hub)) clj-block)
     (doseq [[data code] (map list data-list tx-status-codes)]
       (let [payload (:payload data)
             channel-header (get-in payload [:header :channel-header])
             tx-id (:tx-id channel-header)]
         ;; 2. Transaction Event with tx-id
-        (when-let [ch (tx-id->chan tx-id)]
+        (when-let [ch (tx-id->chan event-hub tx-id)]
           (async/put! ch (make-TxEvent tx-id code)))
         ;; 3. Chaincode event
         (when (= (:type channel-header) (proto/header-types :endorser-transaction))
           (let [chaincode-event (get-in payload [:data :actions 0 :payload :action
                                                  :proposal-response-payload :extension :events])]
-            (when-let [ch (chaincode-id->chan (:chaincode-id chaincode-event))]
+            (when-let [ch (chaincode-id->chan event-hub (:chaincode-id chaincode-event))]
               (async/put! ch chaincode-event))))))))
 
 (defn transaction-observer
-  [ch]
+  [event-hub]
   (reify StreamObserver
     (onNext [this event]
-      (case (.getEventCase ^EventsPackage$Event event)
-        EventsPackage$Event$EventCase/BLOCK (deliver-block-events-to-channels (.getBlock ^EventsPackage$Event event))
-        EventsPackage$Event$EventCase/REGISTER nil ;; init
-        ;; EventsPackage$Event$EventCase/CHAINCODE_EVENT ;; FIXME: not used??
-        EventsPackage$Event$EventCase/UNREGISTER nil;; shutdown
+      (condp = (.getEventCase ^EventsPackage$Event event)
+
+        EventsPackage$Event$EventCase/REGISTER
+        (add-event-go-loop! event-hub)
+        
+        EventsPackage$Event$EventCase/BLOCK
+        (deliver-block-events-to-channels event-hub (.getBlock ^EventsPackage$Event event))
+        
+        EventsPackage$Event$EventCase/CHAINCODE_EVENT nil
+        EventsPackage$Event$EventCase/REJECTION nil
+
+        EventsPackage$Event$EventCase/UNREGISTER
+        (with-issuing-event-go-loop-command! [event-hub :quit])
+        
+        EventsPackage$Event$EventCase/FILTERED_BLOCK nil
         ))
     (onError [this err]
-      ;; FIXME call shutdownNow when Status$Code/INTERNAL or Status$Code/UNAVAILABLE
-      (async/put! ch err))
+      ;; FIXME:
+      ;; 1. log error
+      ;;      call shutdownNow when Status$Code/INTERNAL or 
+      (when (and (instance? StatusRuntimeException err)
+                 (->> (.getStatus ^StatusRuntimeException err)
+                      (.getCode)
+                      (contains? #{Status$Code/INTERNAL Status$Code/UNAVAILABLE})))
+        (with-issuing-event-go-loop-command! [event-hub :quit])))
     (onCompleted [this]
-      (async/put! ch :done))))
-
-#_
+      (with-issuing-event-go-loop-command! [event-hub :quit]))))
+
+
 ;;https://grpc.io/docs/tutorials/basic/java.html
-(defn make-event-hub
-  [user peer]
-  (let [ch (async/chan (async/sliding-buffer default-sliding-buffer-size))
-        ^StreamObserver observer (transaction-observer ch)
-        ^ManagedChannel channel (proto/node->channel peer)]
-    (.chat (EventsGrpc/newStub channel)
-           observer)
-    (let [interest (-> (EventsPackage$Interest/newBuilder)
-                       (.setEventType EventsPackage$Event$EventCase/BLOCK)
-                       (.build))
-          register (-> (EventsPackage$Register/newBuilder)
-                       (.addEvents interest)
-                       (.build))
-          block-event (-> (EventsPackage$Event/newBuilder)
-                          (.setRegister register)
-                          (.setCreator (.toByteString ^Identities$SerializedIdentity
-                                                      (proto/clj->proto (proto/user->serialized-identity user))))
-                          (.build))]
-      (.onNext observer
-               (-> (EventsPackage$SignedEvent/newBuilder)
-                   (.setEventBytes (.toByteString block-event))
-                   (.setSignature (ByteString/copyFrom ^bytes (crypto/sign (.toByteArray block-event) (:private-key user) {:algorithm (:key-algorithm (:crypto-suite user))})))
-                   (.build)))
-      (try
-        (async/go-loop []
-          (let [block (async/<! ch)]
-            (case (.getEventCase ^EventsPackage$Event event)
-              EventsPackage$Event$EventCase/BLOCK (async/put! ch event) ;; ch is like eventQueue
-              EventsPackage$Event$EventCase/REGISTER :FIXME ;; connected, connectedTime
-              )
-            (cond= (type v)
-                   
-                   (instance? Exception v) (when-not (.isShutdown channel)
-                                             (.shutdownNow channel))
-                   ;; FIXME: do something here!!
-                   :else (recur))))
-        (finally (.onCompleted observer))))))
-
 (defn connect
   ([]
    (connect *event-hub*))
   ([event-hub]
-   (let [{:keys [ch user peer]} event-hub
-         ^StreamObserver observer (transaction-observer ch)
+   (let [{:keys [user peer]} event-hub
+         ^StreamObserver observer (transaction-observer event-hub)
          ^ManagedChannel channel (proto/node->channel peer)
          req-observer (.chat (EventsGrpc/newStub channel) observer)
          interest (-> (EventsPackage$Interest/newBuilder)

  (:import io.grpc.stub.StreamObserver
           [clojure_fabric.proto Block ChaincodeEvent]
           [com.google.protobuf ByteString]
           io.grpc.ManagedChannel
           org.hyperledger.fabric.protos.msp.Identities$SerializedIdentity
           [org.hyperledger.fabric.protos.common Common$Block Common$BlockMetadataIndex

  [event-hub]
  (letfn [(%get-all-chans [k]
            (map :ch (vals (deref (k event-hub)))))]
    `[~@(:command-ch ~event-hub)
      ~@(:block-event-chan ~event-hub)
      ~@(%get-all-chans :tx-id->chan-fn-map)
      ~@(%get-all-chans :chaincode-id->chan-fn-map)]))

