<commit_before>import Rx, {Observable as O} from "rx"

function RxAdapter(obs, shared) {
  this.o = obs
  this.shared = shared || false
}

function RxBus() {
  this.s = new Rx.Subject()
}

Object.assign(RxAdapter.prototype, {
  return() {
    return this.shared ? this.o : this.o.share()
  },
  multicast() {
    return new RxAdapter(this.shared ? this.o : this.o.share(), true)
  },
  map(fn) {
    return new RxAdapter(this.o.map(fn))
  },
  filter(fn) {
    return new RxAdapter(this.o.filter(fn))
  },
  doOnCompleted(fn) {
    return new RxAdapter(this.o.doOnCompleted(fn))
  },
  scan(fn, seed) {
    return new RxAdapter(this.o.scan(fn, seed))
  },
  flatMap(fn) {
    return new RxAdapter(this.o.flatMap(x => fn(x).return()))
  },
  flatMapLatest(fn) {
    return new RxAdapter(this.o.flatMapLatest(x => fn(x).return()))
  },
  skipDuplicates(eq) {
    return new RxAdapter(this.o.distinctUntilChanged(x => x, eq))
  },
  toProperty() {
    return new RxAdapter(this.o.shareReplay(1), true)
  },
  hot(toProp) {
    const obs = toProp ? this.o.replay(null, 1) : this.o.publish()
    const disposable = obs.connect()
    const dispose = () => disposable.dispose()
    return [new RxAdapter(obs, toProp), dispose]
  }
})

Object.assign(RxBus.prototype, {
  obs() {
    return new RxAdapter(this.s.asObservable())
  },
  next(val) {
    this.s && this.s.onNext(val)
  },
  completed() {
    if (this.s) {
      const s = this.s
      this.s = void 0
      s.onCompleted()
      s.dispose()
    }
  },
  error() {
    if (this.s) {
      const s = this.s
      this.s = void 0
      s.onError()
      s.dispose()
    }
  }
})


Object.assign(RxAdapter, {
  is(obs) {
    return obs && O.isObservable(obs)
  },
  create(fn) {
    return new RxAdapter(O.create(o => {
      return fn(toObserver(o))
    }))
  },
  just(val) {
    return new RxAdapter(O.just(val))
  },
  error(err) {
    return new RxAdapter(O.throw(err))
  },
  combine(list) {
    return new RxAdapter(list.length === 0 ? O.just([]) : O.combineLatest(list.map(o => o.return())))
  },
  merge(obs) {
    return new RxAdapter(O.merge(obs.map(o => o.return())))
  },<commit_msg>Add O.never() and O.empty()<commit_after>import Rx, {Observable as O} from "rx"

function RxAdapter(obs, shared) {
  this.o = obs
  this.shared = shared || false
}

function RxBus() {
  this.s = new Rx.Subject()
}

Object.assign(RxAdapter.prototype, {
  return() {
    return this.shared ? this.o : this.o.share()
  },
  multicast() {
    return new RxAdapter(this.shared ? this.o : this.o.share(), true)
  },
  map(fn) {
    return new RxAdapter(this.o.map(fn))
  },
  filter(fn) {
    return new RxAdapter(this.o.filter(fn))
  },
  doOnCompleted(fn) {
    return new RxAdapter(this.o.doOnCompleted(fn))
  },
  scan(fn, seed) {
    return new RxAdapter(this.o.scan(fn, seed))
  },
  flatMap(fn) {
    return new RxAdapter(this.o.flatMap(x => fn(x).return()))
  },
  flatMapLatest(fn) {
    return new RxAdapter(this.o.flatMapLatest(x => fn(x).return()))
  },
  skipDuplicates(eq) {
    return new RxAdapter(this.o.distinctUntilChanged(x => x, eq))
  },
  toProperty() {
    return new RxAdapter(this.o.shareReplay(1), true)
  },
  hot(toProp) {
    const obs = toProp ? this.o.replay(null, 1) : this.o.publish()
    const disposable = obs.connect()
    const dispose = () => disposable.dispose()
    return [new RxAdapter(obs, toProp), dispose]
  }
})

Object.assign(RxBus.prototype, {
  obs() {
    return new RxAdapter(this.s.asObservable())
  },
  next(val) {
    this.s && this.s.onNext(val)
  },
  completed() {
    if (this.s) {
      const s = this.s
      this.s = void 0
      s.onCompleted()
      s.dispose()
    }
  },
  error() {
    if (this.s) {
      const s = this.s
      this.s = void 0
      s.onError()
      s.dispose()
    }
  }
})


Object.assign(RxAdapter, {
  is(obs) {
    return obs && O.isObservable(obs)
  },
  create(fn) {
    return new RxAdapter(O.create(o => {
      return fn(toObserver(o))
    }))
  },
  just(val) {
    return new RxAdapter(O.just(val))
  },
  never() {
    return O.never()
  },
  empty() {
    return O.empty()
  },
  error(err) {
    return new RxAdapter(O.throw(err))
  },
  combine(list) {
    return new RxAdapter(list.length === 0 ? O.just([]) : O.combineLatest(list.map(o => o.return())))
  },
  merge(obs) {
    return new RxAdapter(O.merge(obs.map(o => o.return())))
  },
{'fraction_non_alphanumeric': 0.1757188498402556, 'fraction_numerical': 0.0022820629849383844, 'mean_word_length': 2.7719449225473323, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 17, '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': '6045270', 'n_tokens_mistral': 1726, 'n_tokens_neox': 1667, 'n_words': 313}