Code files

Chapter04/creating-readable-writable-streams/composing-duplex-streams/index.js

@@ -0,0 +1,19 @@
+'use strict'
+
+const from = require('from2')
+const to = require('to2')
+const duplexify = require('duplexify')
+
+const rs = from(() => {
+  rs.push(Buffer.from('Hello, World!'))
+  rs.push(null)
+})
+
+const ws = to((data, enc, cb) => {
+  console.log(`Data written: ${data.toString()}`)
+  cb()
+})
+
+const stream = duplexify(ws, rs)
+
+stream.pipe(stream)

Chapter04/creating-readable-writable-streams/composing-duplex-streams/package.json

@@ -0,0 +1,16 @@
+{
+  "name": "composing-duplex-streams",
+  "version": "1.0.0",
+  "description": "",
+  "main": "index.js",
+  "scripts": {
+    "test": "echo \"Error: no test specified\" && exit 1"
+  },
+  "keywords": [],
+  "author": "",
+  "license": "ISC",
+  "dependencies": {
+    "from2": "^2.3.0",
+    "to2": "^1.0.0"
+  }
+}

Chapter04/creating-readable-writable-streams/core-streams/index.js

@@ -0,0 +1,19 @@
+'use strict'
+
+const { Readable, Writable } = require('readable-stream')
+
+const rs = Readable({
+  read: () => {
+    rs.push(Buffer.from('Hello, World!'))
+    rs.push(null) 
+  }
+})
+
+const ws = Writable({
+  write: (data, enc, cb) => {
+    console.log(`Data written: ${data.toString()}`)
+    cb()
+  }
+})
+
+rs.pipe(ws)

Chapter04/creating-readable-writable-streams/from2-to2-streams/index.js

@@ -0,0 +1,20 @@
+'use strict'
+
+const from = require('from2')
+const to = require('to2')
+
+const rs = from(() => {
+  rs.push(Buffer.from('Hello, World!'))
+  rs.push(null) 
+})
+
+// rs.on('data', (data) => {
+//   console.log(data.toString()) 
+// })
+
+const ws = to((data, enc, cb) => {
+  console.log(`Data written: ${data.toString()}`)
+  cb()
+})
+
+rs.pipe(ws)

Chapter04/creating-readable-writable-streams/from2-to2-streams/package.json

@@ -0,0 +1,16 @@
+{
+  "name": "from2-to2-streams",
+  "version": "1.0.0",
+  "description": "",
+  "main": "index.js",
+  "scripts": {
+    "test": "echo \"Error: no test specified\" && exit 1"
+  },
+  "keywords": [],
+  "author": "",
+  "license": "ISC",
+  "dependencies": {
+    "from2": "^2.3.0",
+    "to2": "^1.0.0"
+  }
+}

Chapter04/creating-readable-writable-streams/readable-flow-control/expected-behavior.js

@@ -0,0 +1,16 @@
+'use strict'
+
+const from = require('from2')
+const rs = from((size, cb) => {
+  setTimeout(() => {
+    rs.push('Data 0')
+    setTimeout(() => {
+      rs.push('Data 1')
+      cb()
+    }, 50)
+  }, 100)
+})
+
+rs.on('data', (data) => {
+  console.log(data.toString())
+})

Chapter04/creating-readable-writable-streams/readable-flow-control/package.json

@@ -0,0 +1,16 @@
+{
+  "name": "readable-flow-control",
+  "version": "1.0.0",
+  "description": "",
+  "main": "index.js",
+  "scripts": {
+    "test": "echo \"Error: no test specified\" && exit 1"
+  },
+  "keywords": [],
+  "author": "",
+  "license": "ISC",
+  "dependencies": {
+    "from2": "^2.3.0",
+    "readable-stream": "^2.2.6"
+  }
+}

Chapter04/creating-readable-writable-streams/readable-flow-control/undefined-behavior.js

@@ -0,0 +1,18 @@
+'use strict'
+
+// WARNING: DOES NOT WORK AS EXPECTED
+const { Readable } = require('readable-stream')
+const rs = Readable({
+  read: () => {
+    setTimeout(() => {
+      rs.push('Data 0')
+      setTimeout(() => {
+        rs.push('Data 1')
+      }, 50)
+    }, 100)
+  }
+})
+
+rs.on('data', (data) => {
+  console.log(data.toString())
+})

Chapter04/creating-transform-streams/core-transform-streams/classical.js

@@ -0,0 +1,13 @@
+'use strict'
+
+const { Transform } = require('readable-stream')
+
+class MyTransform extends Transform {
+  _transform (chunk, enc, cb) {
+    cb(null, chunk.toString().toUpperCase())
+  }
+}
+
+const upper = new MyTransform()
+
+process.stdin.pipe(upper).pipe(process.stdout)

Chapter04/creating-transform-streams/core-transform-streams/modern.js

@@ -0,0 +1,11 @@
+'use strict'
+
+const { Transform } = require('readable-stream')
+
+const upper = Transform({
+  transform: (chunk, enc, cb) => {
+    cb(null, chunk.toString().toUpperCase())
+  }
+})
+
+process.stdin.pipe(upper).pipe(process.stdout)

Chapter04/creating-transform-streams/core-transform-streams/package.json

@@ -0,0 +1,12 @@
+{
+  "name": "core-transform-streams",
+  "version": "1.0.0",
+  "description": "",
+  "main": "index.js",
+  "scripts": {
+    "test": "echo \"Error: no test specified\" && exit 1"
+  },
+  "keywords": [],
+  "author": "",
+  "license": "ISC"
+}

Chapter04/creating-transform-streams/core-transform-streams/prototypal.js

@@ -0,0 +1,18 @@
+'use strict'
+
+const stream = require('readable-stream')
+const util = require('util')
+
+function MyTransform(opts) {
+  stream.Transform.call(this, opts)
+}
+
+util.inherits(MyTransform, stream.Transform)
+
+MyTransform.prototype._transform = function (chunk, enc, cb) {
+  cb(null, chunk.toString().toUpperCase())
+}
+
+const upper = new MyTransform()
+
+process.stdin.pipe(upper).pipe(process.stdout)

Chapter04/creating-transform-streams/object-streams/core.js

@@ -0,0 +1,15 @@
+'use strict'
+
+const { Transform } = require('readable-stream')
+const { serialize } = require('ndjson')
+
+const xyz = Transform({
+  objectMode: true,
+  transform: ({x, y}, enc, cb) => { cb(null, {z: x + y}) } 
+})
+
+xyz.pipe(serialize()).pipe(process.stdout)
+
+xyz.write({x: 199, y: 3})
+
+xyz.write({x: 10, y: 12})

Chapter04/creating-transform-streams/object-streams/index.js

@@ -0,0 +1,14 @@
+'use strict'
+
+const through = require('through2')
+const { serialize } = require('ndjson')
+
+const xyz = through.obj(({x, y}, enc, cb) => {
+  cb(null, {z: x + y})
+})
+
+xyz.pipe(serialize()).pipe(process.stdout)
+
+xyz.write({x: 199, y: 3})
+
+xyz.write({x: 10, y: 12})

Chapter04/creating-transform-streams/object-streams/package.json

@@ -0,0 +1,17 @@
+{
+  "name": "object-streams",
+  "version": "1.0.0",
+  "description": "",
+  "main": "index.js",
+  "scripts": {
+    "test": "echo \"Error: no test specified\" && exit 1"
+  },
+  "keywords": [],
+  "author": "",
+  "license": "ISC",
+  "dependencies": {
+    "ndjson": "^1.5.0",
+    "readable-stream": "^2.2.6",
+    "through2": "^2.0.3"
+  }
+}

Chapter04/creating-transform-streams/through-streams/index.js

@@ -0,0 +1,9 @@
+'use strict'
+
+const through = require('through2')
+
+const upper = through((chunk, enc, cb) => {
+  cb(null, chunk.toString().toUpperCase())
+})
+
+process.stdin.pipe(upper).pipe(process.stdout)

Chapter04/creating-transform-streams/through-streams/package.json

@@ -0,0 +1,15 @@
+{
+  "name": "through-streams",
+  "version": "1.0.0",
+  "description": "",
+  "main": "index.js",
+  "scripts": {
+    "test": "echo \"Error: no test specified\" && exit 1"
+  },
+  "keywords": [],
+  "author": "",
+  "license": "ISC",
+  "dependencies": {
+    "through2": "^2.0.3"
+  }
+}

Chapter04/decoupling-io/backpressure/backpressure-with-pipe.js

@@ -0,0 +1,22 @@
+'use strict'
+
+const { Readable, Writable } = require('readable-stream')
+
+var i = 20
+
+const rs = Readable({
+  read: (size) => {
+    setImmediate(function () {
+      rs.push(i-- ? Buffer.alloc(size) : null)
+    })
+  }
+})
+
+const ws = Writable({
+  write: (chunk, enc, cb) => {
+    console.log(ws._writableState.length)
+    setTimeout(cb, 1)
+  }
+})
+
+rs.pipe(ws)

Chapter04/decoupling-io/backpressure/direct-write-no-backpressure.js

@@ -0,0 +1,22 @@
+'use strict'
+
+const { Readable, Writable } = require('readable-stream')
+
+var i = 20
+
+const rs = Readable({
+  read: (size) => {
+    setImmediate(function () {
+      rs.push(i-- ? Buffer.alloc(size) : null)
+    })
+  }
+})
+
+const ws = Writable({
+  write: (chunk, enc, cb) => {
+    console.log(ws._writableState.length)
+    setTimeout(cb, 1)
+  }
+})
+
+rs.on('data', (chunk) => ws.write(chunk))

Chapter04/decoupling-io/backpressure/direct-write-with-backpressure-pull.js

@@ -0,0 +1,40 @@
+'use strict'
+
+const { Readable, Writable } = require('readable-stream')
+
+var i = 20
+
+const rs = Readable({
+  read: (size, cb) => {
+    setImmediate(function () {
+      rs.push(i-- ? Buffer.alloc(size) : null)
+    })
+  }
+})
+
+const ws = Writable({
+  write: (chunk, enc, cb) => {
+    console.log(ws._writableState.length)
+    setTimeout(cb, 1)
+  }
+})
+
+function write (chunk, cb) {
+  const writable = ws.write(chunk)
+  if (writable === false) {
+    ws.once('drain', cb)
+    return
+  }
+  process.nextTick(cb)
+}
+
+function read () {
+  const chunk = rs.read()
+  if (chunk === null) {
+    rs.once('readable', read)
+    return
+  }
+  write(chunk, read)
+}
+
+rs.once('readable', read)

Chapter04/decoupling-io/backpressure/direct-write-with-backpressure.js

@@ -0,0 +1,28 @@
+'use strict'
+
+const { Readable, Writable } = require('readable-stream')
+
+var i = 20
+
+const rs = Readable({
+  read: (size) => {
+    setImmediate(function () {
+      rs.push(i-- ? Buffer.alloc(size) : null)
+    })
+  }
+})
+
+const ws = Writable({
+  write: (chunk, enc, cb) => {
+    console.log(ws._writableState.length)
+    setTimeout(cb, 1)
+  }
+})
+
+rs.on('data', (chunk) => {
+  const writable = ws.write(chunk)
+  if (writable === false) { 
+    rs.pause()
+    ws.once('drain', () => rs.resume())
+  }
+})

Chapter04/decoupling-io/ping-protocol-stream/index.js

@@ -0,0 +1,22 @@
+'use strict'
+
+const through = require('through2')
+const split = require('split2')
+const pumpify = require('pumpify')
+
+function pingProtocol() {
+  const ping = /Ping:\s+(.*)/
+  const protocol = through(each)
+
+  function each (line, enc, cb) {
+    if (ping.test(line)) {
+      cb(null, `Pong: ${line.toString().match(ping)[1]}\n`)
+      return
+    }
+    cb(null, 'Not Implemented\n')
+  }
+
+  return pumpify(split(), protocol)
+}
+
+module.exports = pingProtocol

Chapter04/decoupling-io/ping-protocol-stream/package.json

@@ -0,0 +1,17 @@
+{
+  "name": "foo-protocol-stream",
+  "version": "1.0.0",
+  "description": "",
+  "main": "index.js",
+  "scripts": {
+    "test": "echo \"Error: no test specified\" && exit 1"
+  },
+  "keywords": [],
+  "author": "",
+  "license": "ISC",
+  "dependencies": {
+    "pumpify": "^1.3.5",
+    "split2": "^2.1.1",
+    "through2": "^2.0.3"
+  }
+}

Chapter04/decoupling-io/stream-destruction/index.js

@@ -0,0 +1,25 @@
+'use strict'
+
+var from = require('from2')
+
+function createInfiniteTickStream () {
+  var tick = 0
+  return from.obj((size, cb) => {
+    setImmediate(() => cb(null, {tick: tick++}))
+  })
+}
+
+var stream = createInfiniteTickStream()
+
+stream.on('data', (data) => {
+  console.log(data)
+})
+
+stream.on('close', () => {
+  console.log('(stream destroyed)')
+})
+
+setTimeout(() => {
+  stream.destroy()
+}, 1000)
+

Chapter04/decoupling-io/stream-destruction/package.json

@@ -0,0 +1,15 @@
+{
+  "name": "stream-destruction",
+  "version": "1.0.0",
+  "description": "",
+  "main": "index.js",
+  "scripts": {
+    "test": "echo \"Error: no test specified\" && exit 1"
+  },
+  "keywords": [],
+  "author": "",
+  "license": "ISC",
+  "dependencies": {
+    "from2": "^2.3.0"
+  }
+}

Chapter04/decoupling-io/tcp-server/index.js

@@ -0,0 +1,17 @@
+'use strict'
+
+const net = require('net')
+const pump = require('pump')
+const ping = require('../ping-protocol-stream')
+
+const server = net.createServer((socket) => {
+  const protocol = ping()
+  pump(socket, protocol, socket, closed)
+})
+
+function closed (err) {
+  if (err) console.error('connection closed with error', err)
+  else console.log('connection closed')
+}
+
+server.listen(3000)

Chapter04/decoupling-io/tcp-server/package.json

@@ -0,0 +1,15 @@
+{
+  "name": "tcp-server",
+  "version": "1.0.0",
+  "description": "",
+  "main": "index.js",
+  "scripts": {
+    "test": "echo \"Error: no test specified\" && exit 1"
+  },
+  "keywords": [],
+  "author": "",
+  "license": "ISC",
+  "dependencies": {
+    "pump": "^1.0.2"
+  }
+}

Chapter04/piping-streams-in-production/big-file-server/index.js

@@ -0,0 +1,19 @@
+'use strict'
+
+const fs = require('fs')
+const http = require('http')
+const pump = require('pump')
+
+const server = http.createServer((req, res) => {
+  const stream = fs.createReadStream('big.file')
+  pump(stream, res, done)
+})
+
+function done (err) {
+  if (err) {
+    return console.error('File was not fully streamed to the user', err)
+  }
+  console.log('File was fully streamed to the user')
+}
+
+server.listen(3000)

Chapter04/piping-streams-in-production/big-file-server/package.json

@@ -0,0 +1,15 @@
+{
+  "name": "big-file-server",
+  "version": "1.0.0",
+  "description": "",
+  "main": "index.js",
+  "scripts": {
+    "test": "echo \"Error: no test specified\" && exit 1"
+  },
+  "keywords": [],
+  "author": "",
+  "license": "ISC",
+  "dependencies": {
+    "pump": "^1.0.2"
+  }
+}

Chapter04/piping-streams-in-production/pumpified-pipeline/index.js

@@ -0,0 +1,25 @@
+'use strict'
+
+const { createGzip } = require('zlib')
+const { createCipher } = require('crypto')
+const pumpify = require('pumpify')
+const base64 = require('base64-encode-stream')
+
+function pipeline () {
+  const stream1 = createGzip()
+  const stream2 = createCipher('aes192', 'secretz')
+  const stream3 = base64()
+  return pumpify(stream1, stream2, stream3)
+}
+
+const pipe = pipeline()
+
+pipe.end('written to stream1')
+
+pipe.on('data', (data) => {
+  console.log('stream3 says: ', data.toString())
+})
+
+pipe.on('finish', () => {
+  console.log('all data was successfully flushed to stream3')
+})

Chapter04/piping-streams-in-production/pumpified-pipeline/package.json

@@ -0,0 +1,16 @@
+{
+  "name": "pumpified-pipeline",
+  "version": "1.0.0",
+  "description": "",
+  "main": "index.js",
+  "scripts": {
+    "test": "echo \"Error: no test specified\" && exit 1"
+  },
+  "keywords": [],
+  "author": "",
+  "license": "ISC",
+  "dependencies": {
+    "base64-encode-stream": "^1.0.0",
+    "pumpify": "^1.3.5"
+  }
+}

Chapter04/processing-big-data/infinite-read/index.js

@@ -0,0 +1,9 @@
+'use strict'
+
+const rs = fs.createReadStream('/dev/urandom')
+var size = 0
+
+rs.on('data', (data) => {
+  size += data.length
+  console.log('File size:', size)
+})

Chapter04/processing-big-data/self-read-pull/index.js

@@ -0,0 +1,17 @@
+'use strict'
+
+const fs = require('fs')
+const rs = fs.createReadStream(__filename)
+
+rs.on('readable', () => {
+  var data = rs.read()
+  while (data !== null) {
+    console.log('Read chunk:', data)
+    data = rs.read()
+  }
+})
+
+rs.on('end', () => {
+  console.log('No more data')
+})
+

Chapter04/processing-big-data/self-read/index.js

@@ -0,0 +1,12 @@
+'use strict'
+const fs = require('fs')
+const rs = fs.createReadStream(__filename)
+
+rs.on('data', (data) => {
+  console.log('Read chunk:', data)
+})
+
+rs.on('end', () => {
+  console.log('No more data')
+})
+

Chapter04/using-the-pipe-method/pipe-without-end/broken.js

@@ -0,0 +1,10 @@
+const net = require('net')
+const fs = require('fs')
+
+net.createServer((socket) => {
+  const content = fs.createReadStream(__filename)
+  content.pipe(socket)
+  content.on('end', () => {
+    socket.end('\n======= Footer =======\n')
+  })
+}).listen(3000)

Chapter04/using-the-pipe-method/pipe-without-end/index.js

@@ -0,0 +1,10 @@
+const net = require('net')
+const fs = require('fs')
+
+net.createServer((socket) => {
+  const content = fs.createReadStream(__filename)
+  content.pipe(socket, {end: false})
+  content.on('end', () => {
+    socket.end('\n======= Footer =======\n')
+  })
+}).listen(3000)

Chapter04/using-the-pipe-method/piper/index.js

@@ -0,0 +1,20 @@
+'use strict'
+
+const zlib = require('zlib')
+const map = require('tar-map-stream')
+const decompress = zlib.createGunzip()
+const whoami = process.env.USER || process.env.USERNAME
+const convert = map((header) => {
+  header.uname = whoami
+  header.mtime = new Date()
+  header.name = header.name.replace('node-v0.1.100', 'edon-v0.0.0')
+  return header
+})
+const compress = zlib.createGzip()
+
+process.stdin
+  .pipe(decompress)
+  .pipe(convert)
+  .pipe(compress)
+  .pipe(process.stdout)
+

Chapter04/using-the-pipe-method/piper/package.json

@@ -0,0 +1,15 @@
+{
+  "name": "piper",
+  "version": "1.0.0",
+  "description": "",
+  "main": "index.js",
+  "scripts": {
+    "test": "echo \"Error: no test specified\" && exit 1"
+  },
+  "keywords": [],
+  "author": "",
+  "license": "ISC",
+  "dependencies": {
+    "tar-map-stream": "^1.0.0"
+  }
+}