Applicative instance for Thread

Author: natefaubion

src/Control/Monad/Aff.js

@@ -2,6 +2,9 @@
 /* jshint -W083, -W098 */
 "use strict";
 
+// A unique value for empty.
+var EMPTY = {};
+
 /*
 
 An awkward approximation. We elide evidence we would otherwise need in PS for
@@ -90,47 +93,37 @@ exports.bracket = function (acquire) {
   };
 };
 
-exports.mapThread = function () {
-  var EMPTY = {};
-
-  return function (f) {
-    return function (thread) {
-      var value = EMPTY;
-
-      function force() {
-        if (value === EMPTY) {
-          return new Aff(BIND, thread.join, function (result) {
-            value = new Aff(PURE, f(result));
-            return value;
-          });
-        } else {
-          return value;
-        }
-      }
-
-      return {
-        kill: thread.kill,
-        join: new Aff(BIND, new Aff(PURE, void 0), force)
-      };
-    };
-  };
-}();
+exports.memoAff = function (aff) {
+  var value = EMPTY;
+  return new Aff(BIND, new Aff(PURE, void 0), function () {
+    if (value === EMPTY) {
+      return new Aff(BIND, aff, function (result) {
+        value = new Aff(PURE, result);
+        return value;
+      });
+    } else {
+      return value;
+    }
+  });
+};
 
 exports._delay = function () {
-  var setDelay = function (n, k) {
+  function setDelay(n, k) {
     if (n === 0 && typeof setImmediate !== "undefined") {
       return setImmediate(k);
     } else {
       return setTimeout(k, n);
     }
-  };
-  var clearDelay = function (n, t) {
+  }
+
+  function clearDelay(n, t) {
     if (n === 0 && typeof clearImmediate !== "undefined") {
       return clearImmediate(t);
     } else {
       return clearTimeout(t);
     }
-  };
+  }
+
   return function (right, ms) {
     return new Aff(ASYNC, function (cb) {
       return function () {

src/Control/Monad/Aff.purs

@@ -34,7 +34,7 @@ import Control.Monad.Rec.Class (class MonadRec, Step(..))
 import Control.MonadPlus (class MonadPlus)
 import Control.MonadZero (class MonadZero)
 import Control.Parallel (parSequence_)
-import Control.Parallel.Class (class Parallel)
+import Control.Parallel.Class (class Parallel, parallel, sequential)
 import Control.Plus (class Plus, empty)
 import Data.Either (Either(..), isLeft)
 import Data.Function.Uncurried as Fn
@@ -190,7 +190,25 @@ newtype Thread eff a = Thread
   }
 
 instance functorThread ∷ Functor (Thread eff) where
-  map = mapThread
+  map f (Thread { kill, join }) = Thread
+    { kill
+    , join: memoAff (f <$> join)
+    }
+
+instance applyThread ∷ Apply (Thread eff) where
+  apply t1 t2 = Thread
+    { kill: \err → sequential $ parallel (killThread err t1) *> parallel (killThread err t2)
+    , join: memoAff do
+        f ← joinThread t1
+        a ← joinThread t2
+        pure (f a)
+    }
+
+instance applicativeThread ∷ Applicative (Thread eff) where
+  pure a = Thread
+    { kill: const (pure unit)
+    , join: pure a
+    }
 
 killThread ∷ ∀ eff a. Error → Thread eff a → Aff eff Unit
 killThread e (Thread t) = t.kill e
@@ -242,7 +260,7 @@ foreign import _delay ∷ ∀ a eff. Fn.Fn2 (Unit → Either a Unit) Number (Aff
 foreign import _liftEff ∷ ∀ eff a. Eff eff a → Aff eff a
 foreign import bracket ∷ ∀ eff a b. Aff eff a → (a → Aff eff Unit) → (a → Aff eff b) → Aff eff b
 foreign import makeAff ∷ ∀ eff a. ((Either Error a → Eff eff Unit) → Eff eff (Canceler eff)) → Aff eff a
-foreign import mapThread ∷ ∀ eff a b. (a → b) → Thread eff a → Thread eff b
+foreign import memoAff ∷ ∀ eff a. Aff eff a → Aff eff a
 
 foreign import _launchAff
   ∷ ∀ eff a

test/Test/Main.purs

@@ -347,8 +347,8 @@ test_kill_parallel_alt = assert "kill/parallel/alt" do
   _ ← try $ joinThread t2
   eq "killedfookilledbardone" <$> readRef ref
 
-test_mapThread ∷ ∀ eff. TestAff eff Unit
-test_mapThread = assert "mapThread" do
+test_thread_map ∷ ∀ eff. TestAff eff Unit
+test_thread_map = assert "thread/map" do
   ref ← newRef 0
   let
     mapFn a = runPure do
@@ -364,6 +364,26 @@ test_mapThread = assert "mapThread" do
   n ← readRef ref
   pure (a == 11 && b == 11 && n == 1)
 
+test_thread_apply ∷ ∀ eff. TestAff eff Unit
+test_thread_apply = assert "thread/apply" do
+  ref ← newRef 0
+  let
+    applyFn a b = runPure do
+      unsafeRunRef $ Ref.modifyRef ref (_ + 1)
+      pure (a + b)
+  t1 ← forkAff do
+    delay (Milliseconds 10.0)
+    pure 10
+  t2 ← forkAff do
+    delay (Milliseconds 15.0)
+    pure 12
+  let
+    t3 = applyFn <$> t1 <*> t2
+  a ← joinThread t3
+  b ← joinThread t3
+  n ← readRef ref
+  pure (a == 22 && b == 22 && n == 1)
+
 main ∷ TestEff () Unit
 main = do
   test_pure
@@ -390,4 +410,5 @@ main = do
     test_kill_parallel
     test_parallel_alt
     test_kill_parallel_alt
-    test_mapThread
+    test_thread_map
+    test_thread_apply