From 8891e402d03c8f20e73f571c6a2720dae0b6afda Mon Sep 17 00:00:00 2001
From: Nicholas Kariniemi <nicholas.kariniemi@gmail.com>
Date: Sun, 28 Jul 2013 09:24:01 +0300
Subject: [PATCH] Synchronize added grubs with other clients

---
 project.clj                    |  2 +-
 src-clj/grub/core.clj          | 49 +++++++++++++++++++-----------
 src-cljs/grub_client/core.cljs | 55 ++++++++++++++++++++++------------
 3 files changed, 68 insertions(+), 38 deletions(-)

diff --git a/project.clj b/project.clj
index a666870..c273d97 100644
--- a/project.clj
+++ b/project.clj
@@ -5,7 +5,7 @@
             :url "http://www.eclipse.org/legal/epl-v10.html"}
   :source-paths ["src-clj"]
   :dependencies [[org.clojure/clojure "1.5.1"]
-                 [http-kit "2.1.5"]
+                 [http-kit "2.1.8"]
                  [compojure "1.1.5"]
                  [ring/ring-devel "1.2.0"]
                  [ring/ring-core "1.2.0"]
diff --git a/src-clj/grub/core.clj b/src-clj/grub/core.clj
index e51def9..f8a1ffd 100644
--- a/src-clj/grub/core.clj
+++ b/src-clj/grub/core.clj
@@ -1,23 +1,36 @@
 (ns grub.core
-  (:use [org.httpkit.server 
-         :only [run-server with-channel on-receive send! websocket?]]
-        [compojure.handler :only [site]]
-        [compojure.core :only [defroutes GET POST]])
   (:require [ring.middleware.reload :as reload]
+            [compojure.core :refer [defroutes GET POST]]
+            [compojure.handler :as handler]
             [compojure.route :as route]
+            [org.httpkit.server :as httpkit]
             [hiccup
              [page :refer [html5]]
-             [page :refer [include-js include-css]]]))
+             [page :refer [include-js include-css]]]
+            [clojure.core.async :as async :refer [<! >! >!! chan go close! timeout]]))
 
-(defn async-handler [request]
-  (if-not (:websocket? request)
-    {:status 200 :body "WebSocket server"}
-    (with-channel request channel
-      (on-receive channel (fn [data]
-                            (send! channel data)))
-      (send! channel {:status 200
-                      :headers {"Content-Type" "text/plain"}
-                      :body    "Long polling?"}))))
+(def out-channels (atom []))
+(def channel-id-count (atom 0))
+
+(defn push-grub-to-others [grub my-channel-id]
+  (let [other-channels (fn [] (filter #(not (= (:id %) my-channel-id)) @out-channels))]
+    (go (doseq [{ch :channel} (other-channels)]
+            (>! ch grub)))))
+
+(defn push-new-grubs-to-client [c ws-channel]
+  (go (while true
+        (let [grub (<! c)]
+          (httpkit/send! ws-channel grub)))))
+
+(defn websocket-handler [request]
+  (httpkit/with-channel request ws-channel
+    (let [channel-id (swap! channel-id-count inc)
+          c (chan)]
+      (swap! out-channels conj {:id channel-id :channel c})
+      (println "channel connected:" (.toString ws-channel))
+      (httpkit/on-receive ws-channel #(push-grub-to-others % channel-id))
+      (push-new-grubs-to-client c ws-channel))))
+      
 (defn index-page []
   (html5
    [:head
@@ -30,7 +43,7 @@
      (include-js "/js/main.js")]))
 
 (defroutes routes
-  (GET "/ws" [] async-handler)
+  (GET "/ws" [] websocket-handler)
   (GET "/" [] (index-page))
   (route/files "/")
   (route/not-found "<p>Page not found.</p>"))
@@ -38,8 +51,8 @@
 (def app 
   (let [dev? true]
     (if dev?
-      (reload/wrap-reload (site #'routes))
-      (site routes))))
+      (reload/wrap-reload (handler/site #'routes))
+      (handler/site routes))))
 
 (defn -main [& args]
-    (run-server app {:port 8080}))
+    (httpkit/run-server app {:port 3000}))
diff --git a/src-cljs/grub_client/core.cljs b/src-cljs/grub_client/core.cljs
index 038968c..e5d0d40 100644
--- a/src-cljs/grub_client/core.cljs
+++ b/src-cljs/grub_client/core.cljs
@@ -2,7 +2,7 @@
   (:require [dommy.core :as dommy]
             [cljs.core.async :as async :refer [<! >! chan close! timeout]])
   (:require-macros [dommy.macros :refer [deftemplate sel1 node]]
-                   [cljs.core.async.macros :as m :refer [go alt!]]
+                   [cljs.core.async.macros :as m :refer [go alt! alts!]]
                    [grub-client.macros :refer [log]]))
 
 (deftemplate grub-template [grub]
@@ -27,12 +27,11 @@
       [:span.input-group-btn
        add-grub-btn]]
      [:table.table.table-condensed
-      [:tbody#grubList
-       (for [grub grubs] (grub-template grub))]]]
+      [:tbody#grubList]]]
     [:div.col-lg-4]]])
 
-(defn render-body [grubs]
-  (dommy/prepend! (sel1 :body) (main-template grubs)))
+(defn render-body []
+  (dommy/prepend! (sel1 :body) (main-template)))
 
 (defn push-new-grub [channel]
   (let [new-grub (dommy/value add-grub-text)]
@@ -60,10 +59,14 @@
 (defn append-new-grub [grub]
   (dommy/append! (sel1 :#grubList) (grub-template grub)))
 
+(defn append-new-grubs [chan]
+  (go (while true
+        (let [grub (<! chan)]
+          (append-new-grub grub)))))
+
 (defn add-grubs-to-list [in]
   (go (while true 
         (let [new-grub (<! in)]
-          (log "Added grub: " new-grub)
           (append-new-grub new-grub)))))
 
 (defn filter-empty-grubs [in]
@@ -73,21 +76,35 @@
             (when-not (empty? grub) (>! out grub)))))
     out))
 
-(defn add-new-grubs-to-list []
-  (let [added-grubs (get-added-grubs)
-        filtered-grubs (filter-empty-grubs added-grubs)]
-    (add-grubs-to-list filtered-grubs)))
+(def websocket* (atom nil))
 
-(def test-grubs
-  ["8 dl water"
-   "8 whole peppercorns"
-   "2 bay leaves"
-   "1 - 2 (150 g) onions"
-   "2 dl cream"
-   "1 dl dill"])
+(defn push-grubs-to-server [chan]
+  (let [websocket (js/WebSocket. "ws://localhost:3000/ws")]
+    (aset websocket "onmessage" (fn [event] 
+                                  (let [grub (.-data event)]
+                                    (log "Received grub:" grub)
+                                    (append-new-grub grub))))
+    (go (while true
+          (let [grub (<! chan)]
+            (.send websocket grub))))))
+
+(defn fan-out [in num-chans]
+  (let [out-channels (repeatedly num-chans chan)]
+   (go (while true
+             (let [x (<! in)]
+               (doseq [out out-channels]
+                 (>! out x)))))
+   out-channels))
+
+(defn add-new-grubs-as-they-come []
+  (let [added-grubs (get-added-grubs)
+        filtered-grubs (filter-empty-grubs added-grubs)
+        out-channels (fan-out filtered-grubs 2)]
+    (append-new-grubs (first out-channels))
+    (push-grubs-to-server (second out-channels))))
 
 (defn init []
-  (render-body test-grubs)
-  (add-new-grubs-to-list))
+  (render-body)
+  (add-new-grubs-as-they-come))
 
 (init)