<commit_before>import * as Bacon from 'baconjs'
import { Readable, PassThrough, ReadableOptions } from 'stream'

export function readableToBacon(stream: Readable): Bacon.EventStream<any, any> {
  return Bacon.fromBinder(sink => {
    const sinkValue = (value: any) => sink(value)
    const sinkError = (err: Error) => sink(new Bacon.Error(err))
    const sinkEnd = () => sink(new Bacon.End())

    const passthrough = new PassThrough()
    passthrough.on('data', sinkValue)
    passthrough.on('error', sinkError)
    passthrough.on('close', sinkEnd)
    passthrough.on('end', sinkEnd)

    stream.pipe(passthrough)

    return () => {
      stream.unpipe(passthrough)
      passthrough.end()
    }
  })
}

export function baconToReadable<A>(stream: Bacon.EventStream<any, A>): Readable {
  return new BaconReadable({}, stream)
}

interface BaconReadableEvent {
  type: 'value' | 'error'
  value: any
}

class BaconReadable extends Readable {
  private awaitingData: boolean = false
  private buffer: BaconReadableEvent[] = []
  private unsubscribeStream: () => void

  constructor(options: ReadableOptions, stream: Bacon.EventStream<any, any>) {
    super(options)
    this.unsubscribeStream = stream.subscribe(event => {
      if (event.isEnd()) {
        if (this.awaitingData) {
          this.pushBuffer()
        }
        this.push(null)
        return Bacon.noMore
      }

      if (event.isError()) {
        const errorEvent = event as Bacon.Error<any>
        this.buffer.push({ type: 'error', value: errorEvent.error })
      } else if (event.hasValue()) {
        this.buffer.push({ type: 'value', value: event.value() })
      }

      if (this.awaitingData) {
        this.pushBuffer()
      }
    })
  }<commit_msg>Fix handling end event<commit_after>import * as Bacon from 'baconjs'
import { Readable, PassThrough, ReadableOptions } from 'stream'

export function readableToBacon(stream: Readable): Bacon.EventStream<any, any> {
  return Bacon.fromBinder(sink => {
    const sinkValue = (value: any) => sink(value)
    const sinkError = (err: Error) => sink(new Bacon.Error(err))
    const sinkEnd = () => sink(new Bacon.End())

    const passthrough = new PassThrough()
    passthrough.on('data', sinkValue)
    passthrough.on('error', sinkError)
    passthrough.on('close', sinkEnd)
    passthrough.on('end', sinkEnd)

    stream.pipe(passthrough)

    return () => {
      stream.unpipe(passthrough)
      passthrough.end()
    }
  })
}

export function baconToReadable<A>(stream: Bacon.EventStream<any, A>): Readable {
  return new BaconReadable({}, stream)
}

interface BaconReadableEvent {
  type: 'value' | 'error' | 'end'
  value: any
}

class BaconReadable extends Readable {
  private awaitingData: boolean = false
  private buffer: BaconReadableEvent[] = []
  private unsubscribeStream: () => void

  constructor(options: ReadableOptions, stream: Bacon.EventStream<any, any>) {
    super(options)
    this.unsubscribeStream = stream.subscribe(event => {
      const streamEnded = event.isEnd()
      if (streamEnded) {
        this.buffer.push({ type: 'end', value: null })
      } else if (event.isError()) {
        const errorEvent = event as Bacon.Error<any>
        this.buffer.push({ type: 'error', value: errorEvent.error })
      } else if (event.hasValue()) {
        this.buffer.push({ type: 'value', value: event.value() })
      }

      if (this.awaitingData) {
        this.pushBuffer()
      }

      if (streamEnded) {
        return Bacon.noMore
      } else {
        return undefined
      }
    })
  }
{'fraction_non_alphanumeric': 0.1341222879684418, 'fraction_numerical': 0.0, 'mean_word_length': 3.191263282172373, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 27, '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': '1719408', 'n_tokens_mistral': 1159, 'n_tokens_neox': 1112, 'n_words': 254}