(ns cljgui.events.observe)

(defprotocol observer
  (update! [obs arg]
   "Inform the observer obs that arg has happened. This typically
    implies that obs will call a function with arg as its argument."))

(defprotocol observable
  (subscribe! [able obs] "Subscribe observer obs to observable able.")
  (notify! [able arg]
    "For each subscriber of observable able, call update! with arg.")
  (get-subscribers [able]
    "Return all subscribers associated with observable able")
  (clear-subscribers! [able]
    "Remove all subscribers from observable able."))

(defn make-observer
  "Given a function f, return an observable that maps f to arguments passed
   to it via update! calls."
  [f]
  (reify observer (update! [obs arg] (f arg))))

(defn subscribe
  "Return the result of subscribing an anonymous observer, that
   update!s with f, through notify! arguments from observable origin."
  [f origin]
  (do (subscribe! origin (make-observer f)) origin))

(defn- default-subscribe [subscribers _ obs]
  (if (satisfies? observer obs)
      (dosync (alter subscribers conj obs))
      (throw (Exception. "not a valid observer!"))))

(defn make-observable
  "Primitive function for building new observables.  Used as an entry
   point for making generic observables (which can be notified using
   any type)."
  ([subscribef]
   (let [subscribers (ref [])]
     (reify observable
       (subscribe! [able obs] (subscribef subscribers able obs))
       (notify! [able arg] ;(doall
       (doseq [o (deref subscribers)] (update! o arg)))
       (get-subscribers [able] subscribers)
       (clear-subscribers! [able] (dosync (ref-set subscribers []))))))
  ([] (make-observable default-subscribe)))

(defn- bind-observable
  "Bind an observable to a base observable.  We create specialized
   behavior via the subscriptionf function, which allows us to control
   the binding context.  We keep track of the base observable source
   (via reference) so that we can subscribe to observables at any
   level of composition."
  ([subscribef & base]
   (let [b (first base)
	 subscribers (if b (get-subscribers b) (ref []))]
     (reify observable
       (subscribe! [able obs] (subscribef subscribers able obs))
       (notify! [able arg] (doseq [o (deref subscribers)] (update! o arg)))
       (get-subscribers [able] subscribers)
       (clear-subscribers! [able] (dosync (ref-set subscribers []))))))
  ([] (make-observable default-subscribe)))

(defn map-obs
  "Map function f to the stream of observations arriving from origin.
   Returns a new observable."
  [f origin]
  (bind-observable
    (fn [subscribers _ obs]
      (subscribe! origin (make-observer #(update! obs (f %)))))
    origin))

(defn splitmap-obs
  "Useful in conjunction with split and merge.  Takes either a single
   function, or 2 function args, and applies them to relative args in
   a vector of observables, returning a vector of new observables."
  ([fl fr [origin1 origin2]] [(map-obs fl origin1) (map-obs fr origin2)])
  ([f observers] (splitmap-obs f f observers)))

(defn multimap-obs
  "Return a sequence of observers that are the result of applying f
   via map-obs over each of them in turn."
  [f observercoll] (map (partial map-obs f) observercoll))

(defn choose-obs
  "Imported from the F# lib.  They use Some and None due to static typing, which
   are members of the Option type.  We have nil in clojure, so it's used here.
   Might be able to deprecate this, as it's basically a filter."
  [f origin]
  (bind-observable
    (fn [subscribers _ obs]
      (subscribe! origin
	 (make-observer (fn [arg] (if-let [v (f arg)] (update! obs v))))))
    origin))

(defn filter-obs
  "Filter using a single argument function, filterf.  Returns an
   observable that fires for observations where (= (filterf arg) true)"
  [filterf origin]
  (choose-obs (fn [arg] (if (filterf arg) arg nil)) origin))

(defn partition-obs
  "Split the observation into a vector of 2 observables.  Similar to split-with in
   clojure's seq library.  The first or left observable in the vector fires on
   events that passed the filter, while the right fires on everything else."
  [filterf origin]
    [(filter-obs filterf origin) (filter-obs (comp not filterf) origin)])

;; untested
(defn reductions-obs
  "This is called scan in F#, but is akin to reductions over a
   sequence of events.  We maintain some state, inialized by init.
   f is a function of 2 args, state and the next value.  Each reduction
   replaces the old state with the value of the new reduction."
  [f init origin]
    (let [state (ref init)]
      (bind-observable
	(fn [subscribers _ obs]
	  (make-observer
	    (fn [v] (let [init @state
			  result (f init v)]
	       (dosync (alter state result)
			result)))))
	origin)))

;; somewhat tested
(defn merge-obs
  "Merge two observables into a single observable."
  ([origin1 origin2]
    (bind-observable
      (fn [subscribers able obs]
	(let [h1 (subscribe! origin1 (make-observer #(update! obs %)))
	      h2 (subscribe! origin2 (make-observer #(update! obs %)))]))
      origin1))
   ([[origin1 origin2]] (merge-obs origin1 origin2)))

;; untested
(defn multimerge-obs
  "Merge multiple observables into a single observable."
  [observercoll]
  (reduce merge-obs observercoll))

;; somewhat tested
(defn split-obs [f origin]
  [(choose-obs (fn [v] (if (f v) v nil)) origin)
   (choose-obs (fn [v] (if-not (f v) v nil)) origin)])

(comment
  ;; testing.  this is a simple simulation.....
  (defn balance [split & source]
    (let [signal (if (first source) (first source) (make-observable))]
      (->> signal
      (partition-obs #(<= % split))
      (splitmap-obs
	 (fn [l] [:low l])
	 (fn [r] [:high r]))
      (merge-obs)
      (map-obs (fn [[vtype volume]]
	(let [msg (case vtype
		    :low "Volume is loooooow :"
		    :high "Volume is HIGH!!! :")]
	  (str msg volume)))))))

  ;; create a volume thingy....if volume goes over 50 it thinks its loud.
  (def b (->> (balance 50)
	   (subscribe println)))

  ;; feed a stream of "events", i.e. notifications to the observable
  (doseq [v (range 50)] (notify! b (rand-int 100)))
)