Reverted "lux/control/concurrency/frp" to the previous style of doing FRP.

4 files changed, 264 insertions, 192 deletions

diff --git a/stdlib/source/lux/control/concurrency/frp.lux b/stdlib/source/lux/control/concurrency/frp.lux
index 8db54f28f..5e5178e4b 100644
--- a/stdlib/source/lux/control/concurrency/frp.lux
+++ b/stdlib/source/lux/control/concurrency/frp.lux
@@ -1,132 +1,278 @@
 (.module:
-  [lux #*
+  [lux (#- Source)
    [control
     [functor (#+ Functor)]
     [apply (#+ Apply)]
-    ["." monad (#+ do Monad)]]
-   ["." io (#+ IO io)]
+    ["." monad (#+ do Monad)]
+    [predicate (#+ Predicate)]
+    [equivalence (#+ Equivalence)]]
+   ["." io (#+ IO)]
    [data
+    [maybe ("maybe/." Functor<Maybe>)]
     [collection
      [list ("list/." Monoid<List>)]]]
    [type (#+ :share)
     abstract]]
   [//
-   ["." atom (#+ Atom atom)]
-   ["." promise (#+ Promise)]])
+   ["." atom (#+ Atom)]
+   ["." promise (#+ Promise) ("promise/." Functor<Promise>)]])
 
-(abstract: #export (Channel a)
+(type: #export (Channel a)
   {#.doc "An asynchronous channel to distribute values."}
-  (Atom (List (-> a (IO Any))))
+  (Promise (Maybe [a (Channel a)])))
 
-  (def: #export (channel _)
-    (All [a] (-> Any (Channel a)))
-    (:abstraction (atom (list))))
+(signature: (Resolver a)
+  (: (IO Bit)
+     close)
+  (: (-> a (IO (Maybe (Resolver a))))
+     feed))
 
-  (def: #export (listen listener channel)
-    (All [a] (-> (-> a (IO Any)) (Channel a) (IO Any)))
-    ## TODO: Simplify when possible.
-    (do io.Monad<IO>
-      [_ (atom.update (|>> (#.Cons listener))
-                      (:representation channel))]
-      (wrap [])))
+(def: (resolver resolve)
+  (All [a]
+    (-> (promise.Resolver (Maybe [a (Channel a)]))
+        (Resolver a)))
+  (structure
+   (def: close
+     (resolve #.None))
+   
+   (def: (feed value)
+     (do io.Monad<IO>
+       [#let [[next resolve-next] (:share [a]
+                                          {(promise.Resolver (Maybe [a (Channel a)]))
+                                           resolve}
+                                          {[(Promise (Maybe [a (Channel a)]))
+                                            (promise.Resolver (Maybe [a (Channel a)]))]
+                                           (promise.promise [])})]
+        resolved? (resolve (#.Some [value next]))]
+       (if resolved?
+         (wrap (#.Some (resolver resolve-next)))
+         (wrap #.None))))))
 
-  (def: #export (publish channel value)
-    {#.doc "Publish to a channel."}
-    (All [a] (-> (Channel a) a (IO Any)))
-    (do io.Monad<IO>
-      [listeners (atom.read (:representation channel))]
-      (monad.map @ (function (_ listener) (listener value)) listeners)))
-  )
-
-(def: #export (filter predicate input)
-  (All [a] (-> (-> a Bit) (Channel a) (Channel a)))
-  (let [output (channel [])]
-    (exec (io.run (listen (function (_ value)
-                            (if (predicate value)
-                              (publish output value)
-                              (io [])))
-                          input))
-      output)))
+(abstract: #export (Source a)
+  {}
 
-(def: #export (pipe output input)
-  {#.doc "Copy/pipe the contents of a channel on to another."}
-  (All [a] (-> (Channel a) (Channel a) (IO Any)))
-  (listen (publish output)
-          input))
+  (Atom (Resolver a))
 
-(def: #export (merge inputs)
-  {#.doc "Fuse all the elements in a list of channels by piping them onto a new output channel."}
-  (All [a] (-> (List (Channel a)) (IO (Channel a))))
-  (let [output (channel [])]
+  (def: #export (close source)
+    (All [a] (-> (Source a) (IO Bit)))
     (do io.Monad<IO>
-      [_ (monad.map @ (pipe output) inputs)]
-      (wrap output))))
+      [resolver (atom.read (:representation source))
+       closed? (:: resolver close)]
+      (if closed?
+        ## I closed the source.
+        (wrap true)
+        ## Someone else interacted with the source.
+        (do @
+          [resolver' (atom.read (:representation source))]
+          (if (is? resolver resolver')
+            ## Someone else closed the source.
+            (wrap true)
+            ## Someone else fed the source while I was closing it.
+            (close source))))))
 
-(def: #export (from-promise promise)
-  (All [a] (-> (Promise a) (Channel a)))
-  (let [output (channel [])]
-    (exec (promise.await (publish output) promise)
-      output)))
+  (def: #export (feed source value)
+    (All [a] (-> (Source a) a (IO Bit)))
+    (loop [source (:representation source)]
+      (do io.Monad<IO>
+        [current (atom.read source)
+         ?next (:: current feed value)
+         fed? (case ?next
+                (#.Some next)
+                (atom.compare-and-swap current next source)
 
-(def: #export (poll time action)
-  (All [a] (-> Nat (IO a) (Channel a)))
-  (let [output (channel [])]
-    (exec (io.run
-           (loop [_ []]
-             (do io.Monad<IO>
-               [value action
-                _ (publish output value)]
-               (wrap (promise.await recur (promise.wait time))))))
-      output)))
+                #.None
+                (wrap false))]
+        (if fed?
+          ## I fed the source.
+          (wrap true)
+          ## Someone else interacted with the source.
+          (do @
+            [latter (atom.read source)]
+            (if (is? current latter)
+              ## Someone else closed the source while I was feeding it.
+              (wrap false)
+              ## Someone else fed the source.
+              (recur source)))))))
 
-(def: #export (periodic time)
-  (-> Nat (Channel Any))
-  (let [output (channel [])]
-    (exec (io.run
-           (loop [_ []]
-             (do io.Monad<IO>
-               [_ (publish output [])]
-               (wrap (promise.await recur (promise.wait time))))))
-      output)))
+  (def: source
+    (All [a] (-> (Resolver a) (Source a)))
+    (|>> atom.atom :abstraction))
+  )
 
-(def: #export (iterate f init)
-  (All [a] (-> (-> a (Promise a)) a (Channel a)))
-  (let [output (channel [])]
-    (exec (io.run
-           (loop [zero init]
-             (do io.Monad<IO>
-               [_ (publish output zero)]
-               (wrap (promise.await recur (f zero))))))
-      output)))
+(def: #export (channel _)
+  (All [a] (-> Any [(Channel a) (Source a)]))
+  (let [[promise resolve] (promise.promise [])]
+    [promise (..source (..resolver resolve))]))
+
+(def: #export (listen listener channel)
+  (All [a] (-> (-> a (IO Any)) (Channel a) (IO Any)))
+  (io.io (exec (: (Promise Any)
+                  (loop [channel channel]
+                    (do promise.Monad<Promise>
+                      [cons channel]
+                      (case cons
+                        (#.Some [head tail])
+                        (exec (io.run (listener head))
+                          (recur tail))
+                        
+                        #.None
+                        (wrap [])))))
+           [])))
 
 (structure: #export _ (Functor Channel)
-  (def: (map f input)
-    (let [output (channel [])]
-      (exec (io.run (listen (|>> f (publish output))
-                            input))
-        output))))
+  (def: (map f)
+    (promise/map
+     (maybe/map
+      (function (_ [head tail])
+        [(f head) (map f tail)])))))
 
 (structure: #export _ (Apply Channel)
   (def: functor Functor<Channel>)
 
   (def: (apply ff fa)
-    (let [output (channel [])]
-      (exec (io.run (listen (function (_ f)
-                              (listen (|>> f (publish output))
-                                      fa))
-                            ff))
-        output))))
+    (do promise.Monad<Promise>
+      [cons-f ff
+       cons-a fa]
+      (case [cons-f cons-a]
+        [(#.Some [head-f tail-f]) (#.Some [head-a tail-a])]
+        (wrap (#.Some [(head-f head-a) (apply tail-f tail-a)]))
+
+        _
+        (wrap #.None)))))
 
 (structure: #export _ (Monad Channel)
   (def: functor Functor<Channel>)
 
   (def: (wrap a)
-    (let [output (channel [])]
-      (exec (io.run (publish output a))
-        output)))
+    (promise.resolved (#.Some [a (promise.resolved #.None)])))
 
   (def: (join mma)
-    (let [output (channel [])]
-      (exec (io.run (listen (listen (publish output))
-                            mma))
+    (let [[output source] (channel [])]
+      (exec (io.run (..listen (..listen (..feed source))
+                              mma))
         output))))
+
+(def: #export (filter pass? channel)
+  (All [a] (-> (Predicate a) (Channel a) (Channel a)))
+  (do promise.Monad<Promise>
+    [cons channel]
+    (case cons
+      (#.Some [head tail])
+      (let [tail' (filter pass? tail)]
+        (if (pass? head)
+          (wrap (#.Some [head tail']))
+          tail'))
+      
+      #.None
+      (wrap #.None))))
+
+(def: #export (from-promise promise)
+  (All [a] (-> (Promise a) (Channel a)))
+  (promise/map (function (_ value)
+                 (#.Some [value (promise.resolved #.None)]))
+               promise))
+
+(def: #export (fold f init channel)
+  {#.doc "Asynchronous fold over channels."}
+  (All [a b]
+    (-> (-> b a (Promise a)) a (Channel b)
+        (Promise a)))
+  (do promise.Monad<Promise>
+    [cons channel]
+    (case cons
+      #.None
+      (wrap init)
+      
+      (#.Some [head tail])
+      (do @
+        [init' (f head init)]
+        (fold f init' tail)))))
+
+(def: #export (folds f init channel)
+  {#.doc "A channel of folds."}
+  (All [a b]
+    (-> (-> b a (Promise a)) a (Channel b)
+        (Channel a)))
+  (do promise.Monad<Promise>
+    [cons channel]
+    (case cons
+      #.None
+      (wrap (#.Some [init (wrap #.None)]))
+      
+      (#.Some [head tail])
+      (do @
+        [init' (f head init)]
+        (folds f init' tail)))))
+
+(def: #export (poll milli-seconds action)
+  (All [a] (-> Nat (IO a) (Channel a)))
+  (let [[output source] (channel [])]
+    (exec (io.run (loop [_ []]
+                    (do io.Monad<IO>
+                      [value action
+                       _ (..feed source value)]
+                      (promise.await recur (promise.wait milli-seconds)))))
+      output)))
+
+(def: #export (periodic milli-seconds)
+  (-> Nat (Channel Any))
+  (poll milli-seconds (io.io [])))
+
+(def: #export (iterate f init)
+  (All [a] (-> (-> a (Promise (Maybe a))) a (Channel a)))
+  (do promise.Monad<Promise>
+    [?next (f init)]
+    (case ?next
+      (#.Some next)
+      (wrap (#.Some [init (iterate f next)]))
+      
+      #.None
+      (wrap (#.Some [init (wrap #.None)])))))
+
+(def: (distinct' equivalence previous channel)
+  (All [a] (-> (Equivalence a) a (Channel a) (Channel a)))
+  (do promise.Monad<Promise>
+    [cons channel]
+    (case cons
+      (#.Some [head tail])
+      (if (:: equivalence = previous head)
+        (distinct' equivalence previous tail)
+        (wrap (#.Some [head (distinct' equivalence head tail)])))
+      
+      #.None
+      (wrap #.None))))
+
+(def: #export (distinct equivalence channel)
+  (All [a] (-> (Equivalence a) (Channel a) (Channel a)))
+  (do promise.Monad<Promise>
+    [cons channel]
+    (case cons
+      (#.Some [head tail])
+      (wrap (#.Some [head (distinct' equivalence head tail)]))
+      
+      #.None
+      (wrap #.None))))
+
+(def: #export (consume channel)
+  {#.doc "Reads the entirety of a channel's content and returns it as a list."}
+  (All [a] (-> (Channel a) (Promise (List a))))
+  (do promise.Monad<Promise>
+    [cons channel]
+    (case cons
+      (#.Some [head tail])
+      (:: @ map (|>> (#.Cons head))
+          (consume tail))
+
+      #.None
+      (wrap #.Nil))))
+
+(def: #export (sequential milli-seconds values)
+  (All [a] (-> Nat (List a) (Channel a)))
+  (case values
+    #.Nil
+    (promise.resolved #.None)
+
+    (#.Cons head tail)
+    (promise.resolved (#.Some [head (do promise.Monad<Promise>
+                                      [_ (promise.wait milli-seconds)]
+                                      (sequential milli-seconds tail))]))))
diff --git a/stdlib/source/lux/control/concurrency/promise.lux b/stdlib/source/lux/control/concurrency/promise.lux
index 2530d6080..31d620b64 100644
--- a/stdlib/source/lux/control/concurrency/promise.lux
+++ b/stdlib/source/lux/control/concurrency/promise.lux
@@ -101,8 +101,7 @@
 (structure: #export _ (Monad Promise)
   (def: functor Functor<Promise>)
 
-  (def: (wrap a)
-    (..resolved a))
+  (def: wrap ..resolved)
 
   (def: (join mma)
     (let [[ma resolve] (promise [])]
diff --git a/stdlib/source/lux/control/concurrency/stm.lux b/stdlib/source/lux/control/concurrency/stm.lux
index f54e16baf..31a62413e 100644
--- a/stdlib/source/lux/control/concurrency/stm.lux
+++ b/stdlib/source/lux/control/concurrency/stm.lux
@@ -52,18 +52,14 @@
           (wrap []))
         (write! new-value (:abstraction var)))))
 
-  ## TODO: Remove when possible
-  (def: (helper|follow var)
-    (All [a] (-> (Var a) (frp.Channel a)))
-    (frp.channel []))
   (def: #export (follow target)
     {#.doc "Creates a channel that will receive all changes to the value of the given var."}
     (All [a] (-> (Var a) (IO (frp.Channel a))))
     (do io.Monad<IO>
-      [#let [channel (helper|follow target)
+      [#let [[channel source] (frp.channel [])
              target (:representation target)]
        _ (atom.update (function (_ [value observers])
-                        [value (#.Cons (frp.publish channel) observers)])
+                        [value (#.Cons (frp.feed source) observers)])
                       target)]
       (wrap channel)))
   )
diff --git a/stdlib/test/test/lux/control/concurrency/frp.lux b/stdlib/test/test/lux/control/concurrency/frp.lux
index 04ddd5986..a906ee54b 100644
--- a/stdlib/test/test/lux/control/concurrency/frp.lux
+++ b/stdlib/test/test/lux/control/concurrency/frp.lux
@@ -13,110 +13,41 @@
      ["." list]]]]
   lux/test)
 
-(def: (write! values channel)
-  (All [a] (-> (List a) (Channel a) (IO Any)))
-  (do io.Monad<IO>
-    [_ (monad.map @ (frp.publish channel) values)]
-    (wrap [])))
-
-(def: (read! channel)
-  (All [a] (-> (Channel a) (IO (Atom (List a)))))
-  (do io.Monad<IO>
-    [#let [output (atom (list))]
-     _ (frp.listen (function (_ value)
-                     ## TODO: Simplify when possible.
-                     (do @
-                       [_ (atom.update (|>> (#.Cons value)) output)]
-                       (wrap [])))
-                   channel)]
-    (wrap output)))
-
 (context: "FRP"
   (let [(^open "list/.") (list.Equivalence<List> number.Equivalence<Int>)]
     ($_ seq
         (wrap (do promise.Monad<Promise>
-                [#let [values (list +0 +1 +2 +3 +4 +5)]
-                 output (promise.future
-                         (do io.Monad<IO>
-                           [#let [input (: (Channel Int) (frp.channel []))]
-                            output (read! input)
-                            _ (write! values input)]
-                           (wrap output)))
-                 _ (promise.wait 100)
-                 output (promise.future (atom.read output))]
-                (assert "Can pipe one channel into another."
-                        (list/= values
-                                (list.reverse output)))))
-
-        (wrap (do promise.Monad<Promise>
-                [output (promise.future
-                         (do io.Monad<IO>
-                           [#let [input (: (Channel Int) (frp.channel []))
-                                  elems (frp.filter i/even? input)]
-                            output (read! elems)
-                            _ (write! (list +0 +1 +2 +3 +4 +5) input)]
-                           (wrap output)))
-                 _ (promise.wait 100)
-                 output (promise.future (atom.read output))]
+                [output (|> (list +0 +1 +2 +3 +4 +5)
+                            (frp.sequential 0)
+                            (frp.filter i/even?)
+                            frp.consume)]
                 (assert "Can filter a channel's elements."
-                        (list/= (list +0 +2 +4)
-                                (list.reverse output)))))
-
-        (wrap (do promise.Monad<Promise>
-                [output (promise.future
-                         (do io.Monad<IO>
-                           [#let [left (: (Channel Int) (frp.channel []))
-                                  right (: (Channel Int) (frp.channel []))]
-                            merged (frp.merge (list left right))
-                            output (read! merged)
-                            _ (write! (list +0 +1 +2 +3 +4 +5) left)
-                            _ (write! (list +0 -1 -2 -3 -4 -5) right)]
-                           (wrap output)))
-                 _ (promise.wait 100)
-                 output (promise.future (atom.read output))]
-                (assert "Can merge channels."
-                        (list/= (list +0 +1 +2 +3 +4 +5 +0 -1 -2 -3 -4 -5)
-                                (list.reverse output)))))
+                        (list/= (list +0 +2 +4) output))))
 
         (wrap (do promise.Monad<Promise>
-                [output (promise.future
-                         (do io.Monad<IO>
-                           [#let [inputs (: (Channel Int) (frp.channel []))
-                                  mapped (:: frp.Functor<Channel> map inc inputs)]
-                            output (read! mapped)
-                            _ (write! (list +0 +1 +2 +3 +4 +5) inputs)]
-                           (wrap output)))
-                 _ (promise.wait 100)
-                 output (promise.future (atom.read output))]
+                [output (|> (list +0 +1 +2 +3 +4 +5)
+                            (frp.sequential 0)
+                            (:: frp.Functor<Channel> map inc)
+                            frp.consume)]
                 (assert "Functor goes over every element in a channel."
                         (list/= (list +1 +2 +3 +4 +5 +6)
-                                (list.reverse output)))))
+                                output))))
 
         (wrap (do promise.Monad<Promise>
-                [output (promise.future
-                         (do io.Monad<IO>
-                           [#let [>f< (: (Channel (-> Int Int)) (frp.channel []))
-                                  >a< (: (Channel Int) (frp.channel []))]
-                            output (read! (let [(^open ".") frp.Apply<Channel>]
-                                            (apply >f< >a<)))
-                            _ (write! (list inc) >f<)
-                            _ (write! (list +12345) >a<)]
-                           (wrap output)))
-                 _ (promise.wait 100)
-                 output (promise.future (atom.read output))]
+                [output (frp.consume (:: frp.Apply<Channel> apply
+                                         (frp.sequential 0 (list inc))
+                                         (frp.sequential 0 (list +12345))))]
                 (assert "Apply works over all channel values."
                         (list/= (list +12346)
-                                (list.reverse output)))))
+                                output))))
 
         (wrap (do promise.Monad<Promise>
-                [output (promise.future
-                         (read! (do frp.Monad<Channel>
-                                  [f (frp.from-promise (promise.delay 100 inc))
-                                   a (frp.from-promise (promise.delay 200 +12345))]
-                                  (frp.from-promise (promise.delay 300 (f a))))))
-                 _ (promise.wait 700)
-                 output (promise.future (atom.read output))]
+                [output (frp.consume
+                         (do frp.Monad<Channel>
+                           [f (frp.from-promise (promise/wrap inc))
+                            a (frp.from-promise (promise/wrap +12345))]
+                           (wrap (f a))))]
                 (assert "Valid monad."
                         (list/= (list +12346)
-                                (list.reverse output)))))
+                                output))))
         )))