 (defn envelope-broadcast-observer
   [ch]
   (reify StreamObserver
-    (onNext [this deliver-response]
-      (async/put! ch deliver-response)
-      (when (= (.getTypeCase ^Ab$DeliverResponse deliver-response)
-               Ab$DeliverResponse$TypeCase/STATUS)
-        (async/put! ch :done)))
+    (onNext [this boradcast-response]
+      (async/put! ch boradcast-response))
     (onError [this err]
       (async/put! ch err))
-    (onCompleted [this])))
+    (onCompleted [this]
+      (async/put! ch :done))))
 
 ;; typeout values (from Java SDK)
 ;; defaultProperty(PROPOSAL_WAIT_TIME, "20000");

  (clj->proto [this]
    (cond-> (Configtx$ConfigPolicy/newBuilder)
        version (.setVersion version)
        policy (.setPolicy (clj->proto policy))
        mod-policy (.setModPolicy mod-policy)
        true (.build))))



(defn broadcast-via-orderer
  [orderer envelope]
  (letfn [(broadcast-via-orderer-using-async []
            (let [ch (async/chan 32)
                  callback (envelope-broadcast-observer ch)
                  observer (.broadcast (AtomicBroadcastGrpc/newStub ^ManagedChannel (node->channel orderer))
                                       callback)]
              (.onNext observer ^Common$Envelope (clj->proto envelope))
              (try
                (async/go-loop [result []]
                  (let [[v ch] (async/alts! [ch (async/timeout (:orderer-wait-time timeouts))])]
                    (cond (= v :done) result
                          (nil? v) (throw (Exception. "Timeout to send Evelope using orderer"))
                          (instance? Exception v) (throw v)
                          :else (recur (conj result v)))))
                (finally (.onCompleted observer)))))]
    (async/<!! (broadcast-via-orderer-using-async))))

;;;;
(comment
