medium

    1.2.0 • Public • Published

    Medium

    CSP-style channel library using ES7 async/await keywords.

    Installation

    npm install medium

    First, the requisite naive ping/pong example (ported from Go)

    const { chan, put, close, take, sleep, CLOSED } = require('medium')
    
    const player = async (name, table) => {
      while (true) {
    
        const ball = await take(table)
        if (ball === CLOSED) break
    
        ball.hits++
        console.log(`${name} ${ball.hits}`)
        await sleep(100)
        put(table, ball)
      }
    }
    
    const start = async () => {
    
      const table = chan()
    
      player('ping', table)
      player('pong', table)
    
      put(table, { hits: 0 })
      await sleep(1000)
    
      close(table)
    }
    
    start()

    Channel interactions in a nutshell

    Channels are queues, you can put things onto them and take things off, in a first-in-first-out way. Channels can be closed, after which, they will not receive or deliver values. put and take are both asynchronous actions, and return promises. put promises simply resolve to true if it was able to successfully add its value to the channel, or false if the channel is closed. take promises resolve either to whatever was next in the channel queue, or to the constant CLOSED if the channel is closed. For example:

    const ch1 = chan()
    put(ch1, 1)
    take(ch1).then(console.log)
    // LOGS: 1
    
    take(ch1).then(console.log)
    put(ch1, 2)
    // LOGS: 2
    
    // Notice how it doesn't matter what order the take and put occur in. This is the secret sauce of coordinating asynchronous activites.
    
    take(ch1).then(console.log)
    close(ch1)
    // LOGS: CLOSED
    
    put(ch1, 3).then(console.log)
    // LOGS: false

    The strategy with which a channel handles an excess of puts is implemented as a buffer. The default channel does not allow for any buffered values, so if you put without a waiting take for the value, it will not resolve the put until a corresponding take is added. For example:

    No buffer

    const ch1 = chan()
    put(ch1, 1).then(() => console.log('put 1'))
    put(ch1, 2).then(() => console.log('put 2'))
    take(ch1)
    // LOGS: 'put 1'
    take(ch1)
    // LOGS: 'put 2'

    An example of a different buffer would be a "fixed" buffer, which has N slots for put values to wait for a take. For example:

    Fixed buffer

    const ch = chan()
    const fixedCh = chan(buffers.fixed(2)) // or shortcut with chan(2)
    
    put(ch, 1).then(console.log)
    // LOGS NOTHING
    
    put(fixedCh, 1).then(() => console.log('put 1'))
    // LOGS: put 1
    put(fixedCh, 2).then(() => console.log('put 2'))
    // LOGS: put 2
    put(fixedCh, 3).then(() => console.log('put 3'))
    // LOGS NOTHING
    
    take(fixedCh).then(console.log)
    // LOGS: 1
    // LOGS: put 3

    The other included buffers are, "dropping", which allows N puts, then begins "dropping" them, causing the put to resolve successfully but the value is not added to the channel, and "sliding", which allows N puts, then begins shifting the buffer, dropping the oldest buffered put value and adding the newest to the other end.

    Dropping buffer

    const ch = chan(buffers.dropping(2))
    put(ch, 1)
    put(ch, 2)
    put(ch, 3) // this is dropped
    take(ch).then(console.log)
    // LOGS: 1
    take(ch).then(console.log)
    // LOGS: 2
    take(ch).then(console.log)
    // LOGS NOTHING
    put(ch, 3)
    // LOGS: 3

    Sliding buffer

    const ch = chan(buffers.sliding(2))
    put(ch, 1)
    put(ch, 2)
    put(ch, 3) // this causes the put of 1 to be dropped
    take(ch).then(console.log)
    // LOGS: 2
    take(ch).then(console.log)
    // LOGS: 3

    Building something larger

    Things get much more interesting though when we use async/await to better coordinate our channels.

    import { chan, put, take, sleep, go } from 'medium'
    
    const numbers = chan()
    const oddNumbers = chan()
    
    go(async () => {
      while (true) {
        console.log('an odd number: ', await take(oddNumbers))
      }
    })
    
    go(async () => {
      while (true) {
        let n = await numbers // awaiting a channel is an implied "take"
        if (n % 2 === 1)
          await put(oddNumbers, n)
      }
    })
    
    go(async () => {
      while (true) {
        let randomNum = Math.floor(Math.random() * 100)
        await put(numbers, randomNum)
        await sleep(1000)
      }
    })

    So we have a number being generated every second, and put onto the numbers channel. This is consumed and tested for "oddness", and if it passes, then it is put onto the oddNumbers channel where it is simply console.log'ed.

    What if we want to keep track of the percent odd vs. even? We can put a bit of local state in the process that checks for oddness. However, mutating state sucks, so, we use the function repeat to both act as a while loop and manage state immutably!

    import { chan, put, take, sleep, go, repeat } from 'medium'
    
    const numbers = chan()
    const oddNumbers = chan()
    const stats = chan()
    
    go(async () => {
      while (true) {
        console.log('an odd number: ', await oddNumbers)
      }
    })
    
    go(async () => {
      while (true) {
        console.log('Stats: ', await stats)
      }
    })
    
    go(async () => {
      repeat(async ({ total, odds }) => {
        put(stats, `${odds / total * 100}% odd numbers`)
        
        const n = await numbers
        if (n % 2) {
          put(oddNumbers, n)
          return { total: total + 1, odds: odds + 1 }
        } else {
          return { total: total + 1, odds }
        }
        
      }, { total: 0, odds: 0 })
    })
    
    go(async () => {
      while (true) {
        let randomNum = Math.floor(Math.random() * 100)
        await put(numbers, randomNum)
        await sleep(1000)
      }
    })

    And now we see that, indeed, our universe isn't broken and over time our cumalitive chance of an odd number closes in on 50%.

    We can even take our repeat function one step further, and use repeatTake, since that is exactly what we are doing.

    go(async () => {
      repeatTake(numbers, async (n, { total, odds }) => {
        put(stats, `${odds / total * 100}% odd numbers`)
        
        if (n % 2) {
          put(oddNumbers, n)
          return { total: total + 1, odds: odds + 1 }
        } else {
          return { total: total + 1, odds }
        }
        
      }, { total: 0, odds: 0 })
    })

    So we just change the signature a bit, and our local "repeat" state is passed as the second argument instead of the first.

    More documentation is coming, but the core functionality is ~160LOC, so it should just take a single cup of coffee to read through. I wanted to be sure that the API was built deliberately, and not just a port from some previous effort.

    API

    chan(numOrBuffer=null) -> Chan

    Creates a channel. All arguments are optional. numOfBuffer - Any number or buffer. A number is a shortcut for buffers.fixed(number).

    put(ch, val) -> Promise -> true|false

    Puts a value onto a channel. Returned promise resolves to true if successful, or false if the channel is closed.

    take(ch) -> Promise -> takenValue|CLOSED

    Takes a value from a channel. Returned Promise resolves to taken value or CLOSED constant if the channel is closed.

    go(async function) -> promiseFromInvokedAsyncFunction

    Immediately invokes (and returns) given function.

    sleep(ms) -> Promise

    Creates a promise that will resolve successfully after ms milliseconds.

    CLOSED

    A constant, which all takes on a closed channel receive instead of a value.

    close(ch) -> undefined

    Closes a channel. This causes:

    • all puts and pending puts to resolve to false
    • all takes and pending takes to resolve to the CLOSED constant

    clone(ch) -> Chan

    Makes a new channel, same as the old channel.

    any(...ports) -> Promise -> [theResolvedValue, theSourceChannelOrPromise]

    Like alts in Clojure's core-async.

    ports can be a channel to take from, a promise to resolve, or an array to put data onto a channel, like [ theChannel, valueToPut ].

    If none of them have a pending value, it will resolve with whichever channel receives a value next. If one of the channels has a pending value already, it will simply resolve to that. If more than one channel has a pending value, it selects one in a non-deterministic fashion.

    Always resolves with a double of [ theResolvedValue, theSourceChannel ].

    All non-winning actions will be canceled so that their data does not go missing.

    repeat(async function, seed=null) -> undefined

    This functions like a while loop, except you can track state using its return value. Return false to end. Return a value other than false, and it will be available as the argument to your callback async function. Pass in a seed value as the second argument to repeat.

    repeatTake(ch, async function, seed=null) -> undefined

    This is just like repeat above, except that before it repeats, it waits for a successful take on the given channel. Then it passes this taken value in as the first argument, with any local state being passed as the second argument.

    See the ping/pong example above to see this in action.

    merge(...chs)

    Creates a new channel that will receive all puts to the received channels.

    buffers

    buffers.unbuffered()

    No buffer space. The default choice for when first argument to chan is falsy.

    buffers.fixed(num)

    Buffer has space of num. Any extra puts are parked.

    buffers.sliding(num)

    Buffer simply slides across pending puts as a window of num width. So, oldest puts are dropped as new ones are added.

    buffers.dropping(num)

    Buffer drops, and resolves, any extra puts beyond num.

    Install

    npm i medium

    DownloadsWeekly Downloads

    2,593

    Version

    1.2.0

    License

    ISC

    Unpacked Size

    35.8 kB

    Total Files

    14

    Last publish

    Collaborators

    • bbarr