Here's a mechanism for transforming a discrete input flow into an output flow where the internal process state can be implemented with loop/recur. As an example I provide an implementation of "batching with both a maximum size and a maximum delay" referenced in leonoel/missionary#109.
A common pattern for discrete flows is that we want to create an output flow from an input flow, where:
| ;; Newer version at https://gist.github.com/awwx/6d02e6ea702bdc499bcf847e9b9bb98f | |
| (ns mobserve | |
| (:require | |
| [missionary.core :as m])) | |
| ;; Encapsulate differences between Clojure and ClojureScript | |
| ;; on how IFn and IDeref are implemented. | |
| #?(:clj |
| ;; All our model functions adhere to simple interface, taking any number of arguments, | |
| ;; but always the `:cofx` map as an last argument and producing map of effects (or nothing) | |
| ;; as result: | |
| (defn effects-fn [arg1 arg2 {:keys [db] :as cofx}] | |
| {:db (update db :some-key arg1) | |
| :http-call ...}) | |
| ;; We have both `merge-fx` macro and `merge-effect` functions at our disposal | |
| ;; Whenever we do static merging of effects, we use the former: | |
| (defn final-effects-fn [cofx] |
