         (when-not (.isEmpty ^Queue queue)
           (schedule this))))))
 
+(declare terminate)
+
 (deftype Publisher [source subscriptions options]
   clojure.lang.Seqable
   (seq [p]

               this
               subscriber)]
      (.add ^Set subscriptions sub)
      (signal-subscribe sub)
      sub)))

;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;

        ^Subscriber subscriber (.-subscriber sub)
        source (.-source publisher)
        canceled (.-canceled sub)]
    (try
      (.onSubscribe subscriber sub)
      (catch Throwable t
        (terminate sub (IllegalStateException. "Violated the Reactive Streams rule 2.13"))))
    (when (asyncp/closed? source)
      (try
        (handle-cancel sub)
        (.onComplete subscriber)
        (catch Throwable t
          ;; (IllegalStateException. "Violated the Reactive Streams rule 2.13")
          )))))

(defn- handle-send
  [sub]
