Chapter #4: effect

package.json

@@ -30,12 +30,18 @@
       "import/no-useless-path-segments": "off",
       "unicorn/import-index": "off",
       "unicorn/filename-case": "off",
+      "unicorn/prevent-abbreviations": "off",
       "capitalized-comments": "off",
+      "prefer-destructuring": "off",
+      "no-multi-assign": "off",
+      "prefer-promise-reject-errors": "off",
+      "no-throw-literal": "off",
       "no-return-assign": "off",
       "default-case": "off",
       "no-labels": "off",
       "guard-for-in": "off",
-      "padding-line-between-statements": "off"
+      "padding-line-between-statements": "off",
+      "promise/prefer-await-to-then": "off"
     }
   },
   "devDependencies": {

src/createEffect.js

@@ -0,0 +1,99 @@
+import { createNode } from './createNode.js'
+import { createEvent } from './createEvent.js'
+import { createStore } from './createStore.js'
+import { compute } from './step.js'
+import { launch } from './kernel.js'
+import { watch } from './watch.js'
+import { defer } from './defer.js'
+
+const status = name => ({ status, ...rest }) =>
+  status === name ? rest : undefined
+
+const field = name => object => object[name]
+
+function Payload(params, resolve, reject) {
+  this.params = params
+  this.resolve = resolve
+  this.reject = reject
+}
+
+const onDone = (event, params, resolve) => result => {
+  launch(event, { status: 'done', params, result })
+  resolve(result)
+}
+
+const onFail = (event, params, reject) => error => {
+  launch(event, { status: 'fail', params, error })
+  reject(error)
+}
+
+export const createEffect = ({ handler }) => {
+  const effect = payload => {
+    const deferred = defer()
+    launch(effect, new Payload(payload, deferred.resolve, deferred.reject))
+    return deferred.promise
+  }
+
+  effect.graphite = createNode()
+  effect.watch = watch(effect)
+
+  effect.prepend = fn => {
+    const prepended = createEvent()
+    createNode({
+      from: prepended,
+      seq: [compute(fn)],
+      to: effect,
+    })
+    return prepended
+  }
+
+  effect.use = fn => (handler = fn)
+  effect.use.getCurrent = () => handler
+
+  const anyway = createEvent()
+  const done = anyway.filterMap(status('done'))
+  const fail = anyway.filterMap(status('fail'))
+  const doneData = done.map(field('result'))
+  const failData = fail.map(field('error'))
+
+  effect.finally = anyway
+  effect.done = done
+  effect.fail = fail
+  effect.doneData = doneData
+  effect.failData = failData
+
+  effect.inFlight = createStore(0)
+    .on(effect, x => x + 1)
+    .on(anyway, x => x - 1)
+  effect.pending = effect.inFlight.map(amount => amount > 0)
+
+  effect.graphite.seq.push(
+    compute(data =>
+      data instanceof Payload
+        ? data // we get this data directly
+        : new Payload( // we get this data indirectly through graph
+            data,
+            () => {}, // dumb resolve function
+            () => {} // dumb reject function
+          )
+    ),
+    compute(({ params, resolve, reject }) => {
+      const handleDone = onDone(anyway, params, resolve)
+      const handleFail = onFail(anyway, params, reject)
+      try {
+        const promise = handler(params)
+        if (promise instanceof Promise) {
+          promise.then(handleDone).catch(handleFail)
+        } else {
+          handleDone(promise)
+        }
+      } catch (error) {
+        handleFail(error)
+      }
+      return params
+    })
+  )
+
+  effect.kind = 'effect'
+  return effect
+}

src/defer.js

@@ -0,0 +1,13 @@
+export const defer = () => {
+  const deferred = {}
+
+  deferred.promise = new Promise((resolve, reject) => {
+    deferred.resolve = resolve
+    deferred.reject = reject
+  })
+
+  // we need this to avoid 'unhandled exception' warning
+  deferred.promise.catch(() => {})
+
+  return deferred
+}

src/index.js

@@ -1,3 +1,4 @@
+export { createEffect } from './createEffect.js'
 export { createEvent } from './createEvent.js'
 export { createStore } from './createStore.js'
 export { createApi } from './createApi.js'

src/is.js

@@ -8,3 +8,4 @@ const is = type => any =>
 export const unit = is()
 export const event = is('event')
 export const store = is('store')
+export const effect = is('effect')

src/kernel.js

@@ -2,7 +2,11 @@ import { getGraph } from './getter.js'
 
 const queue = []
 
+let running = false
 const exec = () => {
+  if (running) return
+  running = true
+
   cycle: while (queue.length) {
     let { node, value } = queue.shift()
 
@@ -20,6 +24,8 @@ const exec = () => {
 
     node.next.forEach(node => queue.push({ node, value }))
   }
+
+  running = false
 }
 
 export const launch = (unit, value) => {

test/04_effects.js

@@ -0,0 +1,185 @@
+import ninos from 'ninos'
+import test from 'ava'
+import {
+  createEffect,
+  createEvent,
+  createStore,
+  forward,
+} from '../src/index.js'
+
+const it = ninos(test)
+const sleep = n => new Promise(resolve => setTimeout(resolve, n))
+
+it('#4: Effects should work as expected', async t => {
+  const log = t.context.stub()
+
+  const nplus = createEffect({
+    async handler(n) {
+      return new Promise(resolve => setTimeout(() => resolve(n + 1), 1))
+    },
+  })
+
+  const numbers = createStore([]).on(nplus.doneData, (a, n) => [...a, n])
+
+  numbers.watch(_ => log('numbers', _))
+  nplus.watch(_ => log('nplus', _))
+  nplus.finally.watch(_ => log('nplus.finally', _))
+  nplus.done.watch(_ => log('nplus.done', _))
+  nplus.doneData.watch(_ => log('nplus.doneData', _))
+  nplus.fail.watch(_ => log('nplus.fail', _))
+  nplus.failData.watch(_ => log('nplus.failData', _))
+
+  nplus(1)
+  await sleep(2)
+
+  nplus.use(n => new Promise(resolve => setTimeout(() => resolve(n + 2), 1)))
+
+  nplus(2)
+  await sleep(2)
+
+  nplus.use(
+    n => new Promise((resolve, reject) => setTimeout(() => reject(n + 3), 1))
+  )
+
+  nplus(3)
+  await sleep(2)
+
+  nplus.use(n => n + 4)
+
+  nplus(4)
+  await sleep(2)
+
+  nplus.use(n => {
+    throw n + 5
+  })
+
+  nplus(5)
+  await sleep(2)
+
+  t.deepEqual(
+    log.calls.map(c => c.arguments),
+    [
+      ['numbers', []],
+      ['nplus', 1],
+      ['nplus.finally', { status: 'done', params: 1, result: 2 }],
+      ['nplus.done', { params: 1, result: 2 }],
+      ['nplus.doneData', 2],
+      ['numbers', [2]],
+
+      ['nplus', 2],
+      ['nplus.finally', { status: 'done', params: 2, result: 4 }],
+      ['nplus.done', { params: 2, result: 4 }],
+      ['nplus.doneData', 4],
+      ['numbers', [2, 4]],
+
+      ['nplus', 3],
+      ['nplus.finally', { status: 'fail', params: 3, error: 6 }],
+      ['nplus.fail', { params: 3, error: 6 }],
+      ['nplus.failData', 6],
+
+      ['nplus', 4],
+      ['nplus.finally', { status: 'done', params: 4, result: 8 }],
+      ['nplus.done', { params: 4, result: 8 }],
+      ['nplus.doneData', 8],
+      ['numbers', [2, 4, 8]],
+
+      ['nplus', 5],
+      ['nplus.finally', { status: 'fail', params: 5, error: 10 }],
+      ['nplus.fail', { params: 5, error: 10 }],
+      ['nplus.failData', 10],
+    ]
+  )
+})
+
+it('#4: Effects stores pending and inFlight should work', async t => {
+  const log = t.context.stub()
+
+  const fx = createEffect({
+    async handler(n) {
+      return new Promise(resolve => setTimeout(() => resolve(n), n))
+    },
+  })
+
+  fx.watch(_ => log('fx', _))
+  fx.pending.watch(_ => log('fx.pending', _))
+  fx.inFlight.watch(_ => log('fx.inFlight', _))
+  fx.finally.watch(_ => log('fx.finally', _))
+  fx.done.watch(_ => log('fx.done', _))
+  fx.doneData.watch(_ => log('fx.doneData', _))
+  fx.fail.watch(_ => log('fx.fail', _))
+  fx.failData.watch(_ => log('fx.failData', _))
+
+  fx(11)
+  fx(22)
+  await sleep(30)
+
+  t.deepEqual(
+    log.calls.map(c => c.arguments),
+    [
+      ['fx.pending', false],
+      ['fx.inFlight', 0],
+
+      ['fx', 11],
+      ['fx.inFlight', 1],
+      ['fx.pending', true],
+
+      ['fx', 22],
+      ['fx.inFlight', 2],
+      ['fx.pending', true], // TODO: fix later
+
+      ['fx.finally', { status: 'done', params: 11, result: 11 }],
+      ['fx.done', { params: 11, result: 11 }],
+      ['fx.inFlight', 1],
+      ['fx.doneData', 11],
+      ['fx.pending', true], // TODO: fix later
+
+      ['fx.finally', { status: 'done', params: 22, result: 22 }],
+      ['fx.done', { params: 22, result: 22 }],
+      ['fx.inFlight', 0],
+      ['fx.doneData', 22],
+      ['fx.pending', false],
+    ]
+  )
+})
+
+it('#4: Effects should work with forward', async t => {
+  const log = t.context.stub()
+
+  const event = createEvent()
+  const fx = createEffect({
+    async handler(n) {
+      return new Promise(resolve => setTimeout(() => resolve(n * n), 1))
+    },
+  })
+  forward({ from: event, to: fx })
+
+  event.watch(_ => log('event', _))
+  fx.watch(_ => log('fx', _))
+  fx.done.watch(_ => log('fx.done', _))
+
+  event(5)
+  await sleep(10)
+
+  t.deepEqual(
+    log.calls.map(c => c.arguments),
+    [
+      ['event', 5],
+      ['fx', 5],
+      ['fx.done', { params: 5, result: 25 }],
+    ]
+  )
+})
+
+it('#4: Effects should return Promise', async t => {
+  const fx = createEffect({
+    async handler(n) {
+      return new Promise(resolve => setTimeout(() => resolve(n), 1))
+    },
+  })
+
+  const promise = fx(42)
+  t.true(promise instanceof Promise)
+
+  const result = await promise
+  t.is(result, 42)
+})