<commit_before>import MulticastDisposable from './MulticastDisposable';
import {append, remove, findIndex} from './util';

function dispose(disposable) {
    if (disposable === void 0) {
        return;
    }
    return disposable.dispose();
}

export default class MulticastSource {
  constructor(source) {
      this.source = source;
      this.sinks = [];
      this._disposable = void 0;
  }

  run(sink, scheduler) {
      const n = this.add(sink);
      if (n === 1) {
          this._disposable = this.source.run(this, scheduler);
      }
      return new MulticastDisposable(this, sink);
  }

  _dispose() {
      const disposable = this_disposable;
      this._disposable = void 0;
      return Promise.resolve(disposable).then(dispose);
  }

  add(sink) {
      this.sinks = append(sink, this.sinks);
      return this.sinks.length;
  }

  remove(sink) {
      this.sinks = remove(findIndex(sink, this.sinks), this.sinks);
      return this.sinks.length;
  }

  event(time, value) {
      const s = this.sinks;
      if (s.length === 1) {
          s[0].event(time, value);
          return;
      }
      for (let i = 0; i < s.length; ++i) {
          s[i].event(time, value);
      }
  }

  end(time, value) {
      const s = this.sinks;
      if (s.length === 1) {
          s[0].end(time, value);
          return;
      }
      for (let i = 0; i < s.length; ++i) {
          s[i].end(time, value);
      }
  }

  error(time, err) {
      const s = this.sinks;
      if (s.length === 1) {
          s[0].end(time, err);
          return;
      }
      for (let i = 0; i < s.length; ++i) {
          s[i].end(time, err);
      }
  }
}<commit_msg>fix(MulticastSource): Have error() throw errors and correct reference to disposable<commit_after>import MulticastDisposable from './MulticastDisposable';
import {append, remove, findIndex} from './util';

function dispose(disposable) {
    if (disposable === void 0) {
        return;
    }
    return disposable.dispose();
}

export default class MulticastSource {
  constructor(source) {
      this.source = source;
      this.sinks = [];
      this._disposable = void 0;
  }

  run(sink, scheduler) {
      const n = this.add(sink);
      if (n === 1) {
          this._disposable = this.source.run(this, scheduler);
      }
      return new MulticastDisposable(this, sink);
  }

  _dispose() {
      const disposable = this._disposable;
      this._disposable = void 0;
      return Promise.resolve(disposable).then(dispose);
  }

  add(sink) {
      this.sinks = append(sink, this.sinks);
      return this.sinks.length;
  }

  remove(sink) {
      this.sinks = remove(findIndex(sink, this.sinks), this.sinks);
      return this.sinks.length;
  }

  event(time, value) {
      const s = this.sinks;
      if (s.length === 1) {
          s[0].event(time, value);
          return;
      }
      for (let i = 0; i < s.length; ++i) {
          s[i].event(time, value);
      }
  }

  end(time, value) {
      const s = this.sinks;
      if (s.length === 1) {
          s[0].end(time, value);
          return;
      }
      for (let i = 0; i < s.length; ++i) {
          s[i].end(time, value);
      }
  }

  error(time, err) {
      const s = this.sinks;
      if (s.length === 1) {
          s[0].error(time, err);
          return;
      }
      for (let i = 0; i < s.length; ++i) {
          s[i].error(time, err);
      }
  }
}
{'fraction_non_alphanumeric': 0.15089890951959917, 'fraction_numerical': 0.007662835249042145, 'mean_word_length': 2.0008841732979663, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 3, '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': '2845839', 'n_tokens_mistral': 1234, 'n_tokens_neox': 1171, 'n_words': 263}