<commit_before>if module?
  module.exports = Bacon = {} # for Bacon = require 'baconjs'
  Bacon.Bacon = Bacon # for {Bacon} = require 'baconjs'
else
  if typeof require is 'function' and require.amd?
      define 'bacon', [], -> Bacon
  @Bacon = Bacon = {} # otherwise for execution context

# eventTransformer - should return one value or one or many events
Bacon.fromBinder = (binder, eventTransformer = _.id) ->
  new EventStream (sink) ->
    unbinder = binder (args...) ->
      value = eventTransformer(args...)
      unless value instanceof Array and _.last(value) instanceof Event
        value = [value]

      reply = Bacon.more
      for event in value
        reply = sink(event = toEvent(event))
        if reply == Bacon.noMore or event.isEnd()
          # defer if binder calls handler in sync before returning unbinder
          if unbinder? then unbinder() else Bacon.scheduler.setTimeout (-> unbinder()), 0
          return reply
      reply

# eventTransformer - defaults to returning the first argument to handler
Bacon.$ = asEventStream: (eventName, selector, eventTransformer) ->
  [eventTransformer, selector] = [selector, null] if isFunction(selector)
  Bacon.fromBinder (handler) =>
    @on(eventName, selector, handler)
    => @off(eventName, selector, handler)
  , eventTransformer

(jQuery ? (Zepto ? null))?.fn.asEventStream = Bacon.$.asEventStream

# Wrap DOM EventTarget, Node EventEmitter, or
# [un]bind: (Any, (Any) -> None) -> None interfaces
# common in MVCs as EventStream
#
# target - EventTarget or EventEmitter, source of events
# eventName - event name to bind
# eventTransformer - defaults to returning the first argument to handler
#
# Examples
#
#   Bacon.fromEventTarget(document.body, "click")
#   # => EventStream
#
#   Bacon.fromEventTarget (new EventEmitter(), "data")
#   # => EventStream
#
# Returns EventStream
Bacon.fromEventTarget = (target, eventName, eventTransformer) ->
  sub = target.addEventListener ? (target.addListener ? target.bind)
  unsub = target.removeEventListener ? (target.removeListener ? target.unbind)
  Bacon.fromBinder (handler) ->
    sub.call(target, eventName, handler)
    -> unsub.call(target, eventName, handler)
  , eventTransformer

Bacon.fromPromise = (promise, abort) ->
  Bacon.fromBinder (handler) ->
    promise.then(handler, (e) -> handler(new Error(e)))
    -> promise.abort?() if abort
  , (value) -> [value, end()]


Bacon.noMore = ["<no-more>"]

Bacon.more = ["<more>"]

Bacon.later = (delay, value) ->
  Bacon.sequentially(delay, [value])

Bacon.sequentially = (delay, values) ->
  index = 0
  Bacon.fromPoll delay, ->
    value = values[index++]
    if index < values.length 
      value 
    else if index == values.length
      [value, end()]
    else
      end()

Bacon.repeatedly = (delay, values) ->
  index = 0
  Bacon.fromPoll(delay, -> values[index++ % values.length])

liftCallback = (wrapped) ->
  return (f, args...) ->
    stream = partiallyApplied(wrapped, [(values, callback) ->
      f(values..., callback)])
    Bacon.combineAsArray(args).flatMap(stream)

Bacon.fromCallback = liftCallback (f, args...) ->
  Bacon.fromBinder (handler) ->
    makeFunction(f, args)(handler)
    nop
  , (value) -> [value, end()]

Bacon.fromNodeCallback = liftCallback (f, args...) ->
  Bacon.fromBinder (handler) ->
    makeFunction(f, args)(handler)
    nop
  , (error, value) ->
      return [new Error(error), end()] if error
      [value, end()]

Bacon.fromPoll = (delay, poll) ->
  Bacon.fromBinder (handler) ->
    id = Bacon.scheduler.setInterval(handler, delay)
    -> Bacon.scheduler.clearInterval(id)
  , poll

Bacon.interval = (delay, value) ->
  value = {} unless value?
  Bacon.fromPoll(delay, -> next(value))

Bacon.constant = (value) ->
  new Property(sendWrapped([value], initial), true)

Bacon.never = -> Bacon.fromArray([])

Bacon.once = (value) -> Bacon.fromArray([value])

Bacon.fromArray = (values) ->
  assertArray values
  new EventStream(sendWrapped(values, toEvent))

sendWrapped = (values, wrapper) ->
  (sink) ->
    for value in values
      sink (wrapper value)
    sink (end())
    nop

Bacon.mergeAll = (streams, more...) ->
  if not (streams instanceof Array)
    streams = [streams].concat(more)
  _.fold(streams, Bacon.never(), ((a, b) -> a.merge(b)))

Bacon.zipAsArray = (streams, more...) ->
  if not (streams instanceof Array)
    streams = [streams].concat(more)
  Bacon.zipWith(streams, Array)

Bacon.zipWith = (streams, f, more...) ->
  if isFunction(streams)
    [streams, f] = [[f].concat(more), streams]
  Bacon.when streams, f

Bacon.combineAsArray = (streams, more...) ->
  if not (streams instanceof Array)
    streams = [streams].concat(more)
  for stream, index in streams
    streams[index] = Bacon.constant(stream) if not (stream instanceof Observable)
  if streams.length
    values = (None for s in streams)
    new Property (sink) =>
      ends = (false for s in streams)
      checkEnd = (unsubAll) ->
        if _.all(ends)
          reply = sink end()
          unsubAll() if reply == Bacon.noMore
      initialSent = false
      combiningSink = (stream, index) => (unsubAll) ->
        stream.subscribeInternal (event) =>
          if (event.isEnd())
            ends[index] = true
            checkEnd(unsubAll)
            Bacon.noMore
          else if event.isError()
            reply = sink event
            unsubAll() if reply == Bacon.noMore
            reply
          else
            values[index] = event.value
            if _.all(values, ((x) -> x != None))
              if initialSent and event.isInitial()
                # don't send duplicate Initial
                Bacon.more
              else
                initialSent = true
                valueArrayF = -> (x() for x in values)
                reply = sink(event.apply(valueArrayF))
                unsubAll() if reply == Bacon.noMore
                reply
            else
              Bacon.more
      subs = for stream, index in streams 
        do (stream, index) ->
          combiningSink stream, index
      compositeUnsubscribe subs...
  else
    Bacon.constant([])

Bacon.onValues = (streams..., f) -> Bacon.combineAsArray(streams).onValues(f)

Bacon.combineWith = (f, streams...) ->
  Bacon.combineAsArray(streams).map (values) -> f(values...)
<commit_msg>combineAsArray: Simplify subscription based on previous refactoring<commit_after>if module?
  module.exports = Bacon = {} # for Bacon = require 'baconjs'
  Bacon.Bacon = Bacon # for {Bacon} = require 'baconjs'
else
  if typeof require is 'function' and require.amd?
      define 'bacon', [], -> Bacon
  @Bacon = Bacon = {} # otherwise for execution context

# eventTransformer - should return one value or one or many events
Bacon.fromBinder = (binder, eventTransformer = _.id) ->
  new EventStream (sink) ->
    unbinder = binder (args...) ->
      value = eventTransformer(args...)
      unless value instanceof Array and _.last(value) instanceof Event
        value = [value]

      reply = Bacon.more
      for event in value
        reply = sink(event = toEvent(event))
        if reply == Bacon.noMore or event.isEnd()
          # defer if binder calls handler in sync before returning unbinder
          if unbinder? then unbinder() else Bacon.scheduler.setTimeout (-> unbinder()), 0
          return reply
      reply

# eventTransformer - defaults to returning the first argument to handler
Bacon.$ = asEventStream: (eventName, selector, eventTransformer) ->
  [eventTransformer, selector] = [selector, null] if isFunction(selector)
  Bacon.fromBinder (handler) =>
    @on(eventName, selector, handler)
    => @off(eventName, selector, handler)
  , eventTransformer

(jQuery ? (Zepto ? null))?.fn.asEventStream = Bacon.$.asEventStream

# Wrap DOM EventTarget, Node EventEmitter, or
# [un]bind: (Any, (Any) -> None) -> None interfaces
# common in MVCs as EventStream
#
# target - EventTarget or EventEmitter, source of events
# eventName - event name to bind
# eventTransformer - defaults to returning the first argument to handler
#
# Examples
#
#   Bacon.fromEventTarget(document.body, "click")
#   # => EventStream
#
#   Bacon.fromEventTarget (new EventEmitter(), "data")
#   # => EventStream
#
# Returns EventStream
Bacon.fromEventTarget = (target, eventName, eventTransformer) ->
  sub = target.addEventListener ? (target.addListener ? target.bind)
  unsub = target.removeEventListener ? (target.removeListener ? target.unbind)
  Bacon.fromBinder (handler) ->
    sub.call(target, eventName, handler)
    -> unsub.call(target, eventName, handler)
  , eventTransformer

Bacon.fromPromise = (promise, abort) ->
  Bacon.fromBinder (handler) ->
    promise.then(handler, (e) -> handler(new Error(e)))
    -> promise.abort?() if abort
  , (value) -> [value, end()]


Bacon.noMore = ["<no-more>"]

Bacon.more = ["<more>"]

Bacon.later = (delay, value) ->
  Bacon.sequentially(delay, [value])

Bacon.sequentially = (delay, values) ->
  index = 0
  Bacon.fromPoll delay, ->
    value = values[index++]
    if index < values.length 
      value 
    else if index == values.length
      [value, end()]
    else
      end()

Bacon.repeatedly = (delay, values) ->
  index = 0
  Bacon.fromPoll(delay, -> values[index++ % values.length])

liftCallback = (wrapped) ->
  return (f, args...) ->
    stream = partiallyApplied(wrapped, [(values, callback) ->
      f(values..., callback)])
    Bacon.combineAsArray(args).flatMap(stream)

Bacon.fromCallback = liftCallback (f, args...) ->
  Bacon.fromBinder (handler) ->
    makeFunction(f, args)(handler)
    nop
  , (value) -> [value, end()]

Bacon.fromNodeCallback = liftCallback (f, args...) ->
  Bacon.fromBinder (handler) ->
    makeFunction(f, args)(handler)
    nop
  , (error, value) ->
      return [new Error(error), end()] if error
      [value, end()]

Bacon.fromPoll = (delay, poll) ->
  Bacon.fromBinder (handler) ->
    id = Bacon.scheduler.setInterval(handler, delay)
    -> Bacon.scheduler.clearInterval(id)
  , poll

Bacon.interval = (delay, value) ->
  value = {} unless value?
  Bacon.fromPoll(delay, -> next(value))

Bacon.constant = (value) ->
  new Property(sendWrapped([value], initial), true)

Bacon.never = -> Bacon.fromArray([])

Bacon.once = (value) -> Bacon.fromArray([value])

Bacon.fromArray = (values) ->
  assertArray values
  new EventStream(sendWrapped(values, toEvent))

sendWrapped = (values, wrapper) ->
  (sink) ->
    for value in values
      sink (wrapper value)
    sink (end())
    nop

Bacon.mergeAll = (streams, more...) ->
  if not (streams instanceof Array)
    streams = [streams].concat(more)
  _.fold(streams, Bacon.never(), ((a, b) -> a.merge(b)))

Bacon.zipAsArray = (streams, more...) ->
  if not (streams instanceof Array)
    streams = [streams].concat(more)
  Bacon.zipWith(streams, Array)

Bacon.zipWith = (streams, f, more...) ->
  if isFunction(streams)
    [streams, f] = [[f].concat(more), streams]
  Bacon.when streams, f

Bacon.combineAsArray = (streams, more...) ->
  if not (streams instanceof Array)
    streams = [streams].concat(more)
  for stream, index in streams
    streams[index] = Bacon.constant(stream) if not (stream instanceof Observable)
  if streams.length
    values = (None for s in streams)
    new Property (sink) =>
      ends = (false for s in streams)
      checkEnd = (unsubAll) ->
        if _.all(ends)
          reply = sink end()
          unsubAll() if reply == Bacon.noMore
      initialSent = false
      combiningSink = (index) => (unsubAll) ->
        streams[index].subscribeInternal (event) =>
          if (event.isEnd())
            ends[index] = true
            checkEnd(unsubAll)
            Bacon.noMore
          else if event.isError()
            reply = sink event
            unsubAll() if reply == Bacon.noMore
            reply
          else
            values[index] = event.value
            if _.all(values, ((x) -> x != None))
              if initialSent and event.isInitial()
                # don't send duplicate Initial
                Bacon.more
              else
                initialSent = true
                valueArrayF = -> (x() for x in values)
                reply = sink(event.apply(valueArrayF))
                unsubAll() if reply == Bacon.noMore
                reply
            else
              Bacon.more
      compositeUnsubscribe (combiningSink index for s, index in streams )...
  else
    Bacon.constant([])

Bacon.onValues = (streams..., f) -> Bacon.combineAsArray(streams).onValues(f)

Bacon.combineWith = (f, streams...) ->
  Bacon.combineAsArray(streams).map (values) -> f(values...)

{'fraction_non_alphanumeric': 0.1350878587898545, 'fraction_numerical': 0.0004770613023773555, 'mean_word_length': 2.8011483831973405, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 128, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3900052', 'n_tokens_mistral': 3943, 'n_tokens_neox': 3891, 'n_words': 1045}