From 6572ce55b31b8bb7491cfd517307c3acf679704e Mon Sep 17 00:00:00 2001
From: Artur Girenko <a.girenko@gmail.com>
Date: Sat, 25 Nov 2017 22:25:23 +0100
Subject: [PATCH] support passing a event provider function to dispatch

---
 project.clj                      | 10 +++++-----
 src/clj/pneumatic_tubes/core.clj | 22 +++++++++++++---------
 2 files changed, 18 insertions(+), 14 deletions(-)

diff --git a/project.clj b/project.clj
index bf89aaf..df832ba 100644
--- a/project.clj
+++ b/project.clj
@@ -1,15 +1,15 @@
-(defproject pneumatic-tubes "0.2.0"
+(defproject pneumatic-tubes "0.3.0-SNAPSHOT"
   :description "WebSocket based transport of events between re-frame app and server"
   :url "https://github.com/drapanjanas/pneumatic-tubes"
   :license {:name "Eclipse Public License"
             :url  "http://www.eclipse.org/legal/epl-v10.html"}
-  :dependencies [[org.clojure/clojure "1.9.0-alpha16"]
-                 [org.clojure/clojurescript "1.9.521"]
+  :dependencies [[org.clojure/clojure "1.9.0-RC1"]
+                 [org.clojure/clojurescript "1.9.946"]
                  [org.clojure/tools.logging "0.3.1"]
-                 [org.clojure/core.async "0.3.442"]
+                 [org.clojure/core.async "0.3.442" :scope "provided"]
                  [com.cognitect/transit-cljs "0.8.239"]
                  [com.cognitect/transit-clj "0.8.300"]
-                 [http-kit "2.2.0"]]
+                 [http-kit "2.2.0" :scope "provided"]]
 
   :scm {:name "git"
         :url  "https://github.com/drapanjanas/pneumatic-tubes"}
diff --git a/src/clj/pneumatic_tubes/core.clj b/src/clj/pneumatic_tubes/core.clj
index 912a423..fab3811 100644
--- a/src/clj/pneumatic_tubes/core.clj
+++ b/src/clj/pneumatic_tubes/core.clj
@@ -1,7 +1,7 @@
 (ns pneumatic-tubes.core
   (:use [clojure.tools.logging :only [info warn error]]
         [clojure.set])
-  (:require [clojure.core.async :refer [<! >!! go-loop chan]]))
+  (:require [clojure.core.async :refer [<! >! go go-loop chan]]))
 
 ;; -------- tube-registry-------------------------------------------------------------------
 
@@ -115,7 +115,7 @@
 (defn receive
   "Asynchronously process the incoming event"
   ([receiver from event-v]
-   (>!! (:in-queue receiver) {:from from :event event-v})))
+   (go (>! (:in-queue receiver) {:from from :event event-v}))))
 
 (defn wrap-handlers
   "Wraps a map of handlers with one or more middlewares"
@@ -130,19 +130,23 @@
 
 (defn dispatch
   "Send event vector to one or more tubes.
-  Destination (parameter 'to') can be a map, a predicate function or :all keyword "
-  ([transmitter to event-v]
-   (let [target-tube-ids (map :tube/id (find-tubes-by-criteria @tube-registry to))]
-     (>!! (:out-queue transmitter) {:to target-tube-ids :event event-v})) to))
+  Destination (parameter 'to') can be a map, a predicate function or :all keyword.
+  The event vector (parameter event) ca be either a vector or a function which takes a tube and returns event vector"
+  ([transmitter to event]
+   (let [out-chan (:out-queue transmitter)
+         target-tubes (find-tubes-by-criteria @tube-registry to)
+         event-provider (if (fn? event) event (fn [_] event))]
+     (go (doseq [tube target-tubes]
+           (>! out-chan {:to (:tube/id tube) :event (event-provider tube)}))))
+   to))
 
 (defn- send-to-tube [tube-registry tube-id event-v]
   (let [send! (get-in tube-registry [:send-fns tube-id])]
     (send! event-v)))
 
 (defn- handle-outgoing
-  [tube-registry {tube-ids :to event-v :event}]
-  (doseq [tube-id tube-ids]
-    (send-to-tube tube-registry tube-id event-v)))
+  [tube-registry {tube-id :to event-v :event}]
+  (send-to-tube tube-registry tube-id event-v))
 
 (defn- call-listeners [on-send {to :to event-v :event}]
   (when on-send