<commit_before>import { Observable } from 'rxjs/Observable'
import { of } from 'rxjs/observable/of'
import { merge } from 'rxjs/observable/merge'
import { delay } from 'rxjs/operator/delay'
import { takeUntil } from 'rxjs/operator/takeUntil'
import { share } from 'rxjs/operator/share'

const makeVirtualEmission = (scheduler, value) => {
  return new Observable(observer => {
    scheduler.schedule(() => {
      observer.next(value)
    })
  })
}

const makeVirtualStream = (scheduler, diagram) => {
  const { emissions, completion } = diagram

  const partials = emissions.map(({ x, d }) => (
    makeVirtualEmission(scheduler, d)
      ::delay(x, scheduler)
  ))

  const completion$ = of(null)::delay(completion, scheduler)

  const emission$ = merge(...partials)
    ::takeUntil(completion$)
    ::share()

  return emission$
}
<commit_msg>Move delay to virtual scheduler method<commit_after>import { Observable } from 'rxjs/Observable'
import { of } from 'rxjs/observable/of'
import { merge } from 'rxjs/observable/merge'
import { delay } from 'rxjs/operator/delay'
import { takeUntil } from 'rxjs/operator/takeUntil'
import { share } from 'rxjs/operator/share'

const makeVirtualEmission = (scheduler, value, delay) => {
  return new Observable(observer => {
    scheduler.schedule(() => {
      observer.next(value)
      observer.complete()
    }, delay, value)
  })
}

const makeVirtualStream = (scheduler, diagram) => {
  const { emissions, completion } = diagram

  const partials = emissions.map(({ x, d }) => (
    makeVirtualEmission(scheduler, d, x)
  ))

  const completion$ = of(null)::delay(completion, scheduler)

  const emission$ = merge(...partials)
    ::takeUntil(completion$)
    ::share()

  return emission$
}

{'fraction_non_alphanumeric': 0.14261069580218516, 'fraction_numerical': 0.0, 'mean_word_length': 4.028901734104046, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 13, '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': '1548563', 'n_tokens_mistral': 549, 'n_tokens_neox': 534, 'n_words': 130}