Add world state json server to clj

Author: Chouser

clj/src/n01se/ws.clj

@@ -1,5 +1,4 @@
 (ns n01se.ws
-  (:require [clj-json.core :as json])
   (:import [org.eclipse.jetty.websocket
             WebSocket$OnTextMessage WebSocketClientFactory WebSocket$Connection
             WebSocketServlet]
@@ -28,21 +27,23 @@
                  (onMessage [this data] (onmessage this data))))
         (.get 10 TimeUnit/SECONDS))))
 
-(defn sendjson [ws obj]
-  (.sendMessage ^WebSocket$Connection ws (json/generate-string obj)))
+(defn send-message [^WebSocket$Connection ws text]
+  (.sendMessage ws text))
 
+(defn close [^WebSocket$Connection ws]
+  (close ^WebSocket$Connection ws))
 
-(defn ws-server
+(defn ^Server ws-server
   [& {:keys [port onopen onclose onmessage ws-path]
       :or {port 8080
            ws-path "/"}}]
   (let [onopen (or onopen (fn [this conn] (println "onopen:" conn)))
         onclose (or onclose (fn [this code msg] (println "onclose:" code msg)))
-        onmessage (or onmessage (fn [this data] (println "onmessage:" data)))
+        onmessage (or onmessage (fn [this data] (println "onmessage:" this data)))
         servlet (proxy [WebSocketServlet] []
                   (doGet [request response]
-                    (.. (proxy-super getServletContext)
-                        (getNamedDispatcher (proxy-super getServletName))
+                    (.. (.getServletContext ^WebSocketServlet this)
+                        (getNamedDispatcher (.getServletName ^WebSocketServlet this))
                         (forward request response)))
                   (doWebSocketConnect [request response]
                     (reify WebSocket$OnTextMessage
@@ -51,7 +52,7 @@
                       (onMessage [this data] (onmessage this data)))))
         context (doto (ServletContextHandler.)
                   (.setContextPath "/")
-                  (.addServlet (ServletHolder. servlet) ws-path))
+                  (.addServlet (ServletHolder. servlet) ^String ws-path))
         connector (doto (SelectChannelConnector.)
                     (.setPort port)
                     (.setMaxIdleTime Integer/MAX_VALUE))

clj/src/n01se/xkcd1110.clj

@@ -1,9 +1,16 @@
 (ns n01se.xkcd1110
   (:require [clj-json.core :as json]
-            [n01se.ws :as ws]))
+            [n01se.ws :as ws] ))
 
 (set! *warn-on-reflection* true)
 
+(def max-clients 20)
+(def max-msg-length 500)
+(def push-interval 75)
+
+(defn send-one [ws obj]
+  (ws/send-message ws (json/generate-string obj)))
+
 (defn round [num places]
   (let [factor (Math/pow 10 places)]
     (/ (Math/round (* num factor)) factor)))
@@ -15,10 +22,10 @@
         speed 0.2
         delay 150
         halfpi (/ Math/PI 2)
-        ws (ws/ws-client uri :onmessage (fn [_]))]
-    (ws/sendjson ws {:nick (str "circlebot " i)})
+        ws (ws/ws-client uri :onmessage (fn [_ _]))]
+    (send-one ws {:nick (str "circlebot " i)})
     (loop [theta (* 0.3 i)]
-      (ws/sendjson ws
+      (send-one ws
               {:dx (round (* (Math/cos (+ theta halfpi)) speed) 4)
                :dy (round (* (Math/sin (+ theta halfpi)) speed) 4)
                :x (Math/round (+  -823 (* (Math/cos theta) radius)))
@@ -27,14 +34,100 @@
       (Thread/sleep delay)
       (recur (+ theta (/ (* speed delay) 150))))))
 
+;; server state:
+(defonce last-id (atom 0))
+(defonce client-struct (ref {})) ;; ref of map of client obj to thing
+;; could be another ref in the struct:
+(defonce changes (ref {})) ;; ref of map of client obj to client-delta
+(defonce publisher (agent 0))
+
+(defn get-world []
+  (dosync
+    (into {} (for [{:keys [id data-ref]} (vals @client-struct)]
+               [id @data-ref]))))
+
+(defn send-all [obj]
+  (let [txt (json/generate-string obj)]
+    (doseq [struct (vals @client-struct)]
+      (try
+        (ws/send-message ^WebSocket$Connection (:conn struct) txt)
+        (catch java.io.IOException e
+          ;; presumably this connection will be closed shortly...
+          nil)))))
+
+
+(defn get-id [client]
+  (get-in @client-struct [client :id]))
+
+(defn publish [i]
+  (when (seq @changes)
+    (let [old-changes
+            (dosync
+              (let [old-changes @changes]
+                (alter changes empty)
+                old-changes))]
+      (send-all {:change
+                  (zipmap (map get-id (keys old-changes))
+                          (vals old-changes))})))
+  ;; do it again:
+  (Thread/sleep push-interval)
+  (send-off *agent* publish)
+  (inc i))
+
+(defn remove-client [client]
+  (let [id (get-id client)]
+    (dosync
+      (commute client-struct dissoc client)
+      (commute changes dissoc client))
+    (println "Client count:" (count @client-struct))
+    (send-all {:delete id})))
+
+(defn server-onopen [client conn]
+  (let [id (swap! last-id inc)
+        struct {:id id, :conn conn, :data-ref (ref {})} ]
+    (println "New client ID" id conn)
+    (dosync
+      (commute client-struct assoc client struct))
+    (send-one conn
+      {:id id, :all (get-world)})))
+
+(defn server-onclose [client code msg]
+  (println "Client ID" (get-id client) "disconnected")
+  (remove-client client))
+
+(defn server-onmessage [client data]
+  (if (> (count data) max-msg-length)
+    (do
+      (println "Client ID" (get-id client) "sent oversize message"
+              (count data) "chars")
+      (ws/close (get-in @client-struct [client :conn])))
+    (let [id (get-id client)
+          msg (json/parse-string data true)]
+      (when-not (:whomp msg)
+        (dosync
+          (commute changes update-in [client] merge msg)
+          (let [data-ref (get-in @client-struct [client :data-ref])]
+            (alter data-ref merge msg)))))))
+
+(defn carefully [f]
+  (fn [& args]
+    (try
+      (apply f args)
+      (catch Exception e
+        (.printStackTrace e)
+        (throw e)))))
+
 (defn -main [& [mode arg1 arg2]]
-  (condp = mode
+  (case mode
     "server"
     (let [port (if arg1 (Integer/parseInt arg1) 8090)
           server (ws/ws-server :port port
-                  :onopen (fn [this conn] (println this conn)))]
+                  :onopen (carefully #'server-onopen)
+                  :onclose (carefully #'server-onclose)
+                  :onmessage #'server-onmessage)]
       (prn "server" :port port)
       (.start server)
+      (send-off publisher publish)
       server)
 
     "client"