State management in ki (http://ki-lang.org)

state.ki.js

ki macro (export $name $val) 
         (js exports.$name = $val)

ki require core

// Ported from https://github.com/clojure/clojurescript/blob/master/src/cljs/clojure/data.cljs

ki (ns diff

  (defn equalityPartition [x]
   (cond
    (isMap x) :map
    (isSet x) :set
    // js array is :atom for now
    (isSequential x) :sequential
    :else :atom))

  (defn atomDiff [a b]
   (if (eq a b) [nil nil a] [a b nil]))

  (defn notEmpty [coll]
   (when (seq coll) coll))

  (defn vectorize [m]
   (when (seq m)
    (reduceKV
     (fn [result k v] (assoc result k v))
     (into [] (repeat (apply Math.max (keys m)) nil))
     m)))

  (defn diffAssociativeKey [a b k]
   (let [va (get a k)
         vb (get b k)
         dab (diff va vb)
         a_ (nth dab 0)
         b_ (nth dab 1)
         ab (nth dab 2)
         inA (hasKey a k)
         inB (hasKey b k)
         same (and inA inB
                (or (not (eq ab nil))
                    (and (eq va nil) (eq vb nil))))]
    [(when (and inA (or (not (eq a_ nil)) (not same))) {k a_})
     (when (and inB (or (not (eq b_ nil)) (not same))) {k b_})
     (when same {k ab})]))

  (defn diffAssociative 
   ([a b]
    (if (isMap a)
     (diffAssociative a b (set (union (keys a) (keys b))))
     (diffAssociative a b (range (Math.max (count a) (count b))))))
   ([a b ks]
    (reduce
     (fn [diff1 diff2]
      (map merge diff1 diff2))
     [nil nil nil]
     (map (partial diffAssociativeKey a b) ks))))

  (defmulti diffSimilar 
   (fn [a b]
    (equalityPartition a)))

  (defmethod diffSimilar :map [a b]
   (diffAssociative a b))

  // TODO: fix ki bug here
  (defmethod diffSimilar :set [a b]
   (do 
    [(notEmpty (difference a b))
     (notEmpty (difference b a))
     (notEmpty (intersection a b))]))
  
  (defmethod diffSimilar :sequential [a b]
   (into []
    (map vectorize
     (diffAssociative 
      (if (isVector a) a (into [] a))
      (if (isVector b) b (into [] b))))))

  (defmethod diffSimilar :atom [a b]
   (atomDiff a b))

  (defn diff [a b]
   (if (equals a b)
    [nil nil a]
    (if (eq (equalityPartition a) (equalityPartition b))
     (diffSimilar a b)
     (atomDiff a b))))

  //(export diff diff)
);

ki (ns state

  (def looping (atom false))

  (def ouid (atom 0))
  (def observers (atom []))

  // TODO: possible improvements
  // * only keep versions that are referenced to in the cleanup
  // * discard versions that will not be referenced to in swapState
  (def states (atom {:map {0 {:rev 0}} :lastrev 0}))

  (defn getState [rev]
   (let [rev (if (eq rev :lastrev)
              (get (deref states) :lastrev)
              rev)]
    (getIn (deref states) [:map rev])))

  (defn diffStates [rev1 rev2]
   (diff/diff (dissoc (getState rev1) :rev)
              (dissoc (getState rev2) :rev)))

  (defn swapState [f]
   (swap states 
    (fn [states] 
     (let [lastRev (get states :lastrev)
           lastState (getIn states [:map lastRev])
           newRev (inc lastRev)
           newState (threadf (f lastState) (assoc :rev newRev))]
      (threadf states
       (updateIn [:map] assoc newRev newState)
       (assoc :lastrev newRev))))))

  (defn ensureK [k]
   (if (js k[0] == '#')
    (parseInt (k.substr 1))
    (keyword k)))

  (defn ensureKs [ks]
   (let [ks (toClj ks)
         ks (if (isSequential ks) ks [ks])]
    (reduce 
     (fn [out k] 
      (if (eq (typeof k) 'string') 
       (concat out
        (threadl 
         (k.split '/')
         (remove (fn [el] (eq el '')))
         (map ensureK)))
       (concat out [k])))
     [] 
     ks)))

  (defn get_
   ([ks]
    (get_ ks :lastrev))
   ([ks rev]
    (toJs (getIn (getState rev) (ensureKs ks)))))

  (defn set_ 
   ([ks val]
    (set_ ks val true))
   ([ks val to_clj]
    (threadf
     (swapState 
      (fn [state] 
       (assocIn state (ensureKs ks) (if to_clj (toClj val true) val))))
     (get :lastrev))))

  (defn remove_ [ks]
   (threadf
    (swapState 
     (fn [state] 
      (let [ks (into [] (ensureKs ks))
            dissoc_mv
            (fn [s k]
             (if (isVector s)
              (into [] (concat (subvec s 0 k) (subvec s (inc k))))
              (dissoc s k)))]
       (if (isEmpty (pop ks))
        (dissoc_mv state (peek ks))
        (updateIn state (pop ks) (fn [s] (dissoc_mv s (peek ks))))))))
    (get :lastrev)))

  (defn peek_
   ([ks]
    (peek_ ks :lastrev))
   ([ks rev]
    (toJs (peek (getIn (getState rev) (ensureKs ks))))))

  (defn pop_ [ks]
   (threadf
    (swapState 
     (fn [state] 
      (updateIn state (ensureKs ks) pop)))
    (get :lastrev)))

  (defn push_ 
   ([ks val]
    (push_ ks val true))
   ([ks val to_clj]
    (threadf
     (swapState
      (fn [state]
       (let [conj2 (fn [coll el] (ifNot coll [el] (conj coll el)))]
        (updateIn state (ensureKs ks) conj2 (if to_clj (toClj val true) val)))))
     (get :lastrev))))

  (defn count_ 
   ([ks]
    (count_ ks :lastrev))
   ([ks rev]
    (count (getIn (getState rev) (ensureKs ks)))))
 
  (defn observe [ks callback]
   (let [uid (swap ouid inc)]
    (swap observers 
     conj {:path (ensureKs ks) :callback callback :rev 0 :uid uid})
    uid))

  (defn unobserve [uid]
   (swap observers
    (fn [obss]
     (remove (fn [obs] (eq (get obs :uid) uid)) obss)))
   nil)

  (defn loop [msec]
    // TODO: do nothing if nrev is not changing
    // Keep track of :lastrev (as function argument?)
    // and avoid doing anything if :lastrev is the same
    // *and* there's only one entry in the state map (no history)
    // (an observer could have deferred)
   (swap observers
    (fn [obss]
     (loop [obss obss
            out []
            diffCache {}]
      (if (isEmpty obss)
       out
       (let [obs  (first obss)
             orev (get obs :rev)
             nrev (get (getState :lastrev) :rev)]
        (if (eq orev nrev)
         (recur (rest obss) (conj out obs) diffCache)
         (let [diff  (if (get diffCache :rev) (get diffCache :rev) (diffStates orev :lastrev))
               minus (getIn (nth diff 0) (get obs :path))
               plus  (getIn (nth diff 1) (get obs :path))]
          (recur
           (rest obss)
           (conj out 
            (if (or minus plus)
             (if (neq ((get obs :callback) (toJs plus) (toJs minus) nrev orev) false)
              (assoc obs :rev nrev) 
              obs)
             (assoc obs :rev nrev)))
           (if (get diffCache :rev) 
            diffCache 
            (assoc diffCache orev diff))))))))))
   (whenNot (isEmpty (deref observers))
    (let [minRev (reduce Math.min (map :rev (deref observers)))]
     (swap states updateIn [:map] 
      (fn [m] (into {} (filter (fn [kv] (geq (first kv) minRev)) m))))))
   (when (deref looping)
    (setTimeout (partial loop msec) msec)))

  (defn start 
   ([]
    (start 16))
   ([msec]
    (reset looping true)
    (loop msec)))

  (defn stop []
   (reset looping false))

  (export get get_)
  (export set set_)
  (export peek peek_)
  (export push push_)
  (export pop pop_)
  (export remove remove_)
  (export count count_)
  (export observe observe)
  (export unobserve unobserve)
  (export start start)
  (export stop stop)
);

test_state.js

var state = require('./state');

var oid = state.observe("foo/bar", function(plus, minus) { console.log('OBSERVED: ' + minus + ' -> ' + plus); } );

state.start();

for (var i=0; i<1000; i++) {
  state.setval("foo/bar",i);
}

state.setval("foo/bar",1234);
state.setval("foo/bar",1235);

setTimeout(function() {
  state.setval("foo/bar",0);
  state.setval("foo/bar",1);
  for (var i=0; i<1000; i++) {
    state.setval("foo/bar",i);
  }
}, 500);

setTimeout(function() { state.unobserve(oid); }, 1000);
setTimeout(function() { state.stop(); }, 2000);

// Expected output:
// OBSERVED: null -> 1235
// OBSERVED: 1235 -> 999