test_dispatcher.clj

(require '[clojure.tools.logging :refer [info error]]
         '[clojure.core.async :as async]
         '[com.stuartsierra.component :as component]
         '[de.mpg.shh.util-message-server.message-server :as message-server]
         '[de.mpg.shh.util-message-server.message-dispatcher :as message-dispatcher]
         '[de.mpg.shh.util-message-server.helpers :as message-helpers]
         '[taoensso.encore :as enc])

(defn test-array
  [t]
  (let [check (type (t []))]
    (fn [arg] (instance? check arg))))

(def byte-array?
  (test-array byte-array))

(defn decode-bytes-message
  [maybe-bytes]
  (if (byte-array? maybe-bytes)
    (String. maybe-bytes "UTF-8")
    maybe-bytes))

;; edn transducer
(def dispatcher-request-transducer
  (comp
    (map decode-bytes-message)
    (map enc/read-edn)))

(def message-dispatcher-config
  {:listen-context-config {:host "127.0.0.1"
                           :port 5445
                           :username "shh-test"
                           :password "batman"}
   :listen-address "queue.shh.test.request"
   :dispatcher-request-buffer-size 16
   :dispatcher-request-transducer dispatcher-request-transducer
   :end-points {:test-endpoint-one (fn [request-channel ?data] (info "test-end-point-one: " ?data))
                :test-endpoint-two (fn [request-channel ?data] (info "test-end-point-two: " ?data))}
   :stop-message "[:stop]"})

(def selective-message-dispatcher-config
  (assoc message-dispatcher-config :listen-address "queue.shh.test.response"
                                   :listen-selector "origin <> 'foo'"))

(defonce system (let [request-channel (async/chan)
                      system (atom (component/system-map :request-channel request-channel
                                                         :message-server (component/using (message-server/map->MessageServer {}) [:request-channel])
                                                         :message-dispatcher (component/using (message-dispatcher/map->MessageDispatcher message-dispatcher-config) [:message-server])
                                                         :selective-message-dispatcher (component/using (message-dispatcher/map->MessageDispatcher selective-message-dispatcher-config) [:message-server])))]
                  system))



(info "Hello from Dispatcher test")
(swap! system component/start)


;; Test new synch-send
(info "synch-send: " (message-helpers/synch-send (get-in @system [:request-channel]) "127.0.0.1" 5445 "shh-test" "batman" "queue.shh.test.request" "[:test-endpoint-one \"Hello from test one\"]" :timeout 1))

;; Test selectors
(info "synch-send: " (message-helpers/synch-send (get-in @system [:request-channel]) "127.0.0.1" 5445 "shh-test" "batman" "queue.shh.test.response" "[:test-endpoint-two \"One, two, \"]" :timeout 1))
(info "synch-send: " (message-helpers/synch-send (get-in @system [:request-channel]) "127.0.0.1" 5445 "shh-test" "batman" "queue.shh.test.response" "[:test-endpoint-two \"skip a few, \"]" :timeout 1 :properties {"origin" "foo"}))
(info "synch-send: " (message-helpers/synch-send (get-in @system [:request-channel]) "127.0.0.1" 5445 "shh-test" "batman" "queue.shh.test.response" "[:test-endpoint-two \"ninety nine, one hundred\"]" :timeout 1 :properties {"origin" "bar"}))


(Thread/sleep 1400)
(info "finish sleep")
(swap! system component/stop)
(System/exit 0)