Add everything

.config

@@ -1,6 +1,6 @@
 author-name = Aldwin Vlasblom
 repo-owner = fluture-js
-repo-name = fluture-project
+repo-name = fluture-observe
 source-files = index.js
 module-type = esm
 opening-delimiter = ```js

index.js

@@ -1 +1,145 @@
-//. # Fluture Project
+//# Fluture Observe
+//.
+//. Consume a [Fluture][] Future, observing changes to its state as the
+//. consumption is happening.
+//.
+//. ## Usage
+//.
+//. ```js
+//. import {observe, cata} from 'fluture-observe/index.js';
+//.
+//. const consume = observe (cata ({
+//.   Idle: () => {
+//.     console.log ('Computation is idle.');
+//.   },
+//.   Pending: cancel => {
+//.     console.log ('Computation is pending. Send SIGINT to cancel it');
+//.     process.once ('SIGINT', cancel);
+//.   },
+//.   Canceled: future => {
+//.     console.log ('Computation was canceled. Send SIGINT to restart it');
+//.     process.once ('SIGINT', () => consume (future));
+//.   },
+//.   Crashed: error => {
+//.     console.error ('I am sorry to inform you:', error);
+//.     process.exit (1);
+//.   },
+//.   Rejected: reason => {
+//.     console.log ('The computation rejected with reason', reason);
+//.   },
+//.   Resolved: value => {
+//.     console.log ('The computation resolved with value', value);
+//.   }
+//. }));
+//. ```
+import daggy from 'daggy';
+import {forkCatch, isFuture} from 'fluture/index.js';
+
+//# Computation :: Type
+//.
+//. A [Daggy][] tagged union type representing the state of the consumption of
+//. a Future. The `Cancel` and `Future` types below are imported
+//. [types from Fluture][].
+//.
+//. ```hs
+//. data Computation a b = Idle
+//.                      | Pending Cancel
+//.                      | Cancelled (Future a b)
+//.                      | Crashed Error
+//.                      | Rejected a
+//.                      | Resolved b
+//. ```
+//.
+//. Constructor details are documented below.
+export var Computation = daggy.taggedSum ('Computation', {
+  Idle: [],
+  Pending: ['cancel'],
+  Canceled: ['future'],
+  Crashed: ['exception'],
+  Rejected: ['reason'],
+  Resolved: ['value']
+});
+
+//# Idle :: Computation a b
+//.
+//. Represents a not running computation.
+export const Idle = Computation.Idle;
+
+//# Pending :: Cancel -> Computation a b
+//.
+//. Represents a running computation which can be cancelled.
+export const Pending = Computation.Pending;
+
+//# Canceled :: Future a b -> Computation a b
+//.
+//. Represents a computation that was cancelled and can be restarted.
+export const Canceled = Computation.Canceled;
+
+//# Crashed :: Error -> Computation a b
+//.
+//. Represents a computation which encountered an exception while running.
+export const Crashed = Computation.Crashed;
+
+//# Rejected :: a -> Computation a b
+//.
+//. Represents a computation which rejected with a reason.
+export const Rejected = Computation.Rejected;
+
+//# Resolved :: a -> Computation a b
+//.
+//. Represents a computation which resolved with a value.
+export const Resolved = Computation.Resolved;
+
+//# cata :: { Idle :: () -> c, Pending :: Cancel -> c, Canceled :: Future a b -> c, Crashed :: Error -> c, Rejected :: a -> c, Resolved :: b -> c } -> Computation a b -> c
+//.
+//. [Daggy][]'s catamorphism as a curried function.
+export function cata(cases) {
+  return function(t) {
+    return t.cata (cases);
+  };
+}
+
+//# observe :: (Computation a b -> Any) -> Future a b -> Undefined
+//.
+//. Consume a Future, observing changes to its state. See [usage](#usage).
+export function observe(f) {
+  if (typeof f !== 'function') {
+    throw new TypeError (
+      'observe() expects its first argument to be a Function.'
+    );
+  }
+
+  return function observe(m) {
+    if (!isFuture (m)) {
+      throw new TypeError (
+        'observe() expects its second argument to be a valid Future.'
+      );
+    }
+
+    var settled = false;
+
+    f (Computation.Idle);
+
+    var unsubscribe = m.pipe (forkCatch (function observe$recover(exception) {
+      settled = true;
+      f (Computation.Crashed (exception));
+    }) (function observe$reason(reason) {
+      settled = true;
+      f (Computation.Rejected (reason));
+    }) (function observe$value(value) {
+      settled = true;
+      f (Computation.Resolved (value));
+    }));
+
+    if (settled) return;
+
+    f (Computation.Pending (function observe$cancel() {
+      unsubscribe ();
+      f (Computation.Canceled (m));
+    }));
+  };
+}
+
+//. [Fluture]: https://github.com/fluture-js/Fluture
+//. [Daggy]: https://github.com/fantasyland/daggy
+//. [types from Fluture]: https://github.com/fluture-js/Fluture#types

package.json

@@ -1,8 +1,12 @@
 {
-  "name": "fluture-project",
+  "name": "fluture-observe",
   "version": "0.0.0",
-  "description": "Hopefully something related to Fluture",
-  "tags": ["fluture"],
+  "description": "Consume a Future, observing changes to its state",
+  "tags": [
+    "consume",
+    "fluture",
+    "observe"
+  ],
   "type": "module",
   "main": "index.cjs",
   "module": "index.js",
@@ -16,7 +20,7 @@
   },
   "repository": {
     "type": "git",
-    "url": "git://github.com/fluture-js/fluture-project.git"
+    "url": "git://github.com/fluture-js/fluture-observe.git"
   },
   "files": [
     "/index.cjs",
@@ -27,11 +31,16 @@
   ],
   "author": "Aldwin Vlasblom <[email protected]> (https://github.com/Avaq)",
   "license": "MIT",
-  "dependencies": {},
-  "peerDependencies": {},
+  "dependencies": {
+    "daggy": "^1.4.0"
+  },
+  "peerDependencies": {
+    "fluture": ">=12.2.0 <13.0.0"
+  },
   "devDependencies": {
     "c8": "^7.1.0",
     "codecov": "^3.2.0",
+    "fluture": "^12.2.0",
     "oletus": "^3.0.0",
     "rollup": "^2.0.0",
     "sanctuary-scripts": "^4.0.0"

rollup.config.js

@@ -3,7 +3,7 @@ export default {
   output: {
     format: 'umd',
     file: 'index.cjs',
-    name: 'flutureProject',
+    name: 'flutureObserve',
     interop: false
   }
 };

test/index.js

@@ -1 +1,135 @@
-import '../index.js';
+import * as fo from '../index.js';
+import * as fl from 'fluture/index.js';
+import {isDeepStrictEqual} from 'util';
+import test from 'oletus';
+
+function always(x) {
+  return function(y) {
+    return x;
+  };
+}
+
+function equals(a) {
+  return function(b) {
+    return isDeepStrictEqual (a, b);
+  };
+}
+
+function crash(x) {
+  return fl.Future (function() {
+    throw x;
+  });
+}
+
+function expectSequence(future) {
+  return function expectSequence(sequence) {
+    return new Promise (function(res, rej) {
+      let t;
+      let i = 0;
+      resetTimeout ();
+      future.pipe (fo.observe (c => {
+        if (i === sequence.length) {
+          rej (new Error ('More events emitted than expected: ' + c));
+          return;
+        }
+        if (!sequence[i] (c)) {
+          rej (new Error ('Assertion ' + i + ' failed'));
+          return;
+        }
+        i += 1;
+        if (i === sequence.length) {
+          clearTimeout (t);
+          setTimeout (res, 100);
+        }
+      }));
+      function resetTimeout() {
+        clearTimeout (t);
+        t = setTimeout (rej, 100, new Error ('Not enough events emitted after ' + i));
+      }
+    });
+  };
+}
+
+test ('observe, bad first argument', ({throws}) => {
+  throws (function() {
+    fo.observe (null);
+  }, new TypeError (
+    'observe() expects its first argument to be a Function.'
+  ));
+});
+
+test ('observe, bad second argument', ({throws}) => {
+  throws (function() {
+    fo.observe (always) (null);
+  }, new TypeError (
+    'observe() expects its second argument to be a valid Future.'
+  ));
+});
+
+test ('observe a forever pending Future', () => expectSequence (fl.never) ([
+  fo.Idle.is,
+  fo.Pending.is
+]));
+
+test ('observe and cancel a Future', () => expectSequence (fl.never) ([
+  fo.Idle.is,
+  fo.cata ({
+    Idle: always (false),
+    Pending: cancel => {
+      setImmediate (cancel);
+      return true;
+    },
+    Canceled: always (false),
+    Crashed: always (false),
+    Rejected: always (false),
+    Resolved: always (false)
+  }),
+  equals (fo.Canceled (fl.never))
+]));
+
+test ('observe an immediately crashing Future', () => expectSequence (crash (42)) ([
+  fo.Idle.is,
+  fo.cata ({
+    Idle: always (false),
+    Pending: always (false),
+    Canceled: always (false),
+    Crashed: x => x instanceof Error,
+    Rejected: always (false),
+    Resolved: always (false)
+  })
+]));
+
+test ('observe an eventually crashing Future', () => expectSequence (fl.chain (crash) (fl.after (10) (42))) ([
+  fo.Idle.is,
+  fo.Pending.is,
+  fo.cata ({
+    Idle: always (false),
+    Pending: always (false),
+    Canceled: always (false),
+    Crashed: x => x instanceof Error,
+    Rejected: always (false),
+    Resolved: always (false)
+  })
+]));
+
+test ('observe an immediately rejecting Future', () => expectSequence (fl.reject (42)) ([
+  fo.Idle.is,
+  equals (fo.Rejected (42))
+]));
+
+test ('observe an eventually rejecting Future', () => expectSequence (fl.rejectAfter (10) (42)) ([
+  fo.Idle.is,
+  fo.Pending.is,
+  equals (fo.Rejected (42))
+]));
+
+test ('observe an immediately resolving Future', () => expectSequence (fl.resolve (42)) ([
+  fo.Idle.is,
+  equals (fo.Resolved (42))
+]));
+
+test ('observe an eventually resolving Future', () => expectSequence (fl.after (10) (42)) ([
+  fo.Idle.is,
+  fo.Pending.is,
+  equals (fo.Resolved (42))
+]));