channel.js

import test from 'tape';
import { emitter, channel, eventChannel, END, UNDEFINED_INPUT_ERROR } from '../src/internal/channel'
import { buffers } from '../src/internal/buffers'

const eq = x => y => x === y

test('emitter', assert => {
  assert.plan(1);

  const em = emitter()
  const actual = []

  const unsub1 = em.subscribe(e => actual.push(`1:${e}`))
  const unsub2 = em.subscribe(e => actual.push(`2:${e}`))

  em.emit('e1')
  unsub1()
  em.emit('e2')
  unsub2()
  em.emit('e3')

  const expected = ['1:e1', '2:e1', '2:e2']
  assert.deepEqual(actual, expected, 'emitter should notify subscribers')

});

test('Unbuffered channel', assert => {

  let chan = channel(buffers.none())
  let actual = []
  const logger = () => (ac) => actual.push(ac)

  try {
    chan.put(undefined)
  } catch (e) {
    assert.equal(e.message, UNDEFINED_INPUT_ERROR, 'channel should reject undefined messages')
  }

  chan = channel(buffers.none())

  chan.take(logger(), eq(1))
  const cb = logger()
  chan.take(cb, eq(1))

  chan.put(1)
  assert.deepEqual(actual, [1], 'channel must notify takers')

  cb.cancel()
  chan.put(1)
  assert.deepEqual(actual, [1], 'channel must discard cancelled takes')

  actual = []
  chan.take(logger())
  chan.take(logger())
  chan.close()
  assert.deepEqual(actual,  [END, END], 'closing a channel must resolve all takers with END ')

  actual = []
  chan.take(logger())
  assert.deepEqual(actual, [END], 'closed channel must resolve new takers with END')
  chan.put('action-after-end')
  assert.deepEqual(actual, [END], 'channel must reject messages after being closed')

  assert.end()
});

test('buffered channel', assert => {

  const buffer = []
  const spyBuffer = {
    isEmpty: () => !buffer.length,
    put: (it) => buffer.push(it),
    take: () => buffer.shift()
  }

  let chan = channel(spyBuffer)
  let log = []
  const taker = () => (ac) => log.push(ac)

  const state = () => [chan.__closed__, chan.__takers__, buffer, log]

  var t1 = taker()
  chan.take(t1)
  assert.deepEqual(
    state(),
    [
      /* closed? */ false,
      /* takers  */ [t1],
      /* buffer  */ [],
      /* log     */ []
    ],
    'channel must queue pending takers if there are no buffered messages')

    const t2 = taker()
    chan.take(t2)
    chan.put(1)
    assert.deepEqual(
      state(),
      [
        /* closed? */ false,
        /* takers  */ [t2],
        /* buffer  */ [],
        /* log     */ [1]
      ],
      'channel must resolve the oldest penfing taker with a new message')

    chan.put(2)
    chan.put(3)
    chan.put(4)
    //try {
    //  chan.put(5)
    //} catch(err) {
    //  assert.equal(err.message, BUFFER_OVERFLOW)
    //}
    assert.deepEqual(
      state(),
      [
        /* closed? */ false,
        /* takers  */ [],
        /* buffer  */ [3,4],
        /* log     */ [1,2]
      ],
      'channel must buffer new messages if there are no takers')

    chan.take(taker())
    assert.deepEqual(
      state(),
      [
        /* closed? */ false,
        /* takers  */ [],
        /* buffer  */ [4],
        /* log     */ [1,2,3]
      ],
      'channel must resolve new takers if there are buffered messages')

    chan.close()
    assert.deepEqual(
      state(),
      [
        /* closed? */ true,
        /* takers  */ [],
        /* buffer  */ [4],
        /* log     */ [1,2,3]
      ],
      'channel must set closed state to true')

    chan.close()
    assert.deepEqual(
      state(),
      [
        /* closed? */ true,
        /* takers  */ [],
        /* buffer  */ [4],
        /* log     */ [1,2,3]
      ],
      'closing an already closed channel should be noop')

    chan.put('hi')
    chan.put('I said hi')
    assert.deepEqual(
      state(),
      [
        /* closed? */ true,
        /* takers  */ [],
        /* buffer  */ [4],
        /* log     */ [1,2,3]
      ],
      'putting on an already closed channel should be noop')

    chan.take(taker())
    assert.deepEqual(
      state(),
      [
        /* closed? */ true,
        /* takers  */ [],
        /* buffer  */ [],
        /* log     */ [1,2,3,4]
      ],
      'closed channel must resolve new takers with any buffered message')

      chan.take(taker())
      assert.deepEqual(
        state(),
        [
          /* closed? */ true,
          /* takers  */ [],
          /* buffer  */ [],
          /* log     */ [1,2,3,4, END]
        ],
        'closed channel must resolve new takers with END if there are no buffered message')

  assert.end()
});

test('event channel', assert => {

  const em = emitter()
  let chan = eventChannel(em.subscribe)
  let actual = []

  chan.take((ac) => actual.push(ac))
  em.emit('action-1')
  assert.deepEqual(actual, ['action-1'], 'eventChannel must notify takers on a new action')

  em.emit('action-1')
  assert.deepEqual(actual, ['action-1'], 'eventChannel must notify takers only once')

  actual = []
  chan.take((ac) => actual.push(ac), ac => ac === 'action-xxx')
  chan.close()
  assert.deepEqual(actual, [END], 'eventChannel must notify all pending takers on END')

  actual = []
  chan.take((ac) => actual.push(ac), ac => ac === 'action-yyy')
  assert.deepEqual(actual, [END], 'eventChannel must notify all new takers if closed')


  assert.end()
});