simulator.clj

(ns sossity.simulator
  (:require
   [clojure.core.async :as a
    :refer [>! <! >!! <!! go chan buffer close! thread alts! alts!! timeout pub sub unsub unsub-all go-loop put!]]

   [loom.graph :refer :all]
   [loom.alg :refer :all]
   [loom.io :refer :all]
   [loom.attr :refer :all]))

;use doseq or doall to realize all the pubs?

(defn take-and-print [channel prefix]
  (go-loop []
    (println prefix " sink: " (<! channel))
    (recur)))

(defn pass-on [in-channel node out-channels]
  (go-loop []
    (let [v (<! in-channel)]
      (println node ": " v)
      (doseq  [c out-channels]
        (println "channel" c)
        (>! c (assoc v :chans (conj (:chans v) node)))))
    (recur)))

(def top (digraph {:a [:b :c] :b [:d] :c nil :d nil}))

(defn build-node [input-pub input-topic  output-topics]
  "returns a seq of output publications for a node to talk to"
  (let [input (chan)
        out-chans (repeatedly (count output-topics) chan)
        out-pubs (doall (map #(pub % (fn [x] (:topic x))) out-chans))]
    (sub input-pub input-topic input)
    (pass-on input input-topic out-chans)
    out-pubs))

(defn build-end-node [input-pub input-topic]
  (let [input (chan)]
    (sub input-pub input-topic input)
    (take-and-print input "out")))

(defn test-build-node []
  (let
   [publisher (chan)
    publication (pub publisher #(:topic %))
    out-topics ["unitA" "unitB"]]

    (let [out-pubs (build-node publication :pipeA out-topics)]
      (doall (map #(let [sc (chan)]
                     (sub %1 :pipeA sc)
                     (take-and-print sc %2)) out-pubs out-topics)))
    (put! publisher {:topic :pipeA :dest "zzz/#home"})))

(defn test-build-node-b []
  (let
   [publisher (chan)
    publication (pub publisher #(:topic %))
    out-nodes ["unitA" "unitB"]]
    (let [out-pubs (build-node publication :pipeA out-nodes)]
      (doseq [p out-pubs]
        (build-end-node p :pipeA)))
    (put! publisher {:topic :pipeA :dest "zzz/#home"})))

;;do a reduce to collect the pubs?

#_(defn handle-graph-node [g node in-pub]
    (if (= 0 (in-degree g node))                                          ;initial node?
)

    (if (> 0 (out-degree g node))
      (build-node)))

(defn compose-cluster [g])

(defn pubsubtest [derp]
  (let [publisher (chan)
        publication (pub publisher #(:topic %))

        out-one (chan)
        out-pub (pub out-one #(:topic %))

        out-two (chan)
        out-two-pub (pub out-two #(:topic %))

        subscriber-one (chan)

        subscriber-two-a (chan)

        subscriber-two-b (chan)]

    (sub publication :account-created subscriber-one)

    (pass-on subscriber-one "subscriber-one" [out-one out-two])

    (sub out-pub :account-created     subscriber-two-a)
    (sub out-two-pub :account-created     subscriber-two-b) (take-and-print subscriber-two-a "subscriber-two-a")
    (take-and-print subscriber-two-b "subscriber-two-b")

    (put! publisher {:topic :account-created :dest "/#home"})))