   [[:PuppetDBServer shared-globals]
    [:MessageListenerService register-listener]]
   (init [this context]
-    (let [response-pub-chan (async/chan)]
+    (let [response-chan (async/chan)
+          response-mult (async/mult response-chan)
+          response-chan-for-pub (async/chan)]
+      (async/tap response-mult response-chan-for-pub)
       (assoc context
              :stats (atom {:received-commands 0
                            :executed-commands 0})
-             :response-pub-chan response-pub-chan
-             :response-pub (async/pub response-pub-chan :id))))
+             :response-chan response-chan
+             :response-mult response-mult
+             :response-chan-for-pub response-chan-for-pub
+             :response-pub (async/pub response-chan-for-pub :id))))
 
   (start [this context]
     (let [{:keys [scf-write-db catalog-hash-debug-dir]} (shared-globals)
           config {:db scf-write-db
                   :catalog-hash-debug-dir catalog-hash-debug-dir}
-          {:keys [response-pub-chan response-pub]} context]
+          {:keys [response-chan response-pub]} context]
       (register-listener
        supported-command?
-       #(process-command-and-respond! % config response-pub-chan (:stats context)))
+       #(process-command-and-respond! % config response-chan (:stats context)))
       context))
 
-  (stop [this {:keys [response-pub-chan] :as context}]
-    (async/close! response-pub-chan)
-    (dissoc context :response-pub :response-pub-chan))
+  (stop [this context]
+    (async/unsub-all (:response-pub context))
+    (async/untap-all (:response-mult context))
+    (async/close! (:response-chan-for-pub context))
+    (async/close! (:response-chan context))
+    (dissoc context :response-pub :response-chan :response-chan-for-pub :response-mult))
 
   (stats [this]
     @(:stats (service-context this)))

      result))

  (response-pub [this]
    (-> this service-context :response-pub)))

    count of the commands that the current instance has processed
    without triggering an exception).")

  (response-mult [this]
    "Returns a core.async mult to which {:id :exception} maps are written after
     each command has been processed. " )

  (response-pub [this]
    "Returns a core.async pub to which {:id :exception} maps are written after
     each command has been processed. The topic for each message is its
     id (command uuid)."))

(defn make-cmd-processed-message [cmd ex]
  (merge {:id (-> cmd :annotations :id)}
         (when ex {:exception ex})))

(defn process-command-and-respond! [cmd config response-pub-chan stats-atom]
