Node.JS Streams examples

streams1.coffee

util = require 'util'

printArgs = (method) ->
  (args...) -> console.log method, util.inspect(args)



# Read a file and redirect it to standart output

fs = require 'fs'
stream = fs.createReadStream(__filename)
# stream.pipe(process.stdout)
# stream =====> stdout
#         pipe

console.log stream.constructor.name
# -> ReadStream

# What is the nature of process.stdout?
# Let's call pipe method with null argument.
# Each error give us a missing method:
# stream.pipe null
# pipe needs an argument with all theese method.
stream.pipe
  on: printArgs('on')
  once: printArgs('once')
  emit: printArgs('emit')
  write: printArgs('write')
  end: printArgs('end')

console.log process.stdout.constructor.name
# -> WriteStream

# The big picture:
# ReadStream =====> WriteStream
#             pipe

stream.on 'close', -> console.log 'ReadStream closed'
stream.pipe(process.stdout)

streams2.coffee

Readable = require('stream').Readable
util = require 'util'

Counter = (opt) ->
  Readable.call(this, opt)
  this._max = 1000
  this._index = 1

util.inherits Counter, Readable
# Other words to say this:
# class Counter extends Readable

Counter.prototype._read = ->
  i = this._index++
  if i > this._max
    this.push null
  else
    str = '' + i
    buf = new Buffer(str, 'ascii')
    this.push buf

# In coffee you can also write this:

class Counter extends Readable

  constructor: (opt) ->
    super opt
    @_max = 1000
    @_index = 1

  _read: ->
    i = @_index++
    if i > @_max
      @push null
    else
      str = "#{i}"
      buf = new Buffer str, 'ascii'
      @push buf

# To run:
#c = new Counter()
#c.pipe process.stdout


# Let's write a stream that takes a string as arguments and stream it to stdout

class StringStream extends Readable

  constructor: (@str, opt) ->
    super opt

  _read: ->
    @push @str
    @push null

s = new StringStream "toto"
s.pipe(process.stdout)

# That's it.

streams3.coffee

Readable = require('stream').Readable
Writable = require('stream').Writable
Transform = require('stream').Transform


class StringStream extends Readable

  constructor: (@str, opt) ->
    super opt

  _read: ->
    @push @str
    @push null

# Let's pipe this stream on a custom WriteStream.
class PrefixStream extends Writable
  
  constructor: (@prefix, opt) ->
    super opt

  _write: (chunk, encoding, callback) ->
    console.log "#{@prefix}#{chunk}"


r = new StringStream "world!"
w = new PrefixStream "Hello, "
#r.pipe w

# Awesome, but not neat.
# _write prefixes the chunk and write it with console.log: not that good.

# Let's use a Transfom instead of a Writable.

class Prefixer extends Transform

  constructor: (@prefix, opt) ->
    super opt

  _transform: (chunk, encoding, callback) ->
    @push "#{@prefix}#{chunk}"


# And finally:
t = new Prefixer "Hello, "
r.pipe(t).pipe(process.stdout)

# Cleaner, neater.


require 'event-stream'