laminar.clj

(deftest sse-subscribe-events-test
  (riemann.logging/suppress
   ["riemann.transport" "riemann.core" "riemann.pubsub"]
   (let [s1       (tcp-server)
         s2       (sse-server)
         index    (wrap-index (index/index))
         pubsub   (pubsub/pubsub-registry)
         core     (transition!
                   (core)
                   {:index    index
                    :pubsub   pubsub
                    :services [s1 s2]
                    :streams  [index]})
         client   (client/tcp-client)
         convert  (comp json/parse-string
                        second
                        (partial re-matches #"data: (.*)\n\n")
                        formats/bytes->string)
         response (sync-http-request
                     {:method :get
                      :url    "http://127.0.0.1:5558/index?query=true"})]
     (try           
       (client/send-event client {:service "service1"})
       (client/send-event client {:service "service2"})
       (let [[r2 r1] (->> response
                          :body
                          (take* 2)
                          (map* convert)
                          channel->lazy-seq)]
         (is (#{"service1" "service2"} (get r1 "service")))
         (is (#{"service1" "service2"} (get r2 "service"))))
       (finally
         (stop! core))))))