                    :else (recur))))
         (finally (.onCompleted observer))))))
 
-
-(defrecord EventHub [peer-address peer-tls-certificate peer-tls-certificate-override
-                     tx-registrants
-                     connected
-                     mtx chaincode-registrants block-registrants  
-                     grpc-client
-                     interested-events
-                     event-client-factory client])
-
-(defn make-event-hub [m]
-  (map->EventHub m))
-
-(def ^:dynamic *event-hub* nil)
-
 (defn connect
   ([]
-   (connect event-hub))
+   (connect *event-hub*))
+  ([event-hub]
+   (let [ch (:ch event-hub)
+         ^StreamObserver observer (transaction-observer ch)
+         ^ManagedChannel channel (proto/node->channel peer)
+         req-observer (.chat (EventsGrpc/newStub channel) observer)
+         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 req-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)))
+     (swap! (:observer event-hub) req-observer))))
+
+
+(defn disconnect
+  ([]
+   (disconnect *event-hub*))
   ([event-hub]
    (when-not (:connected event-hub)
      (if-not (:peer-address event-hub)
        (throw (Exception. "Peer address is missing!"))
        ))))
 
-
-;; connect
-;; disconnect
-;; get-peer-address
-;; connected?
+(defn get-peer-address
+  ([]
+   (get-peer-address *event-hub*))
+  ([event-hub]
+   (when-not (:connected event-hub)
+     (if-not (:peer-address event-hub)
+       (throw (Exception. "Peer address is missing!"))
+       ))))
+
+(defn connected?
+  ([]
+   (connected? *event-hub*))
+  ([event-hub]
+   (when-not (:connected event-hub)
+     (if-not (:peer-address event-hub)
+       (throw (Exception. "Peer address is missing!"))
+       ))))
+
+;; Not required
+;; 
 ;; set-peer-address
-;; unregister-block-event
-;; unregister-chaincode-event
-;; unregister-tx-event
-
 ;; set-interests
 ;; add-chaincode-interest
 ;; remove-chaincode-interest

;;                      :filtered-block})

(defonce default-sliding-buffer-size 32)
(defonce command-ch (async/chan))
;; Block event - any block events
(defonce block-event-chan (async/sliding-buffer default-sliding-buffer-size))
(defonce block-event-handlers (atom {}))
;; Transaction event - matching tx-id
(defonce tx-id->chan-fn-map (atom {}))
;; Chaincode event - matching chaincode-id
(defonce chaincode-id->chan-fn-map (atom {}))

(defrecord ChanFun [ch fn])
(defn make-chan-fun
  [fn]
  (->ChanFun (async/chan) fn))

(defn register-block-event
  [fn-name fn]
  (swap! block-event-handlers assoc fn-name fn)
  (async/put! command-ch :refresh))

(defn unregister-block-event
  [fn-name]
  (swap! block-event-handlers dissoc fn-name)
  (async/put! command-ch :refresh))

(defn- %unregister-event
  [id ref]
  (when-let [existing (get (deref ref) id)]
    (async/close! (:ch existing))
    (swap! ref dissoc id))
  (async/put! command-ch :refresh))

(defn- %register-event
  [id fn ref]
  (when-let [existing (get (deref ref) id)]
    (async/close! (:ch existing)))
  (swap! ref assoc id (make-chan-fun fn))
  (async/put! command-ch :refresh))

(defn register-tx-event
  [tx-id fn]
  (%register-event tx-id fn tx-id->chan-fn-map))

(defn unregister-tx-event
  [tx-id]
  (%unregister-event tx-id tx-id->chan-fn-map))

(defn register-chaincode-event
  [chaincode-id fn]
  (%register-event chaincode-id fn chaincode-id->chan-fn-map))

(defn unregister-chaincode-event
  [chaincode-id]
  (%unregister-event chaincode-id chaincode-id->chan-fn-map))

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

;; Block event is proto/Block
;; Chaincode event is proto/ChaincodeEvent

;;      - Java: part of the Java SDK for Hyperledger Fabric. Source code can be found here
;;      - Golang: an example event listener client can be found here
;;

(defrecord EventHub [peer user ch ;; ch - output ch
                     command-ch
                     block-event-chan ;; input ch
                     block-event-handlers
                     tx-id->chan-fn-map
                     chaincode-id->chan-fn-map
                     observer

                     ;; connected
                     ;; mtx
                     ;; grpc-client
                     ;; interested-events
                     ;; event-client-factory client
                     ])

(defn make-event-hub [user peer]
  (->EventHub user peer (async/chan (async/sliding-buffer default-sliding-buffer-size))
              (async/chan) ;; command ch
              ;; block event ch & handlers
              (async/chan (async/sliding-buffer default-sliding-buffer-size))
              (atom {})
              ;; tx-id->chan-fn-map
              (atom {})
              ;; chaincode-id->chan-fn-map
              (atom {})
              (atom nil)))

(def ^:dynamic *event-hub* nil)

(defonce block-metadata-index-map
  (zipmap [:signatures :last-config :transactions-filter :orderer]
