foo.clj

(ns private.tmp.foo
  (:require [manifold.stream :as s]
            [clojure.core.async :as a]))

(def sock (atom nil))
(def destination-chan (a/chan))
(defn get-sock [] (s/stream))
(defn stay-connected [sock-atom sock-fn dest-chan]
  (letfn [(disconnect-handler []
            (reset! sock-atom (sock-fn))
            (s/connect @sock-atom dest-chan {:downstream? false})
            (s/on-closed @sock-atom disconnect-handler))]
  (reset! sock (sock-fn))
  (s/connect @sock-atom dest-chan {:downstream? false})
  (s/on-closed @sock-atom disconnect-handler)))

;; connect up my sock atom with the destination chan in a resilient fashion
(stay-connected sock get-sock destination-chan)

;; Put a message onto destination-chan via @sock
(s/put! @sock "Hello")

;; Take it out of destination-chan
(a/go (println (a/<! destination-chan)))

;; Close the sock, which should get re-opened and re-connected to destination-chan automatially
(s/close! @sock)

;; Put a new message onto destination-chan via the reconnected @sock
(s/put! @sock "World")

;; Show that it ends up on destination chan as well
(a/go (println (a/<! destination-chan)))