- Re-designed actors so that their messages are now functions with access to the actor's state, and to the actor itself.

- When creating channels and promises, the type is now mandatory.
- Minor refactorings.

3 files changed, 95 insertions, 94 deletions

diff --git a/stdlib/test/test/lux/concurrency/actor.lux b/stdlib/test/test/lux/concurrency/actor.lux
index a8f6ed7fb..8ec792baf 100644
--- a/stdlib/test/test/lux/concurrency/actor.lux
+++ b/stdlib/test/test/lux/concurrency/actor.lux
@@ -1,79 +1,80 @@
 (;module:
   lux
   (lux [io #- run]
-       (control monad)
+       (control monad
+                ["ex" exception])
        (data [number]
              text/format
              ["R" result])
-       (concurrency [promise #+ Promise Monad<Promise> "Promise/" Monad<Promise>]
+       (concurrency ["P" promise "P/" Monad<Promise>]
+                    ["T" task]
                     ["&" actor #+ actor:]))
   lux/test)
 
-(actor: Adder
-  Int
+(actor: Counter
+  Nat
 
-  (method: (add! [offset Int])
-    [Int Int]
-    (let [*state*' (i.+ offset *state*)]
-      (wrap (#;Right [*state*' [*state* *state*']]))))
-  
-  (stop:
-    (wrap [])))
+  ((count! state self)
+   Nat
+   (let [state' (n.inc state)]
+     (T;return [state' state'])))
+
+  ([cause state]
+   (P/wrap (log! (if (ex;match? &;Killed cause)
+                   (format "Counter was killed: " (%n state))
+                   cause)))))
 
 (context: "Actors"
-  (let [counter-proc (: (&;Behavior Int (Promise Int))
-                        [(function [self output state]
-                           (let [state' (i.inc state)]
-                             (exec (io;run (promise;resolve state' output))
-                               (Promise/wrap (#;Right state')))))
-                         (function [?error state] (Promise/wrap []))])]
-    ($_ seq
-        (test "Can check where an actor is alive."
-              (let [counter (: (&;Actor Int (Promise Int))
-                               (io;run (&;spawn 0 counter-proc)))]
-                (&;alive? counter)))
-        
-        (test "Can poison/kill actors."
-              (let [counter (: (&;Actor Int (Promise Int))
-                               (io;run (&;spawn 0 counter-proc)))]
-                (and (io;run (&;poison counter))
-                     (not (&;alive? counter)))))
-        
-        (test "Can't poison an already poisoned actor."
-              (let [counter (: (&;Actor Int (Promise Int))
-                               (io;run (&;spawn 0 counter-proc)))]
-                (and (io;run (&;poison counter))
-                     (not (io;run (&;poison counter))))))
-        
-        (do Monad<Promise>
-          [#let [counter (: (&;Actor Int (Promise Int))
-                            (io;run (&;spawn 0 counter-proc)))
-                 output-1 (: (Promise Int) (promise;promise))
-                 output-2 (: (Promise Int) (promise;promise))
-                 output-3 (: (Promise Int) (promise;promise))]
-           _ (&;send output-1 counter)
-           _ (&;send output-2 counter)
-           _ (&;send output-3 counter)
-           =1 output-1
-           =2 output-2
-           =3 output-3]
-          (test "Can send messages to actors."
-                (and (i.= 1 =1)
-                     (i.= 2 =2)
-                     (i.= 3 =3))))
-        
-        (do Monad<Promise>
-          [#let [adder (: Adder
-                          (io;run (&;spawn 0 Adder//new)))]
-           t1 (add! 1 adder)
-           t2 (add! 2 adder)
-           t3 (add! 3 adder)
-           #let [_ (io;run (&;poison adder))]]
-          (test "Can use custom-defined actors."
-                (case [t1 t2 t3]
-                  [[0 1] [1 3] [3 6]]
-                  true
+  ($_ seq
+      (test "Can check where an actor is alive."
+            (io;run (do Monad<IO>
+                      [counter (new-Counter +0)]
+                      (wrap (&;alive? counter)))))
+      
+      (test "Can kill actors."
+            (io;run (do Monad<IO>
+                      [counter (new-Counter +0)
+                       killed? (&;kill counter)]
+                      (wrap (and killed?
+                                 (not (&;alive? counter)))))))
+
+      (test "Can poison actors."
+            (io;run (do Monad<IO>
+                      [counter (new-Counter +0)
+                       poisoned? (&;poison counter)]
+                      (wrap (and poisoned?
+                                 (not (&;alive? counter)))))))
+      
+      (test "Cannot kill an already dead actor."
+            (io;run (do Monad<IO>
+                      [counter (new-Counter +0)
+                       first-time (&;kill counter)
+                       second-time (&;kill counter)]
+                      (wrap (and first-time
+                                 (not second-time))))))
+
+      (test "Cannot poison an already dead actor."
+            (io;run (do Monad<IO>
+                      [counter (new-Counter +0)
+                       first-time (&;kill counter)
+                       second-time (&;poison counter)]
+                      (wrap (and first-time
+                                 (not second-time))))))
+
+      (do P;Monad<Promise>
+        [result (do T;Monad<Task>
+                  [#let [counter (io;run (new-Counter +0))]
+                   output-1 (count! counter)
+                   output-2 (count! counter)
+                   output-3 (count! counter)]
+                  (wrap (and (n.= +1 output-1)
+                             (n.= +2 output-2)
+                             (n.= +3 output-3))))]
+        (test "Can send messages to actors."
+              (case result
+                (#R;Success outcome)
+                outcome
 
-                  _
-                  false)))
-        )))
+                (#R;Error error)
+                false)))
+      ))
diff --git a/stdlib/test/test/lux/concurrency/frp.lux b/stdlib/test/test/lux/concurrency/frp.lux
index 2d9a45167..3447a55b2 100644
--- a/stdlib/test/test/lux/concurrency/frp.lux
+++ b/stdlib/test/test/lux/concurrency/frp.lux
@@ -4,13 +4,13 @@
        (control monad)
        (data [number]
              text/format)
-       (concurrency [promise #+ Promise Monad<Promise> "Promise/" Monad<Promise>]
+       (concurrency ["P" promise #+ "P/" Monad<Promise>]
                     ["&" frp]))
   lux/test)
 
-(def: (List->Chan values)
+(def: (to-channel values)
   (-> (List Int) (&;Chan Int))
-  (let [_chan (: (&;Chan Int) (&;chan))]
+  (let [_chan (&;chan Int)]
     (io;run (do Monad<IO>
               [_ (mapM @ (function [value] (&;write value _chan))
                        values)
@@ -19,8 +19,8 @@
 
 (context: "FRP"
   ($_ seq
-      (do Monad<Promise>
-        [elems (&;consume (List->Chan (list 0 1 2 3 4 5)))]
+      (do P;Monad<Promise>
+        [elems (&;consume (to-channel (list 0 1 2 3 4 5)))]
         (test "Can consume a chan into a list."
               (case elems
                 (^ (list 0 1 2 3 4 5))
@@ -29,9 +29,9 @@
                 _
                 false)))
 
-      (do Monad<Promise>
-        [elems (&;consume (let [input (List->Chan (list 0 1 2 3 4 5))
-                                output (: (&;Chan Int) (&;chan))]
+      (do P;Monad<Promise>
+        [elems (&;consume (let [input (to-channel (list 0 1 2 3 4 5))
+                                output (&;chan Int)]
                             (exec (&;pipe input output)
                               output)))]
         (test "Can pipe one channel into another."
@@ -42,8 +42,8 @@
                 _
                 false)))
 
-      (do Monad<Promise>
-        [elems (&;consume (&;filter i.even? (List->Chan (list 0 1 2 3 4 5))))]
+      (do P;Monad<Promise>
+        [elems (&;consume (&;filter i.even? (to-channel (list 0 1 2 3 4 5))))]
         (test "Can filter a channel's elements."
               (case elems
                 (^ (list 0 2 4))
@@ -52,9 +52,9 @@
                 _
                 false)))
 
-      (do Monad<Promise>
-        [elems (&;consume (&;merge (list (List->Chan (list 0 1 2 3 4 5))
-                                         (List->Chan (list 0 -1 -2 -3 -4 -5)))))]
+      (do P;Monad<Promise>
+        [elems (&;consume (&;merge (list (to-channel (list 0 1 2 3 4 5))
+                                         (to-channel (list 0 -1 -2 -3 -4 -5)))))]
         (test "Can merge channels."
               (case elems
                 (^ (list 0 1 2 3 4 5 0 -1 -2 -3 -4 -5))
@@ -63,13 +63,13 @@
                 _
                 false)))
       
-      (do Monad<Promise>
-        [output (&;fold (function [base input] (Promise/wrap (i.+ input base))) 0 (List->Chan (list 0 1 2 3 4 5)))]
+      (do P;Monad<Promise>
+        [output (&;fold (function [base input] (P/wrap (i.+ input base))) 0 (to-channel (list 0 1 2 3 4 5)))]
         (test "Can fold over a channel."
               (i.= 15 output)))
 
-      (do Monad<Promise>
-        [elems (&;consume (&;distinct number;Eq<Int> (List->Chan (list 0 0 0 1 2 2 3 3 3 3 4 4 4 5 5))))]
+      (do P;Monad<Promise>
+        [elems (&;consume (&;distinct number;Eq<Int> (to-channel (list 0 0 0 1 2 2 3 3 3 3 4 4 4 5 5))))]
         (test "Can avoid immediate repetition in the channel."
               (case elems
                 (^ (list 0 1 2 3 4 5))
@@ -78,8 +78,8 @@
                 _
                 false)))
 
-      (do Monad<Promise>
-        [elems (&;consume (&;once (:: promise;Monad<Promise> wrap 12345)))]
+      (do P;Monad<Promise>
+        [elems (&;consume (&;once (:: @ wrap 12345)))]
         (test "Can convert a promise into a single-value channel."
               (case elems
                 (^ (list 12345))
@@ -88,8 +88,8 @@
                 _
                 false)))
 
-      (do Monad<Promise>
-        [elems (&;consume (:: &;Functor<Chan> map i.inc (List->Chan (list 0 1 2 3 4 5))))]
+      (do P;Monad<Promise>
+        [elems (&;consume (:: &;Functor<Chan> map i.inc (to-channel (list 0 1 2 3 4 5))))]
         (test "Functor goes over every element in a channel."
               (case elems
                 (^ (list 1 2 3 4 5 6))
@@ -98,7 +98,7 @@
                 _
                 false)))
 
-      (do Monad<Promise>
+      (do P;Monad<Promise>
         [elems (&;consume (let [(^open) &;Applicative<Chan>]
                             (apply (wrap i.inc) (wrap 12345))))]
         (test "Applicative works over all channel values."
@@ -109,7 +109,7 @@
                 _
                 false)))
 
-      (do Monad<Promise>
+      (do P;Monad<Promise>
         [elems (&;consume (do &;Monad<Chan>
                             [f (wrap i.inc)
                              a (wrap 12345)]
diff --git a/stdlib/test/test/lux/concurrency/promise.lux b/stdlib/test/test/lux/concurrency/promise.lux
index 30802085b..305cfe0f9 100644
--- a/stdlib/test/test/lux/concurrency/promise.lux
+++ b/stdlib/test/test/lux/concurrency/promise.lux
@@ -5,7 +5,7 @@
                 pipe)
        (data [number]
              text/format)
-       (concurrency ["&" promise])
+       (concurrency ["&" promise "&/" Monad<Promise>])
        ["R" math/random])
   lux/test)
 
@@ -49,14 +49,14 @@
               (and ?left (not ?right))))
 
       (test "Can poll a promise for its value."
-            (and (|> (&;poll (:: &;Monad<Promise> wrap true))
+            (and (|> (&;poll (&/wrap true))
                      (case> (#;Some true) true _ false))
                  (|> (&;poll (&;delay +200 true))
                      (case> #;None true _ false))))
 
-      (test "Cant re-resolve a resolved promise."
-            (and (not (io;run (&;resolve false (:: &;Monad<Promise> wrap true))))
-                 (io;run (&;resolve true (: (&;Promise Bool) (&;promise))))))
+      (test "Cannot re-resolve a resolved promise."
+            (and (not (io;run (&;resolve false (&/wrap true))))
+                 (io;run (&;resolve true (&;promise Bool)))))
 
       (do &;Monad<Promise>
         [?none (&;time-out +100 (&;delay +200 true))