Sync: initial

deps.edn

@@ -37,6 +37,7 @@
  cheshire/cheshire {:mvn/version "5.10.0"}
  com.cognitect/transit-clj {:mvn/version "1.0.324"}
  com.cognitect/transit-cljs {:mvn/version "0.8.269"}
+ org.clojure/core.async {:mvn/version "1.6.681"}
  }
  }
 

docs/sync.md

@@ -0,0 +1,51 @@
+Server assigns id:
+
+- Changes when client reconnects
+
+Client assigns id:
+
+- What to do with client before :catch-up message?
+- Where to store send-fn?
+- Can track how far each client is. Why?
+
+
+```
+tx :: {:tx-data [...]
+ :tx-id <any>
+ :server-idx <long>}
+```
+
+# Client connects to a server
+
+```
+SND {:message :catching-up
+ :patterns [<pattern> ...]
+ :server-idx <long>?}
+
+RCV {:message :catched-up
+ :snapshot <serializable db>
+ :server-idx <long>}
+
+or
+
+RCV {:message :catched-up
+ :txs [<tx> ...]}
+```
+
+# Client makes a transaction
+
+```
+SND {:message :transacting
+ :server-idx server-idx
+ :txs [{:tx-data ...
+ :tx-id ...} ...]}
+```
+
+# Server broadcasts a transaction
+
+```
+RCV {:message :transacted
+ :tx-data ...
+ :tx-id ...
+ :server-idx ...} ...]}
+```

project.clj

@@ -85,7 +85,8 @@
  :test {:dependencies [[metosin/jsonista "0.3.3"]
  [cheshire "5.10.0"]
  [com.cognitect/transit-clj "1.0.324"]
- [com.cognitect/transit-cljs "0.8.269"]]}
+ [com.cognitect/transit-cljs "0.8.269"]
+ [org.clojure/core.async "1.6.681"]]}
  :bench {:dependencies [[criterium "0.4.6"]
  [metosin/jsonista "0.3.3"]
  [com.clojure-goes-fast/clj-async-profiler "0.5.1"]]}

src/datascript/db.cljc

@@ -1779,3 +1779,7 @@
  :else
  (raise "Bad entity type at " entity ", expected map or vector"
  {:error :transact/syntax, :tx-data entity})))))
+
+(defn tx-from-datoms [datoms]
+ (mapv #(vector (if (:added %) :db/add :db/retract) (:e %) (:a %) (:v %)) datoms))
+

src/datascript/sync/client.cljc

@@ -0,0 +1,73 @@
+(ns datascript.sync.client
+ (:require
+ [datascript.conn :as conn]
+ [datascript.db :as db]
+ [datascript.serialize :as serialize]))
+
+(defn client-id []
+ (long (* (rand) 9007199254740991)))
+
+(def *last-tx-id
+ (atom 0))
+
+(defn new-tx-id [client-id]
+ [client-id (swap! *last-tx-id inc)])
+
+(defn on-tx [conn report]
+ (when-not (:server? (:tx-meta report))
+ (let [{:keys [client-id send-fn server-idx pending]} (meta conn)
+ tx {:tx-data (db/tx-from-datoms (:tx-data report))
+ :tx-id (new-tx-id client-id)}]
+ (send-fn
+ {:message :transacting
+ :server-idx @server-idx
+ :txs [tx]})
+ (swap! pending conj tx))))
+
+(defn create-conn [patterns send-fn]
+ (let [res (atom nil :meta
+ {:client-id (client-id)
+ :server-db (atom nil)
+ :pending (atom #?(:clj clojure.lang.PersistentQueue/EMPTY
+ :cljs cljs.core.PersistentQueue.EMPTY))
+ :server-idx (atom nil)
+ :send-fn send-fn
+ :listeners (atom {})})]
+ (send-fn {:message :catching-up})
+ res))
+
+(defn server-message [conn body]
+ (case (:message body)
+ :catched-up
+ (let [{:keys [snapshot server-idx]} body
+ db (serialize/from-serializable snapshot)]
+ (reset! conn db)
+ (reset! (:server-db (meta conn)) db)
+ (reset! (:server-idx (meta conn)) server-idx)
+ (conn/listen! conn :sync #(on-tx conn %)))
+
+ :transacted
+ (let [{:keys [tx-data tx-id server-idx]} body
+ {*server-db :server-db
+ *server-idx :server-idx
+ *pending :pending
+ *listeners :listeners} (meta conn)
+ report (conn/with @*server-db tx-data {:server? true})
+ server-db' (:db-after report)]
+ (reset! *server-db server-db')
+ (reset! *server-idx server-idx)
+ (if (= tx-id (:tx-id (peek @*pending)))
+ (swap! *pending pop)
+ (do
+ (reset! conn (reduce conn/db-with server-db' @*pending))
+ (doseq [[_ callback] @*listeners]
+ (callback report))))))
+ nil)
+
+(defn server-disconnected [conn]
+ ;; TODO impl me
+ )
+
+(defn server-connected [conn]
+ ;; TODO impl me
+ )

src/datascript/sync/server.cljc

@@ -0,0 +1,54 @@
+(ns datascript.sync.server
+ (:require
+ [datascript.conn :as conn]
+ [datascript.db :as db]
+ [datascript.serialize :as serialize]))
+
+(defn- client [conn channel]
+ (get @(:clients (meta conn)) channel))
+
+(defn on-tx [conn report]
+ ;; TODO filter what to send where
+ (let [msg {:message :transacted
+ :tx-data (db/tx-from-datoms (:tx-data report))
+ :tx-id (:tx-id (:tx-meta report))
+ :server-idx (:db/current-tx (:tempids report))}]
+ (doseq [[channel {:keys [status send-fn]}] @(:clients (meta conn))
+ ; :let [_ (prn "broadcasting to" channel status)]
+ :when (= :active status)]
+ (send-fn channel msg))))
+
+(defn client-connected [conn channel send-fn]
+ (let [*clients (:clients (meta conn))
+ clients' (swap! *clients assoc channel
+ {:status :connected
+ :send-fn send-fn})]
+ (when (= 1 (count clients'))
+ (conn/listen! conn :sync #(on-tx conn %)))
+ nil))
+
+(defn client-message [conn channel body]
+ (case (:message body)
+ :catching-up
+ (let [{:keys [patterns server-idx]} body ;; TODO delta from server-idx
+ {:keys [send-fn]} (client conn channel)
+ db @conn]
+ (send-fn channel
+ {:message :catched-up
+ :snapshot (serialize/serializable db) ;; TODO patterns
+ :server-idx (:max-tx db)})
+ ;; TODO race - external txs between (:max-tx db) and after :status :active
+ (swap! (:clients (meta conn)) update channel assoc :status :active))
+
+ :transacting
+ (doseq [{:keys [tx-data tx-id]} (:txs body)]
+ ;; TODO handle exception here
+ (conn/transact! conn tx-data {:tx-id tx-id})))
+ nil)
+
+(defn client-disconnected [conn channel]
+ (let [*clients (:clients (meta conn))
+ clients' (swap! *clients dissoc channel)]
+ (when (= 0 (count clients'))
+ (conn/unlisten! conn :sync))
+ nil))

test/datascript/test.cljc

@@ -36,6 +36,7 @@
  datascript.test.query-v3
  datascript.test.serialize
  #?(:clj datascript.test.storage)
+ #?(:clj datascript.test.sync)
  datascript.test.transact
  datascript.test.tuples
  datascript.test.validation

test/datascript/test/core.cljc

@@ -97,6 +97,14 @@
  #?(:clj (transit-read (.getBytes ^String s "UTF-8") :json)
  :cljs (transit-read s :json)))
 
+#?(:clj
+ (def lock (Object.)))
+
+(defn log [& args]
+ #?(:clj (locking lock
+ (apply println args))
+ :cljs (apply println args)))
+
 ;; Core tests
 
 (deftest test-protocols

test/datascript/test/sync.clj

@@ -0,0 +1,87 @@
+(ns datascript.test.sync
+ (:require
+ [clojure.core.async :as async :refer [>! <! go go-loop]]
+ [clojure.edn :as edn]
+ [clojure.test :as t :refer [is are deftest testing]]
+ [datascript.core :as d]
+ [datascript.sync.server :as server]
+ [datascript.sync.client :as client]
+ [datascript.test.core :as tdc]))
+
+(defn wait-on [*atom cond]
+ (let [*p (promise)]
+ (add-watch *atom :wait
+ (fn [_ _ _ new]
+ (when (cond new)
+ (deliver *p true))))
+ (when (cond @*atom)
+ (deliver *p true))
+ (let [p (deref *p 100 :timeout)]
+ (remove-watch *atom :wait)
+ (when (= :timeout p)
+ (throw (ex-info "Timeout" {}))))))
+
+(def freeze
+ ; identity
+ tdc/transit-write-str
+ )
+
+(def thaw
+ ; identity
+ tdc/transit-read-str
+ )
+
+(defn setup []
+ (let [server (d/create-conn)
+ ch (async/chan 10)
+ ch1 (async/chan 10)
+ ch2 (async/chan 10)
+ c1 (client/create-conn nil #(go (>! ch [:c1 (freeze %)])))
+ c2 (client/create-conn nil #(go (>! ch [:c2 (freeze %)])))]
+ (server/client-connected server :c1 (fn [_ msg] (go (>! ch1 (freeze msg)))))
+ (server/client-connected server :c2 (fn [_ msg] (go (>! ch2 (freeze msg)))))
+ (go-loop []
+ (when-some [msg (<! ch1)]
+ (let [msg (thaw msg)]
+ (tdc/log "C1 RCV" msg)
+ (client/server-message c1 msg)
+ (recur))))
+ (go-loop []
+ (when-some [msg (<! ch2)]
+ (let [msg (thaw msg)]
+ (tdc/log "C2 RCV" msg)
+ (client/server-message c2 msg)
+ (recur))))
+ (go-loop []
+ (when-some [[id msg] (<! ch)]
+ (let [msg (thaw msg)]
+ (tdc/log "SRV RCV" id msg)
+ (server/client-message server id msg)
+ (recur))))
+ (wait-on c1 some?)
+ (wait-on c2 some?)
+ {:server server
+ :c1 c1
+ :c2 c2}))
+
+(defn wait-all [{:keys [server c1 c2]}]
+ (wait-on (:pending (meta c1)) empty?)
+ (wait-on (:pending (meta c2)) empty?)
+ (wait-on (:server-idx (meta c1)) #(= % (:max-tx @server)))
+ (wait-on (:server-idx (meta c2)) #(= % (:max-tx @server))))
+
+(deftest test-sync
+ (let [{:keys [server c1 c2] :as setup} (setup)]
+ (d/transact! c1 [[:db/add 1 :name "Ivan"]])
+ (wait-all setup)
+ (is (= #{[1 :name "Ivan"]}
+ (tdc/all-datoms @c1)))
+ (is (= #{[1 :name "Ivan"]}
+ (tdc/all-datoms @c2)))
+ (is (= #{[1 :name "Ivan"]}
+ (tdc/all-datoms @server)))))
+
+
+; (t/test-ns *ns*)
+; (t/run-test-var #'test-conn)
+