gistfile1.coffee

class EventStream
  constructor: (@options = {}, @stream = undefined) ->
    unless @stream?
      streamType = EventStream.subjectType(@options.repeating)
      @stream = new streamType()

  publish: (eventName, args...) =>
    @stream.onNext.call(@stream, [eventName].concat(args))

  on: (eventName, callback) =>
    @stream.filter((ev) =>
      ev[0] is eventName).
    subscribe (args) =>
      callback(args.slice(1)...)

  map: (fun) =>
    subjectType = EventStream.subjectType(@options.repeating)
    metaSubject = new subjectType()
    new EventStream({}, subjectType.create(metaSubject, Rx.Observable.merge(@stream, metaSubject).map(fun)))

  @merge: (options, streams...) =>
    subjectType = EventStream.subjectType(options.repeating)
    metaSubject = new subjectType()

    eventStreams = streams.map (eventStream) => eventStream.stream
    eventStreams.push(metaSubject)

    new EventStream(options, subjectType.create(metaSubject, Rx.Observable.merge(eventStreams...)))

  @subjectType: (repeating) =>
    return Rx.ReplaySubject if repeating
    Rx.Subject

module.exports = EventStream