Rama + Electric

rama_electric.clj

;; ## General helpers
(ns app.helpers.rama
  (:require
    [missionary.core :as m]
    [taoensso.timbre :as timbre])
  (:import
    [com.rpl.rama Depot PState Path ProxyState$Callback RamaModule]
    [com.rpl.rama.test InProcessCluster LaunchConfig]
    [hyperfiddle.electric Failure Pending]))


(defn pstate
  ^PState [^InProcessCluster cluster module ^String name]
  (.clusterPState cluster (.getName module) name))


(defn depot
  ^Depot [^InProcessCluster cluster module ^String name]
  (.clusterDepot cluster (.getName module) name))


(deftype ProxyCallback
  [f]

  ProxyState$Callback

  (change
    [v new _diff _old]
    (when new (f new))))


(defn subscribe
  [^PState pstate ^Path path]
  (->> (m/observe (fn [!]
                    (timbre/debug "Starting subscription")
                    (let [p (.proxyAsync pstate path (->ProxyCallback !))]
                      (fn []
                        (timbre/debug "Closing subscription")
                        (.close @p)))))
    (m/reductions {} (Failure. (Pending.)))
    (m/relieve {})))


;; ## Module
(ns app.modules.wordCount
  (:require
    [app.helpers.rama :as rama])
  (:import
    [com.rpl.rama Agg CompoundAgg Depot PState Path RamaModule]))


(deftype SimpleWordCountModule
  []

  RamaModule

  (define
    [_ setup topologies]
    (.declareDepot setup "*words" (Depot/random))
    (let [s      (.stream topologies "s")
          source (.source s "*words")]
      (.pstate s "$$word-counts" (PState/mapSchema String Long))
      (.pstate s "$$words" (PState/mapSchema String (PState/setSchema String)))
      (-> source
        (.out (into-array String ["*word"]))
        (.anchor "root")
        (.hashPartition "*word")
        (.compoundAgg "$$word-counts" (CompoundAgg/map (into-array Object ["*word" (Agg/count)])))
        (.hook "root")
        (.globalPartition)
        (.localTransform "$$words" (-> (Path/key (into-array String ["all-words"]))
                                     (.nullToSet)
                                     (.voidSetElem)
                                     (.termVal "*word")))))))


;; ## Electric
(ns ^{:clj-kondo/ignore true
      :dev/always       true}
  app.client.root
  (:require
    #?@(:clj  [[app.modules.wordCount :refer [->SimpleWordCountModule]]
               [app.helpers.rama :as rama]])
    [hyperfiddle.electric-dom2 :as dom]
    [clojure.string :as str]
    [hyperfiddle.electric :as e])
  #?(:clj (:import
            [com.rpl.rama Path]
            [com.rpl.rama.test LaunchConfig InProcessCluster]
            [app.modules.wordCount SimpleWordCountModule])))

#?(:clj (defonce cluster
          (let [c (InProcessCluster/create)]
            (.launchModule c (->SimpleWordCountModule) (LaunchConfig. 1 1))
            c)))

(e/def pstate-words #?(:clj (rama/pstate cluster SimpleWordCountModule "$$words")))
(e/def pstate-word-counts #?(:clj (rama/pstate cluster SimpleWordCountModule "$$word-counts")))
(e/def words-depot #?(:clj (rama/depot cluster SimpleWordCountModule "*words")))


#?(:clj (defn <-words
          [pstate]
          (rama/subscribe pstate (Path/key (into-array String ["all-words"])))))


#?(:clj (defn <-word-score
          [pstate word]
          (rama/subscribe pstate (Path/key (into-array String [word])))))


(e/defn Word
  [word score]
  (e/client
    (dom/div (dom/props {:class ["flex" "space-x-2"]})
      (dom/div (dom/text score))
      (dom/div (dom/text word)))))


(e/defn Root
  []
  (e/client
    ;; Scoreboard
    (dom/div (dom/props {:class ["flex" "flex-col" "p-6" "space-y-2"]})
      (dom/div
        (dom/text "Scoreboard")
        (e/server
          (let [word-scores (e/for-by identity [word (new (<-words pstate-words))]
                              [word (new (<-word-score pstate-word-counts word))])]
            (e/for-by first [[word score] (sort-by second > word-scores)]
              (Word. word score)))))
      ;; Input
      (dom/input (dom/props {:class ["border"]})
        (dom/on "keydown"
          (e/fn [e]
            (when (= "Enter" (.-key e))
              (let [value (-> e .-target .-value)]
                (when-not (str/blank? value)
                  (e/server (e/offload #(.append words-depot value))))))))))))