channels-examples.js

var Channels = require('./channels')
  , Emitter = require('events').EventEmitter

var repeater = Channels.createRepeating(250, 'hello', 'world')
  , emitter = new Emitter()
  , emitterTestChannel = Channels.fromEmitter(emitter, 'test')
  , mergedChannel = Channels.merge(repeater, emitterTestChannel)
  , mappedChannel = Channels.map(function() {
                      return Array.prototype.join.call(arguments, ' ')
                    }, repeater)
  , reducedChannel = Channels.reduce(function(channelArgs) {
                       var flattenedArgs = []
                       channelArgs.forEach(function(args) {
                         flattenedArgs = flattenedArgs.concat(args)
                       })
                       return flattenedArgs.join(', ')
                     }, repeater, emitterTestChannel)

repeater.on('update', function() {
  console.log("\n", 'repeater', arguments)
})

emitterTestChannel.on('update', function() {
  console.log("\n", 'emitterTestChannel', arguments)
})

mergedChannel.on('update', function() {
  console.log("\n", 'mergedChannel', arguments)
})

mappedChannel.on('update', function() {
  console.log("\n", 'mappedChannel', arguments)
})

reducedChannel.on('update', function() {
  console.log("\n", 'reducedChannel', arguments)
})

setInterval(function() {
  emitter.emit('test', 'this is only a test')
}, 500)

channels.js

var Emitter = require('events').EventEmitter

function slice(arrayLike, start, end) {
  return Array.prototype.slice.call(arrayLike, start, end)
}

function toArray(arrayLike) {
  return slice(arrayLike, 0)
}

var Channels = {
  createRepeating: function(interval) {
    var args = slice(arguments, 1)
      , channel = new Emitter()
    args.unshift('update')
    setInterval(function() {
      channel.emit.apply(channel, args)
    }, interval)
    return channel
  },
  fromEmitter: function(emitter, eventName) {
    var channel = new Emitter()
    emitter.on(eventName, function() {
      var args = toArray(arguments)
      args.unshift('update')
      channel.emit.apply(channel, args)
    })
    return channel;
  },
  merge: function() {
    var channels = toArray(arguments)
      , mergedChannel = new Emitter()
    channels.forEach(function(channel) {
      channel.on('update', function() {
        var args = toArray(arguments)
        args.unshift('update')
        mergedChannel.emit.apply(mergedChannel, args)
      })
    })
    return mergedChannel
  },
  map: function(f, channel) {
    var mappedChannel = new Emitter()
    channel.on('update', function() {
      mappedChannel.emit('update', f.apply(null, toArray(arguments)))
    })
    return mappedChannel
  },
  reduce: function(f) {
    var channels = slice(arguments, 1)
      , values = []
      , reducedChannel = new Emitter()
    for(var i = 0; i < channels.length; ++i) {
      values[i] = []
      channels[i].on('update', makeOnUpdate(i))
    }
    return reducedChannel

    function makeOnUpdate(index) {
      return function() {
        values[index] = toArray(arguments)
        reducedChannel.emit.call(reducedChannel, 'update', f(values))
      }
    }
  },
  interleave: function() {
    var channels = toArray(arguments)
      , interleavedChannel = new Emitter()
      , index = 0
      , updates = []
    channels.forEach(function() {
      var channelUpdates = []
      updates.push(channelUpdates)
      channels.on('update', function() {
        channelUpdates.push(toArray(arguments))
      })
      tryEmitUpdate()
    })
    return interleavedChannel

    function tryEmitUpdate() {
      var channelUpdates = updates[index]
      if (channelUpdates.length > 0) {
        var update = channelUpdates.unshift()
        index = (i + 1) % channels.length
        update.unshift('update')
        interleavedChannel.emit.apply(interleavedChannel, update)
      }
    }
  }
}

module.exports = Channels