<commit_before>/** @license MIT License (c) copyright 2010-2016 original author or authors */
/** @author <NAME> */
/** @author <NAME> */

import { disposeBoth, disposeNone, tryDispose } from '@most/disposable'

/**
 * Given a stream of streams, return a new stream that adopts the behavior
 * of the most recent inner stream.
 * @param {Stream} stream of streams on which to switch
 * @returns {Stream} switching stream
 */
export const switchLatest = stream => new Switch(stream)

class Switch {
  constructor (source) {
    this.source = source
  }

  run (sink, scheduler) {
    const switchSink = new SwitchSink(sink, scheduler)
    return disposeBoth(switchSink, this.source.run(switchSink, scheduler))
  }
}

class SwitchSink {
  constructor (sink, scheduler) {
    this.sink = sink
    this.scheduler = scheduler
    this.current = null
    this.ended = false
  }

  event (t, stream) {
    this._disposeCurrent(t) // TODO: capture the result of this dispose
    this.current = new Segment(t, Infinity, this, this.sink)
    this.current.disposable = stream.run(this.current, this.scheduler)
  }

  end (t) {
    this.ended = true
    this._checkEnd(t)
  }

  error (t, e) {
    this.ended = true
    this.sink.error(t, e)
  }

  dispose () {
    return this._disposeCurrent(this.scheduler.now())
  }

  _disposeCurrent (t) {
    if (this.current !== null) {
      return this.current._dispose(t)
    }
  }

  _disposeInner (t, inner) {
    inner._dispose(t) // TODO: capture the result of this dispose
    if (inner === this.current) {
      this.current = null
    }
  }

  _checkEnd (t) {
    if (this.ended && this.current === null) {
      this.sink.end(t)
    }
  }

  _endInner (t, inner) {
    this._disposeInner(t, inner)
    this._checkEnd(t)
  }

  _errorInner (t, e, inner) {
    this._disposeInner(t, inner)
    this.sink.error(t, e)
  }
}

class Segment {
  constructor (min, max, outer, sink) {
    this.min = min
    this.max = max
    this.outer = outer
    this.sink = sink
    this.disposable = disposeNone()
  }

  event (t, x) {
    if (t < this.max) {
      this.sink.event(Math.max(t, this.min), x)
    }
  }

  end (t) {
    this.outer._endInner(Math.max(t, this.min), this)
  }

  error (t, e) {
    this.outer._errorInner(Math.max(t, this.min), e, this)
  }

  _dispose (t) {
    this.max = t
    tryDispose(t, this.disposable, this.sink)
  }
}<commit_msg>chore(switchLatest): avoid mutating Segment (#74)<commit_after>/** @license MIT License (c) copyright 2010-2016 original author or authors */
/** @author <NAME> */
/** @author <NAME> */

import { disposeBoth, tryDispose } from '@most/disposable'

/**
 * Given a stream of streams, return a new stream that adopts the behavior
 * of the most recent inner stream.
 * @param {Stream} stream of streams on which to switch
 * @returns {Stream} switching stream
 */
export const switchLatest = stream => new Switch(stream)

class Switch {
  constructor (source) {
    this.source = source
  }

  run (sink, scheduler) {
    const switchSink = new SwitchSink(sink, scheduler)
    return disposeBoth(switchSink, this.source.run(switchSink, scheduler))
  }
}

class SwitchSink {
  constructor (sink, scheduler) {
    this.sink = sink
    this.scheduler = scheduler
    this.current = null
    this.ended = false
  }

  event (t, stream) {
    this._disposeCurrent(t) // TODO: capture the result of this dispose
    this.current = new Segment(stream, t, Infinity, this, this.sink, this.scheduler)
  }

  end (t) {
    this.ended = true
    this._checkEnd(t)
  }

  error (t, e) {
    this.ended = true
    this.sink.error(t, e)
  }

  dispose () {
    return this._disposeCurrent(this.scheduler.now())
  }

  _disposeCurrent (t) {
    if (this.current !== null) {
      return this.current._dispose(t)
    }
  }

  _disposeInner (t, inner) {
    inner._dispose(t) // TODO: capture the result of this dispose
    if (inner === this.current) {
      this.current = null
    }
  }

  _checkEnd (t) {
    if (this.ended && this.current === null) {
      this.sink.end(t)
    }
  }

  _endInner (t, inner) {
    this._disposeInner(t, inner)
    this._checkEnd(t)
  }

  _errorInner (t, e, inner) {
    this._disposeInner(t, inner)
    this.sink.error(t, e)
  }
}

class Segment {
  constructor (source, min, max, outer, sink, scheduler) {
    this.min = min
    this.max = max
    this.outer = outer
    this.sink = sink
    this.disposable = source.run(this, scheduler)
  }

  event (t, x) {
    if (t < this.max) {
      this.sink.event(Math.max(t, this.min), x)
    }
  }

  end (t) {
    this.outer._endInner(Math.max(t, this.min), this)
  }

  error (t, e) {
    this.outer._errorInner(Math.max(t, this.min), e, this)
  }

  _dispose (t) {
    this.max = t
    tryDispose(t, this.disposable, this.sink)
  }
}
{'fraction_non_alphanumeric': 0.13641111343221224, 'fraction_numerical': 0.003760183831209526, 'mean_word_length': 2.7523510971786833, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 9, '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': '6704554', 'n_tokens_mistral': 1751, 'n_tokens_neox': 1691, 'n_words': 433}