A simple PubSub system written in Clojure.

pubsub.clj

(defn publish
  ([a & ch-vals]
     (let [key-vals (mapcat (fn [[channel val]]
                              [channel {:val val
                                        :id (java.util.UUID/randomUUID)}])
                         (partition 2 ch-vals))]
       (apply swap! a assoc key-vals))))


(defn subscribe
  [a request-id channel-names sub-fn]
  (add-watch a request-id (fn [k r old-map new-map]
                            (let [changes (keep (fn [[channel-name {:keys [val id]}]]
                                                  (let [change [channel-name val]]
                                                    (if-let [old-channel (old-map channel-name)]
                                                      (when (or (not= val
                                                                      (:val old-channel))
                                                                (not= id
                                                                      (:id old-channel)))
                                                        change)
                                                      change)))
                                                new-map)]
                              (when (seq changes)
                                (sub-fn changes))))))


(defn unsubscribe
  [a request-id]
  (remove-watch a request-id))


(comment
  (def channels (atom {}))

  (subscribe channels :request-id [:a :b :c] (fn [changes]
                                               (println "Changes occured" changes)))

  (publish channels :a 1) ;; Changes occured ([:a 1])

  (publish channels :b 2) ;; Changes occured ([:b 2])

  (publish channels :a 3) ;; Changes occured ([:a 3])

  (publish channels :a 1 :b 2)  ;; Changes occured ([:b 2] [:a 1])
  )