1

Reading a Node Stream I want to be able to receive a stream of text, and trigger the continuation of my stream.

The following code solves my purposes but I recently read that we are using Subject too much when we may not need to. I'll use this code to read files from disk or from S3.

Is it possible to replace my subjects with observable or is it possible to do it without using Subject?

Best practice is use Subject sparingly because of different reasons: my important is that is not reusable, if something fails in the processing I can't kickstart it again.

I think I can do my continueStream using an event-emitter as a data source and digest it via a Observable.from. Is it worthwhile to follow down this path?

import { split, mapSync } from 'event-stream'

import { Subject } from 'rxjs/Subject'
import 'rxjs/add/operator/finally'
import 'rxjs/add/operator/catch'

// This will receive your Node Stream
// It will return two observable, one for getting the lines
// and another to continue pulling from the stream
function streamProcessing (stream: NodeJS.ReadableStream) {
  const source = new Subject()
  const continueStream = new Subject()
  const s = stream
    .pipe(split())
    .on('data', (line) => {
      s.pause()
      source.next({
        line: line
      })
    })
    .on('error', (err) => {
      source.error(err)
    })
    .on('end', () => {
      source.complete()
    })
  continueStream.subscribe(() => {
    s.resume()
  })
  return {
    text$: source,
    continue$: continueStream
  }
}

// this is my control function, basically what I'll
// use on my unit test or on my main function
export function processTxt (asciiStream: NodeJS.ReadableStream) {
  return new Promise((res, rej) => {
    const { text$, continue$ } = streamProcessing(asciiStream)
    text$
      .finally(() => {
        console.log('ended')
        res('ended')
      })
      .subscribe((lineOfText) => {
        console.log(lineOfText)
        continue$.next()
      })
  })
}
  • Can you clarify some assertions you're making in your question? In particular, I'd like to know what "using Subject too much" means, and what you consider "better" and "possible." Try to be specific; "best practice" is merely a synonym for "better," and anything is possible given enough time, money and effort. – Robert Harvey Jan 22 '18 at 17:54
1

I found the solution that only uses 1 Subject for the pause.

It looks something like this:

At some point I decided it would be nice to pause, resume the stream in the same function and be very very very careful where I actually call it. This is essential, most of the issues I had where because I was pausing, resuming in the wrong places.

enum StreamEventType {
  PAUSE,
  RESUME
}

function pauseStream (s: NodeJS.ReadableStream,
                      flag: StreamEventType) {
  switch (flag) {
    case StreamEventType.PAUSE:
      s.pause()
      break
    case StreamEventType.RESUME:
      s.resume()
      break
  }
}

I return both my text stream, and my pause observable (Subject) like follows:

import { split, map } from 'event-stream'

function streamProcessing (stream: NodeJS.ReadableStream) {
 const playPauseStream = new Subject()
  const sourceObservable = Observable.create((observer) => {
    const s = stream
      .pipe(split())
      .on('data', (line) => { observer.next(line) })
      .on('error', (err) => { observer.error(err) })
      .on('end', () => { observer.complete() })
    playPauseStream.subscribe(f => pauseStream(s, (f as StreamEventType)))
  })
  return {
    text$: sourceObservable,
    playPause$: playPauseStream
  }
}

The Pause/Resume (❚❚ ►) Observable is more essential (and delicate) than what you expect.

To consume this stream I have 2 Use Cases, one is XML and the other One is just \fpage breaks. The latter is easier so I did that first. We need something that will retrieve the text line by line, check for \f and break up accordingly, and just return the playPause switch with buffer obtained. Before publishing the buffer we will save to somewhere in the world we pause the stream.

One page of text is not really that much of a concern 'memory-wise'. We will create a function that does this with the argument observer, and then we will use that observer to create an observable:

function documentHandlers (asciiStream: NodeJS.ReadableStream) {
  const { text$, playPause$ } = streamProcessing(asciiStream)
  function obs (observer) {
    let buffer: string[] = []
    text$
      .finally(() => { observer.complete() })
      .catch((err) => Observable.throw(err))
      .subscribe((l) => {
        const line = (l as string)
        const firstChar = line.charAt(0)
        if (firstChar !== '\f' || buffer.length === 0) {
          buffer.push(line)
        } else {
          const bf = buffer.slice()
          buffer = [line]
          playPause$.next(StreamEventType.PAUSE)
          observer.next(bf)
        }
      })
  }
  return {
    documentObserver: obs,
    playPauseObservable: playPause$
  }
}

Now time to create the observable. Pay attention to the argument cbPromise, which returns a promise, once that promise is resolved then resume the stream:

const documentStream = (asciiStream, cbPromise) => {
  const { documentObserver, playPauseObservable } = documentHandlers(asciiStream)
  const document$ = Observable.create(documentObserver)
    .map(document =>
      Observable.fromPromise(cbPromise(document)
        .then(_ => {
          playPauseObservable.next(StreamEventType.RESUME)
          return true
        })
        .catch(e => false)
      )
    )
  return document$
}

This finally takes us full circle, now we implement one Function that Unit Tests this behavior (or Integrate Tests?):

  • Read File from Disk, Save to Disk
  • Read File from Local S3, Save to Local S3
  • Read File from S3, Save to S3

The reason why it was so important to break it up is that we need to throttle our S3 saving to less than 300 requests per second, this thing can handle thousands per second and My implementation does not have to care about this. Furthermore my unit tests can:

  • Check if the thing throttles requests to 300 per second or less.
  • Check if the files output are correct and look as expected.

We also don't want to do a lot of setup to do this tests, so in my unit tests I do the function that will save to disk, local S3, and web S3 with the following signature:

const processFileTest = (document: string[]) => {
  return new Promise((res, rej) => {
    // save to disk right here
    res()
  })
}

Then pass that to my function and subscribe to execute the process on the stream:

const documentObservable: Observable<any> = documentStream(stream, processFileTest).subscribe()

I really like doing it this way because it gives me control flexibility over the stream. Please feel free to leave comments.

If you are wondering What the hell is the stream, in the case of a local file is the following:

const stream = createReadStream('C:\Some Path\To File.txt')
      .setEncoding('ascii')

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.