It's about time to embrace Node.js streams

Let's shed some light on Node.js streams! With very practical examples we'll learn when and how to use streams in Node.js and in the Browser. At the end of this talk, you will be able to take full advantage from Streams and write elegant JavaScript applications that are much more memory efficient! This talk has the ambition to shed some light on Node.js streams. It will provide very practical examples about when and how you should be using streams and it will progress into more in-depth stream topics. Don't worry, if you want to use Node.js streams in the browser, I'll have you covered as well! At the end of this talk you will be able to understand streams better and not fear them anymore! In my experience with JavaScript/Node.js devs I find out that Streams is a topic still mostly unexplored and sometimes even feared. This is probably one of the most beautiful features of Node.js and it can be adopted also in the browser. Once you grasp the fundamentals, you'll be able to solve some ordinary programming challenges in a much more elegant and efficient way, so this is a very interesting topic. In this talk, I will cover the following topics: Streams: when and how; Different types of streams; Built-in and custom streams; Composability; Stream utils; Streams in the browser.

Transcript

1. Luciano Mammino (@loige) IT’S ABOUT TIME TO IT’S ABOUT TIME TO EMBRACE STREAMS EMBRACE STREAMS IT’S ABOUT TIME TO IT’S ABOUT TIME TO EMBRACE STREAMS EMBRACE STREAMS Luciano Mammino (@loige) 05/03/2019 loige.link/streams-dub 1
2. loige.link/streams-dub code: loige.link/streams-examples 2
3. HELLO, I AM LUCIANO! HELLO, I AM LUCIANO! Cloud Architect Let's connect! Blog: Twitter: GitHub: loige.co @loige @lmammino loige.link/node-patterns fstack.link with @mariocasciaro with @andreaman87 3
4. AGENDA AGENDA 01. Buffers VS Streams 02. Stream types & APIs 03. Pipe() 04. Streams utilities 05. Writing custom streams 06. Streams in the Browser @loige 4
5. 01. BUFFERS VS 01. BUFFERS VS STREAMS STREAMS @loige 5
6. BUFFER BUFFER: DATA STRUCTURE TO STORE AND : DATA STRUCTURE TO STORE AND TRANSFER ARBITRARY BINARY DATA TRANSFER ARBITRARY BINARY DATA @loige *Note that this is loading all the content of the file in memory * 6
7. STREAM STREAM: ABSTRACT INTERFACE FOR : ABSTRACT INTERFACE FOR WORKING WITH STREAMING DATA WORKING WITH STREAMING DATA @loige *It does not load all the data straight away * 7
8. LET'S SOLVE THIS PROBLEM LET'S SOLVE THIS PROBLEM 1. Read the content of a file 2. copy it to another file* * cp in Node.js @loige 8
9. THE BUFFER WAY THE BUFFER WAY // buffer-copy.js const { readFileSync, writeFileSync } = require('fs') const [,, src, dest] = process.argv const content = readFileSync(src) writeFileSync(dest, content) @loige 9
10. THE STREAM WAY THE STREAM WAY // stream-copy.js const { createReadStream, createWriteStream } = require('fs') const [,, src, dest] = process.argv const srcStream = createReadStream(src) const destStream = createWriteStream(dest) srcStream.on('data', (data) => destStream.write(data)) @loige * Careful: this implementation is not optimal * 10
11. MEMORY COMPARISON (~600MB FILE) MEMORY COMPARISON (~600MB FILE) node ­­inspect­brk buffer­copy.js assets/poster.psd ~/Downloads/poster.psd @loige 11
12. MEMORY COMPARISON (~600MB FILE) MEMORY COMPARISON (~600MB FILE) node ­­inspect­brk stream­copy.js assets/poster.psd ~/Downloads/poster.psd @loige 12
13. LET'S TRY WITH A BIG FILE (~10GB) LET'S TRY WITH A BIG FILE (~10GB) @loige 13
14. LET'S TRY WITH A BIG FILE (~10GB) LET'S TRY WITH A BIG FILE (~10GB) node ­­inspect­brk stream­copy.js assets/the­matrix­hd.mkv ~/Downloads/the­matrix­hd.mkv @loige 14
15. IF BYTES IF BYTES WERE WERE BLOCKS... BLOCKS... @loige 15
16. BIG BUFFER APPROACH BIG BUFFER APPROACH @loige 16
17. STREAMING APPROACH STREAMING APPROACH @loige 17
18. STREAMS VS BUFFERS STREAMS VS BUFFERS Streams keep a low memory footprint even with large amounts of data Streams allows you to process data as soon as it arrives Stream processing generally does not block the event loop @loige 18
19. 03. STREAM TYPES 03. STREAM TYPES & APIS & APIS @loige 19
20. ALL STREAMS ARE ALL STREAMS ARE EVENT EMITTERS EVENT EMITTERS A stream instance is an object that emits events when its internal state changes, for instance: s.on('readable', () => {}) // ready to be consumed s.on('data', (chunk) => {}) // new data is available s.on('error', (err) => {}) // some error happened s.on('end', () => {}) // no more data available The events available depend from the type of stream @loige 20
21. WRITABLE WRITABLE STREAMS STREAMS A writable stream is an abstraction that allows to write data over a destination Examples: fs writeStream process.stdout, process.stderr HTTP request (client-side) HTTP response (server-side) AWS S3 PutObject (body parameter) @loige 21
22. WRITABLE STREAMS - WRITABLE STREAMS - METHODS METHODS writable.write(chunk, [encoding], [callback]) writable.end([chunk], [encoding], [callback]) @loige 22
23. writable.on('drain') writable.on('close') writable.on('finish') writable.on('error', (err) => {}) WRITABLE STREAMS - WRITABLE STREAMS - EVENTS EVENTS @loige 23
24. // writable-http-request.js const http = require('http') const req = http.request( { hostname: 'enx6b07hdu6cs.x.pipedream.net', method: 'POST' }, resp => { console.log(`Server responded with "${resp.statusCode}"`) } ) req.on('finish', () => console.log('request sent')) req.on('close', () => console.log('Connection closed')) req.on('error', err => console.error(`Request failed: ${err}`)) req.write('writing some content...\n') req.end('last write & close the stream') @loige 24
25. // writable-http-request.js const http = require('http') const req = http.request( { hostname: 'enx6b07hdu6cs.x.pipedream.net', method: 'POST' }, resp => { console.log(`Server responded with "${resp.statusCode}"`) } ) req.on('finish', () => console.log('request sent')) req.on('close', () => console.log('Connection closed')) req.on('error', err => console.error(`Request failed: ${err}`)) req.write('writing some content...\n') req.end('last write & close the stream') @loige 24
26. // writable-http-request.js const http = require('http') const req = http.request( { hostname: 'enx6b07hdu6cs.x.pipedream.net', method: 'POST' }, resp => { console.log(`Server responded with "${resp.statusCode}"`) } ) req.on('finish', () => console.log('request sent')) req.on('close', () => console.log('Connection closed')) req.on('error', err => console.error(`Request failed: ${err}`)) req.write('writing some content...\n') req.end('last write & close the stream') @loige 24
27. // writable-http-request.js const http = require('http') const req = http.request( { hostname: 'enx6b07hdu6cs.x.pipedream.net', method: 'POST' }, resp => { console.log(`Server responded with "${resp.statusCode}"`) } ) req.on('finish', () => console.log('request sent')) req.on('close', () => console.log('Connection closed')) req.on('error', err => console.error(`Request failed: ${err}`)) req.write('writing some content...\n') req.end('last write & close the stream') @loige 24
28. @loige 25
29. loige.link/writable-http-req @loige 26
30. BACKPRESSURE BACKPRESSURE When writing large amounts of data you should make sure you handle the stop write signal and the drain event loige.link/backpressure @loige 27
31. // stream-copy-safe.js const { createReadStream, createWriteStream } = require('fs') const [, , src, dest] = process.argv const srcStream = createReadStream(src) const destStream = createWriteStream(dest) srcStream.on('data', data => { const canContinue = destStream.write(data) if (!canContinue) { // we are overflowing the destination, we should pause srcStream.pause() // we will resume when the destination stream is drained destStream.once('drain', () => srcStream.resume()) } }) @loige 28
32. // stream-copy-safe.js const { createReadStream, createWriteStream } = require('fs') const [, , src, dest] = process.argv const srcStream = createReadStream(src) const destStream = createWriteStream(dest) srcStream.on('data', data => { const canContinue = destStream.write(data) if (!canContinue) { // we are overflowing the destination, we should pause srcStream.pause() // we will resume when the destination stream is drained destStream.once('drain', () => srcStream.resume()) } }) @loige 28
33. // stream-copy-safe.js const { createReadStream, createWriteStream } = require('fs') const [, , src, dest] = process.argv const srcStream = createReadStream(src) const destStream = createWriteStream(dest) srcStream.on('data', data => { const canContinue = destStream.write(data) if (!canContinue) { // we are overflowing the destination, we should pause srcStream.pause() // we will resume when the destination stream is drained destStream.once('drain', () => srcStream.resume()) } }) @loige 28
34. // stream-copy-safe.js const { createReadStream, createWriteStream } = require('fs') const [, , src, dest] = process.argv const srcStream = createReadStream(src) const destStream = createWriteStream(dest) srcStream.on('data', data => { const canContinue = destStream.write(data) if (!canContinue) { // we are overflowing the destination, we should pause srcStream.pause() // we will resume when the destination stream is drained destStream.once('drain', () => srcStream.resume()) } }) @loige 28
35. // stream-copy-safe.js const { createReadStream, createWriteStream } = require('fs') const [, , src, dest] = process.argv const srcStream = createReadStream(src) const destStream = createWriteStream(dest) srcStream.on('data', data => { const canContinue = destStream.write(data) if (!canContinue) { // we are overflowing the destination, we should pause srcStream.pause() // we will resume when the destination stream is drained destStream.once('drain', () => srcStream.resume()) } }) @loige 28
36. READABLE READABLE STREAMS STREAMS A readable stream represents a source from which data is consumed. Examples: fs readStream process.stdin HTTP response (client-side) HTTP request (server-side) AWS S3 GetObject (data field) It supports two modes for data consumption: flowing and paused (or non- flowing) mode. @loige 29
37. READABLE STREAMS - READABLE STREAMS - METHODS METHODS readable.read([size]) readable.pause() readable.resume() @loige 30
38. READABLE STREAMS - READABLE STREAMS - EVENTS EVENTS readable.on('readable') readable.on('data', (chunk) => {}) readable.on('end') readable.on('error', (err) => {}) @loige 31
39. READABLE STREAMS - READABLE STREAMS - FLOWING MODE FLOWING MODE Data is read from source automatically and chunks are emitted as soon as they are available. @loige 32
40. @loige 1 2 3 Source data Readable stream in flowing mode data listener READABLE STREAMS - READABLE STREAMS - FLOWING MODE FLOWING MODE Data is read from source automatically and chunks are emitted as soon as they are available. 33
41. @loige 1 2 3 Source data Readable stream in flowing mode Read data listener READABLE STREAMS - READABLE STREAMS - FLOWING MODE FLOWING MODE Data is read from source automatically and chunks are emitted as soon as they are available. 34
42. @loige 1 2 3 Source data Readable stream in flowing mode data listener data READABLE STREAMS - READABLE STREAMS - FLOWING MODE FLOWING MODE Data is read from source automatically and chunks are emitted as soon as they are available. 35
43. @loige 2 3 Source data Readable stream in flowing mode data listener Read READABLE STREAMS - READABLE STREAMS - FLOWING MODE FLOWING MODE Data is read from source automatically and chunks are emitted as soon as they are available. 36
44. @loige 2 3 Source data Readable stream in flowing mode data listener data READABLE STREAMS - READABLE STREAMS - FLOWING MODE FLOWING MODE Data is read from source automatically and chunks are emitted as soon as they are available. 37
45. @loige 3 Source data Readable stream in flowing mode data listener Read READABLE STREAMS - READABLE STREAMS - FLOWING MODE FLOWING MODE Data is read from source automatically and chunks are emitted as soon as they are available. 38
46. @loige 3 Source data Readable stream in flowing mode data listener data READABLE STREAMS - READABLE STREAMS - FLOWING MODE FLOWING MODE Data is read from source automatically and chunks are emitted as soon as they are available. 39
47. @loige Source data Readable stream in flowing mode Read data listener (end) READABLE STREAMS - READABLE STREAMS - FLOWING MODE FLOWING MODE Data is read from source automatically and chunks are emitted as soon as they are available. 40
48. @loige Source data Readable stream in flowing mode data listener end (end) When no more data is available, end is emitted. READABLE STREAMS - READABLE STREAMS - FLOWING MODE FLOWING MODE Data is read from source automatically and chunks are emitted as soon as they are available. 41
49. // count-emojis-flowing.js const { createReadStream } = require('fs') const { EMOJI_MAP } = require('emoji') // from npm const emojis = Object.keys(EMOJI_MAP) const file = createReadStream(process.argv[2]) let counter = 0 file.on('data', chunk => { for (let char of chunk.toString('utf8')) { if (emojis.includes(char)) { counter++ } } }) file.on('end', () => console.log(`Found ${counter} emojis`)) file.on('error', err => console.error(`Error reading file: ${err}`)) @loige 42
50. // count-emojis-flowing.js const { createReadStream } = require('fs') const { EMOJI_MAP } = require('emoji') // from npm const emojis = Object.keys(EMOJI_MAP) const file = createReadStream(process.argv[2]) let counter = 0 file.on('data', chunk => { for (let char of chunk.toString('utf8')) { if (emojis.includes(char)) { counter++ } } }) file.on('end', () => console.log(`Found ${counter} emojis`)) file.on('error', err => console.error(`Error reading file: ${err}`)) @loige 42
51. loige.link/st_patrick @loige 43
52. READABLE STREAMS - READABLE STREAMS - PAUSED MODE PAUSED MODE A consumer has to call the read method explicitly to read chunks of data from the stream. The stream sends a readable event to signal that new data is available. @loige 44
53. @loige Source data Readable stream in paused mode consumer 1 2 3 < > (internal buffer) READABLE STREAMS - READABLE STREAMS - PAUSED MODE PAUSED MODE A consumer has to call the read method explicitly to read chunks of data from the stream. The stream sends a readable event to signal that new data is available. 45
54. @loige Source data consumer 1 2 3 < > (internal buffer) Read Readable stream in paused mode READABLE STREAMS - READABLE STREAMS - PAUSED MODE PAUSED MODE A consumer has to call the read method explicitly to read chunks of data from the stream. The stream sends a readable event to signal that new data is available. 46
55. @loige Source data consumer 1 2 3 < > (internal buffer) readable Readable stream in paused mode READABLE STREAMS - READABLE STREAMS - PAUSED MODE PAUSED MODE A consumer has to call the read method explicitly to read chunks of data from the stream. The stream sends a readable event to signal that new data is available. 47
56. @loige Source data consumer 1 2 3 < > (internal buffer) Nothing happens until the consumer decides to read the data Readable stream in paused mode READABLE STREAMS - READABLE STREAMS - PAUSED MODE PAUSED MODE A consumer has to call the read method explicitly to read chunks of data from the stream. The stream sends a readable event to signal that new data is available. 48
57. < > @loige Source data consumer 1 2 3 (internal buffer) read() Readable stream in paused mode READABLE STREAMS - READABLE STREAMS - PAUSED MODE PAUSED MODE A consumer has to call the read method explicitly to read chunks of data from the stream. The stream sends a readable event to signal that new data is available. 49
58. < > @loige Source data consumer 2 3 (internal buffer) Readable stream in paused mode READABLE STREAMS - READABLE STREAMS - PAUSED MODE PAUSED MODE A consumer has to call the read method explicitly to read chunks of data from the stream. The stream sends a readable event to signal that new data is available. 50
59. < > @loige Source data consumer null 2 3 (internal buffer) read() Using read() with an empty buffer will return null (stop reading signal) Readable stream in paused mode READABLE STREAMS - READABLE STREAMS - PAUSED MODE PAUSED MODE A consumer has to call the read method explicitly to read chunks of data from the stream. The stream sends a readable event to signal that new data is available. 51
60. < > @loige Source data consumer 2 3 (internal buffer) Readable stream in paused mode READABLE STREAMS - READABLE STREAMS - PAUSED MODE PAUSED MODE A consumer has to call the read method explicitly to read chunks of data from the stream. The stream sends a readable event to signal that new data is available. 52
61. < > @loige Source data consumer 2 3 (internal buffer) Read Readable stream in paused mode READABLE STREAMS - READABLE STREAMS - PAUSED MODE PAUSED MODE A consumer has to call the read method explicitly to read chunks of data from the stream. The stream sends a readable event to signal that new data is available. 53
62. < > @loige Source data consumer 2 3 (internal buffer) readable Readable stream in paused mode READABLE STREAMS - READABLE STREAMS - PAUSED MODE PAUSED MODE A consumer has to call the read method explicitly to read chunks of data from the stream. The stream sends a readable event to signal that new data is available. 54
63. < > @loige Source data consumer 2 3 (internal buffer) Readable stream in paused mode READABLE STREAMS - READABLE STREAMS - PAUSED MODE PAUSED MODE A consumer has to call the read method explicitly to read chunks of data from the stream. The stream sends a readable event to signal that new data is available. 55
64. < > @loige Source data consumer 2 3 (internal buffer) read() Readable stream in paused mode READABLE STREAMS - READABLE STREAMS - PAUSED MODE PAUSED MODE A consumer has to call the read method explicitly to read chunks of data from the stream. The stream sends a readable event to signal that new data is available. 56
65. < > @loige Source data consumer 3 (internal buffer) Readable stream in paused mode READABLE STREAMS - READABLE STREAMS - PAUSED MODE PAUSED MODE A consumer has to call the read method explicitly to read chunks of data from the stream. The stream sends a readable event to signal that new data is available. 57
66. < > @loige Source data consumer 3 (internal buffer) Read Readable stream in paused mode READABLE STREAMS - READABLE STREAMS - PAUSED MODE PAUSED MODE A consumer has to call the read method explicitly to read chunks of data from the stream. The stream sends a readable event to signal that new data is available. 58
67. < > @loige Source data consumer 3 (internal buffer) readable Readable stream in paused mode READABLE STREAMS - READABLE STREAMS - PAUSED MODE PAUSED MODE A consumer has to call the read method explicitly to read chunks of data from the stream. The stream sends a readable event to signal that new data is available. 59
68. < > @loige Source data consumer 3 (internal buffer) Readable stream in paused mode READABLE STREAMS - READABLE STREAMS - PAUSED MODE PAUSED MODE A consumer has to call the read method explicitly to read chunks of data from the stream. The stream sends a readable event to signal that new data is available. 60
69. < > @loige Source data consumer 3 (internal buffer) read() Readable stream in paused mode READABLE STREAMS - READABLE STREAMS - PAUSED MODE PAUSED MODE A consumer has to call the read method explicitly to read chunks of data from the stream. The stream sends a readable event to signal that new data is available. 61
70. < > @loige Source data consumer (internal buffer) Readable stream in paused mode READABLE STREAMS - READABLE STREAMS - PAUSED MODE PAUSED MODE A consumer has to call the read method explicitly to read chunks of data from the stream. The stream sends a readable event to signal that new data is available. 62
71. < > @loige Source data consumer (internal buffer) Read (end) Readable stream in paused mode READABLE STREAMS - READABLE STREAMS - PAUSED MODE PAUSED MODE A consumer has to call the read method explicitly to read chunks of data from the stream. The stream sends a readable event to signal that new data is available. 63
72. < > @loige Source data consumer (internal buffer) (end) end Readable stream in paused mode READABLE STREAMS - READABLE STREAMS - PAUSED MODE PAUSED MODE A consumer has to call the read method explicitly to read chunks of data from the stream. The stream sends a readable event to signal that new data is available. 64
73. // count-emojis-paused.js const { createReadStream } = require('fs') const { EMOJI_MAP } = require('emoji') // from npm const emojis = Object.keys(EMOJI_MAP) const file = createReadStream(process.argv[2]) let counter = 0 file.on('readable', () => { let chunk while ((chunk = file.read()) !== null) { for (let char of chunk.toString('utf8')) { if (emojis.includes(char)) { counter++ } } } }) file.on('end', () => console.log(`Found ${counter} emojis`)) file.on('error', err => console.error(`Error reading file: ${err}`)) @loige 65
74. // count-emojis-paused.js const { createReadStream } = require('fs') const { EMOJI_MAP } = require('emoji') // from npm const emojis = Object.keys(EMOJI_MAP) const file = createReadStream(process.argv[2]) let counter = 0 file.on('readable', () => { let chunk while ((chunk = file.read()) !== null) { for (let char of chunk.toString('utf8')) { if (emojis.includes(char)) { counter++ } } } }) file.on('end', () => console.log(`Found ${counter} emojis`)) file.on('error', err => console.error(`Error reading file: ${err}`)) @loige 65
75. // count-emojis-paused.js const { createReadStream } = require('fs') const { EMOJI_MAP } = require('emoji') // from npm const emojis = Object.keys(EMOJI_MAP) const file = createReadStream(process.argv[2]) let counter = 0 file.on('readable', () => { let chunk while ((chunk = file.read()) !== null) { for (let char of chunk.toString('utf8')) { if (emojis.includes(char)) { counter++ } } } }) file.on('end', () => console.log(`Found ${counter} emojis`)) file.on('error', err => console.error(`Error reading file: ${err}`)) @loige 65
76. @loige 66
77. READABLE STREAMS READABLE STREAMS MODE SWITCH CONDITIONS MODE SWITCH CONDITIONS All readable streams are created in paused mode paused streams can be switched to flowing mode with: stream.on('data', () => {}) stream.resume() stream.pipe() flowing streams can switch back to paused with: stream.pause() stream.unpipe() for all attached streams @loige 67
78. READABLE STREAMS READABLE STREAMS FLOWING FLOWING VS VS PAUSED PAUSED Push VS Pull mental models Flowing is simpler to use Paused gives you more control on how data is consumed from the source Whichever you pick, stay consistent! @loige 68 . 1
79. BONUS MODE BONUS MODE Readable streams are also Async Iterators (Node.js 10+) Warning: still experimental @loige 68 . 2
80. // count-emojis-async-iterator.js const { createReadStream } = require('fs') const { EMOJI_MAP } = require('emoji') // from npm async function main () { const emojis = Object.keys(EMOJI_MAP) const file = createReadStream(process.argv[2]) let counter = 0 for await (let chunk of file) { for (let char of chunk.toString('utf8')) { if (emojis.includes(char)) { counter++ } } } console.log(`Found ${counter} emojis`) } main() @loige 68 . 3
81. // count-emojis-async-iterator.js const { createReadStream } = require('fs') const { EMOJI_MAP } = require('emoji') // from npm async function main () { const emojis = Object.keys(EMOJI_MAP) const file = createReadStream(process.argv[2]) let counter = 0 for await (let chunk of file) { for (let char of chunk.toString('utf8')) { if (emojis.includes(char)) { counter++ } } } console.log(`Found ${counter} emojis`) } main() @loige 68 . 3
82. Still experimental in Node.js 11 If you like this API and don't want to rely on an experimental core feature: 2ality.com/2018/04/async-iter-nodejs.html github.com/lorenzofox3/for-await @loige 68 . 4
83. OBJECT MODE OBJECT MODE Readable streams can emit objects if this mode is enabled // readable-timer.js const { Readable } = require('stream') const timerStream = new Readable({ objectMode: true, read () { this.push(new Date()) } }) timerStream.on('data', (currentDate) => { // prints the current second console.log(currentDate.getSeconds()) }) @loige 69
84. OBJECT MODE OBJECT MODE Readable streams can emit objects if this mode is enabled // readable-timer.js const { Readable } = require('stream') const timerStream = new Readable({ objectMode: true, read () { this.push(new Date()) } }) timerStream.on('data', (currentDate) => { // prints the current second console.log(currentDate.getSeconds()) }) @loige 69
85. OBJECT MODE OBJECT MODE Readable streams can emit objects if this mode is enabled // readable-timer.js const { Readable } = require('stream') const timerStream = new Readable({ objectMode: true, read () { this.push(new Date()) } }) timerStream.on('data', (currentDate) => { // prints the current second console.log(currentDate.getSeconds()) }) @loige This is an object 69
86. OBJECT MODE OBJECT MODE Readable streams can emit objects if this mode is enabled // readable-timer.js const { Readable } = require('stream') const timerStream = new Readable({ objectMode: true, read () { this.push(new Date()) } }) timerStream.on('data', (currentDate) => { // prints the current second console.log(currentDate.getSeconds()) }) @loige 69
87. OTHER TYPES OF STREAM OTHER TYPES OF STREAM Duplex Stream streams that are both Readable and Writable. (net.Socket) Transform Stream Duplex streams that can modify or transform the data as it is written and read. (zlib.createGzip(), crypto.createCipheriv()) @loige 70
88. ANATOMY OF A TRANSFORM STREAM ANATOMY OF A TRANSFORM STREAM 1. write data transform stream 3. read transformed data 2. transform the data (readable stream) (writable stream) @loige 71
89. GZIP EXAMPLE GZIP EXAMPLE 1. write data transform stream 3. read transformed data 2. transform the data (readable stream) (writable stream) @loige Uncompressed data Compressed data compress gzip.createGzip() 72
90. gzipStream.on('data', data => { const canContinue = destStream.write(data) if (!canContinue) { gzipStream.pause() destStream.once('drain', () => { gzipStream.resume() }) } }) gzipStream.on('end', () => { destStream.end() }) // TODO: handle errors! >:) // stream-copy-gzip.js const { createReadStream, createWriteStream } = require('fs') const { createGzip } = require('zlib') const [, , src, dest] = process.argv const srcStream = createReadStream(src) const gzipStream = createGzip() const destStream = createWriteStream(dest) srcStream.on('data', data => { const canContinue = gzipStream.write(data) if (!canContinue) { srcStream.pause() gzipStream.once('drain', () => { srcStream.resume() }) } }) srcStream.on('end', () => { // check if there's buffered data left const remainingData = gzipStream.read() if (remainingData !== null) { destStream.write() } gzipStream.end() }) @loige 73
91. gzipStream.on('data', data => { const canContinue = destStream.write(data) if (!canContinue) { gzipStream.pause() destStream.once('drain', () => { gzipStream.resume() }) } }) gzipStream.on('end', () => { destStream.end() }) // TODO: handle errors! >:) // stream-copy-gzip.js const { createReadStream, createWriteStream } = require('fs') const { createGzip } = require('zlib') const [, , src, dest] = process.argv const srcStream = createReadStream(src) const gzipStream = createGzip() const destStream = createWriteStream(dest) srcStream.on('data', data => { const canContinue = gzipStream.write(data) if (!canContinue) { srcStream.pause() gzipStream.once('drain', () => { srcStream.resume() }) } }) srcStream.on('end', () => { // check if there's buffered data left const remainingData = gzipStream.read() if (remainingData !== null) { destStream.write() } gzipStream.end() }) @loige 73
92. 03. PIPE() 03. PIPE() @loige 74
93. readable .pipe(tranform1) .pipe(transform2) .pipe(transform3) .pipe(writable) readable.pipe(writableDest) @loige Connects a readable stream to a writable stream A transform stream can be used as a destination as well It returns the destination stream allowing for a chain of pipes 75
94. // stream-copy-gzip-pipe.js const { createReadStream, createWriteStream } = require('fs') const { createGzip } = require('zlib') const [, , src, dest] = process.argv const srcStream = createReadStream(src) const gzipStream = createGzip() const destStream = createWriteStream(dest) srcStream .pipe(gzipStream) .pipe(destStream) @loige 76
95. // stream-copy-gzip-pipe.js const { createReadStream, createWriteStream } = require('fs') const { createGzip } = require('zlib') const [, , src, dest] = process.argv const srcStream = createReadStream(src) const gzipStream = createGzip() const destStream = createWriteStream(dest) srcStream .pipe(gzipStream) .pipe(destStream) @loige 76
96. // stream-copy-gzip-pipe.js const { createReadStream, createWriteStream } = require('fs') const { createGzip } = require('zlib') const [, , src, dest] = process.argv const srcStream = createReadStream(src) const gzipStream = createGzip() const destStream = createWriteStream(dest) srcStream .pipe(gzipStream) .pipe(destStream) @loige 76
97. readable .pipe(decompress) .pipe(decrypt) .pipe(convert) .pipe(encrypt) .pipe(compress) .pipe(writeToDisk) Setup complex pipelines with pipe @loige This is the most common way to use streams 77
98. readable .on('error', handleErr) .pipe(decompress) .on('error', handleErr) .pipe(decrypt) .on('error', handleErr) .pipe(convert) .on('error', handleErr) .pipe(encrypt) .on('error', handleErr) .pipe(compress) .on('error', handleErr) .pipe(writeToDisk) .on('error', handleErr) Handling errors (correctly) @loige 78
99. readable .on('error', handleErr) .pipe(decompress) .on('error', handleErr) .pipe(decrypt) .on('error', handleErr) .pipe(convert) .on('error', handleErr) .pipe(encrypt) .on('error', handleErr) .pipe(compress) .on('error', handleErr) .pipe(writeToDisk) .on('error', handleErr) Handling errors (correctly) @loige handleErr should end and destroy the streams (it doesn't happen automatically) 78
100. readable .on('error', handleErr) .pipe(decompress) .on('error', handleErr) .pipe(decrypt) .on('error', handleErr) .pipe(convert) .on('error', handleErr) .pipe(encrypt) .on('error', handleErr) .pipe(compress) .on('error', handleErr) .pipe(writeToDisk) .on('error', handleErr) Handling errors (correctly) @loige handleErr should end and destroy the streams (it doesn't happen automatically) 78
101. 04. STREAM UTILITIES 04. STREAM UTILITIES @loige 79
102. // stream-copy-gzip-pipeline.js const { pipeline } = require('stream') const { createReadStream, createWriteStream } = require('fs') const { createGzip } = require('zlib') const [, , src, dest] = process.argv pipeline( createReadStream(src), createGzip(), createWriteStream(dest), function onEnd (err) { if (err) { console.error(`Error: ${err}`) process.exit(1) } console.log('Done!') } ) stream.pipeline(...streams, callback) @loige 80
103. // stream-copy-gzip-pipeline.js const { pipeline } = require('stream') const { createReadStream, createWriteStream } = require('fs') const { createGzip } = require('zlib') const [, , src, dest] = process.argv pipeline( createReadStream(src), createGzip(), createWriteStream(dest), function onEnd (err) { if (err) { console.error(`Error: ${err}`) process.exit(1) } console.log('Done!') } ) stream.pipeline(...streams, callback) @loige Can pass multiple streams (they will be piped) 80
104. // stream-copy-gzip-pipeline.js const { pipeline } = require('stream') const { createReadStream, createWriteStream } = require('fs') const { createGzip } = require('zlib') const [, , src, dest] = process.argv pipeline( createReadStream(src), createGzip(), createWriteStream(dest), function onEnd (err) { if (err) { console.error(`Error: ${err}`) process.exit(1) } console.log('Done!') } ) stream.pipeline(...streams, callback) @loige Can pass multiple streams (they will be piped) The last argument is a callback. If invoked with an error, it means the pipeline failed at some point. All the streams are ended and destroyed correctly. 80
105. // stream-copy-gzip-pump.js const pump = require('pump') // from npm const { createReadStream, createWriteStream } = require('fs') const { createGzip } = require('zlib') const [, , src, dest] = process.argv pump( // just swap pipeline with pump! createReadStream(src), createGzip(), createWriteStream(dest), function onEnd (err) { if (err) { console.error(`Error: ${err}`) process.exit(1) } console.log('Done!') } ) stream.pipeline is available in Node.js 10+ In older systems you can use pump - npm.im/pump @loige 81
106. // stream-copy-gzip-pump.js const pump = require('pump') // from npm const { createReadStream, createWriteStream } = require('fs') const { createGzip } = require('zlib') const [, , src, dest] = process.argv pump( // just swap pipeline with pump! createReadStream(src), createGzip(), createWriteStream(dest), function onEnd (err) { if (err) { console.error(`Error: ${err}`) process.exit(1) } console.log('Done!') } ) stream.pipeline is available in Node.js 10+ In older systems you can use pump - npm.im/pump @loige 81
107. pumpify(...streams) - Create reusable pieces of pipeline npm.im/pumpify @loige Let's create EncGz, an application that helps us to read and write encrypted- gzipped files 82
108. // encgz-stream.js - utility library const { createCipheriv, createDecipheriv, randomBytes, createHash } = require('crypto') const { createGzip, createGunzip } = require('zlib') const pumpify = require('pumpify') // from npm // calculates md5 of the secret (trimmed) function getChiperKey (secret) {} function createEncgz (secret) { const initVect = randomBytes(16) const cipherKey = getChiperKey(secret) const encryptStream = createCipheriv('aes256', cipherKey, initVect) const gzipStream = createGzip() const stream = pumpify(encryptStream, gzipStream) stream.initVect = initVect return stream } @loige 83
109. // encgz-stream.js - utility library const { createCipheriv, createDecipheriv, randomBytes, createHash } = require('crypto') const { createGzip, createGunzip } = require('zlib') const pumpify = require('pumpify') // from npm // calculates md5 of the secret (trimmed) function getChiperKey (secret) {} function createEncgz (secret) { const initVect = randomBytes(16) const cipherKey = getChiperKey(secret) const encryptStream = createCipheriv('aes256', cipherKey, initVect) const gzipStream = createGzip() const stream = pumpify(encryptStream, gzipStream) stream.initVect = initVect return stream } @loige 83
110. // encgz-stream.js - utility library const { createCipheriv, createDecipheriv, randomBytes, createHash } = require('crypto') const { createGzip, createGunzip } = require('zlib') const pumpify = require('pumpify') // from npm // calculates md5 of the secret (trimmed) function getChiperKey (secret) {} function createEncgz (secret) { const initVect = randomBytes(16) const cipherKey = getChiperKey(secret) const encryptStream = createCipheriv('aes256', cipherKey, initVect) const gzipStream = createGzip() const stream = pumpify(encryptStream, gzipStream) stream.initVect = initVect return stream } @loige 83
111. // encgz-stream.js - utility library const { createCipheriv, createDecipheriv, randomBytes, createHash } = require('crypto') const { createGzip, createGunzip } = require('zlib') const pumpify = require('pumpify') // from npm // calculates md5 of the secret (trimmed) function getChiperKey (secret) {} function createEncgz (secret) { const initVect = randomBytes(16) const cipherKey = getChiperKey(secret) const encryptStream = createCipheriv('aes256', cipherKey, initVect) const gzipStream = createGzip() const stream = pumpify(encryptStream, gzipStream) stream.initVect = initVect return stream } @loige 83
112. // encgz-stream.js (...continue from previous slide) function createDecgz (secret, initVect) { const cipherKey = getChiperKey(secret) const decryptStream = createDecipheriv('aes256', cipherKey, initVect) const gunzipStream = createGunzip() const stream = pumpify(gunzipStream, decryptStream) return stream } module.exports = { createEncgz, createDecgz } @loige 84
113. // encgz-stream.js (...continue from previous slide) function createDecgz (secret, initVect) { const cipherKey = getChiperKey(secret) const decryptStream = createDecipheriv('aes256', cipherKey, initVect) const gunzipStream = createGunzip() const stream = pumpify(gunzipStream, decryptStream) return stream } module.exports = { createEncgz, createDecgz } @loige 84
114. // encgz-stream.js (...continue from previous slide) function createDecgz (secret, initVect) { const cipherKey = getChiperKey(secret) const decryptStream = createDecipheriv('aes256', cipherKey, initVect) const gunzipStream = createGunzip() const stream = pumpify(gunzipStream, decryptStream) return stream } module.exports = { createEncgz, createDecgz } @loige 84
115. // encgz-stream.js (...continue from previous slide) function createDecgz (secret, initVect) { const cipherKey = getChiperKey(secret) const decryptStream = createDecipheriv('aes256', cipherKey, initVect) const gunzipStream = createGunzip() const stream = pumpify(gunzipStream, decryptStream) return stream } module.exports = { createEncgz, createDecgz } @loige 84
116. // encgz-stream.js (...continue from previous slide) function createDecgz (secret, initVect) { const cipherKey = getChiperKey(secret) const decryptStream = createDecipheriv('aes256', cipherKey, initVect) const gunzipStream = createGunzip() const stream = pumpify(gunzipStream, decryptStream) return stream } module.exports = { createEncgz, createDecgz } @loige 84
117. // encgz.js - CLI to encrypt and gzip (from stdin to stdout) const { pipeline } = require('stream') const { createEncgz } = require('./encgz-stream') const [, , secret] = process.argv const encgz = createEncgz(secret) console.error(`init vector: ${encgz.initVect.toString('hex')}`) pipeline( process.stdin, encgz, process.stdout, function onEnd (err) { if (err) { console.error(`Error: ${err}`) process.exit(1) } } ) @loige 85
118. // encgz.js - CLI to encrypt and gzip (from stdin to stdout) const { pipeline } = require('stream') const { createEncgz } = require('./encgz-stream') const [, , secret] = process.argv const encgz = createEncgz(secret) console.error(`init vector: ${encgz.initVect.toString('hex')}`) pipeline( process.stdin, encgz, process.stdout, function onEnd (err) { if (err) { console.error(`Error: ${err}`) process.exit(1) } } ) @loige 85
119. // encgz.js - CLI to encrypt and gzip (from stdin to stdout) const { pipeline } = require('stream') const { createEncgz } = require('./encgz-stream') const [, , secret] = process.argv const encgz = createEncgz(secret) console.error(`init vector: ${encgz.initVect.toString('hex')}`) pipeline( process.stdin, encgz, process.stdout, function onEnd (err) { if (err) { console.error(`Error: ${err}`) process.exit(1) } } ) @loige 85
120. // encgz.js - CLI to encrypt and gzip (from stdin to stdout) const { pipeline } = require('stream') const { createEncgz } = require('./encgz-stream') const [, , secret] = process.argv const encgz = createEncgz(secret) console.error(`init vector: ${encgz.initVect.toString('hex')}`) pipeline( process.stdin, encgz, process.stdout, function onEnd (err) { if (err) { console.error(`Error: ${err}`) process.exit(1) } } ) @loige 85
121. // decgz.js - CLI to gunzip and decrypt (from stdin to stdout) const { pipeline } = require('stream') const { createDecgz } = require('./encgz-stream') const [, , secret, initVect] = process.argv const decgz = createDecgz(secret, Buffer.from(initVect, 'hex')) pipeline( process.stdin, decgz, process.stdout, function onEnd (err) { if (err) { console.error(`Error: ${err}`) process.exit(1) } } ) @loige 86
122. // decgz.js - CLI to gunzip and decrypt (from stdin to stdout) const { pipeline } = require('stream') const { createDecgz } = require('./encgz-stream') const [, , secret, initVect] = process.argv const decgz = createDecgz(secret, Buffer.from(initVect, 'hex')) pipeline( process.stdin, decgz, process.stdout, function onEnd (err) { if (err) { console.error(`Error: ${err}`) process.exit(1) } } ) @loige 86
123. // decgz.js - CLI to gunzip and decrypt (from stdin to stdout) const { pipeline } = require('stream') const { createDecgz } = require('./encgz-stream') const [, , secret, initVect] = process.argv const decgz = createDecgz(secret, Buffer.from(initVect, 'hex')) pipeline( process.stdin, decgz, process.stdout, function onEnd (err) { if (err) { console.error(`Error: ${err}`) process.exit(1) } } ) @loige 86
124. // decgz.js - CLI to gunzip and decrypt (from stdin to stdout) const { pipeline } = require('stream') const { createDecgz } = require('./encgz-stream') const [, , secret, initVect] = process.argv const decgz = createDecgz(secret, Buffer.from(initVect, 'hex')) pipeline( process.stdin, decgz, process.stdout, function onEnd (err) { if (err) { console.error(`Error: ${err}`) process.exit(1) } } ) @loige 86
125. @loige 87
126. stream.finished(streams, callback) Get notified when a stream is not readable, writable anymore - Node.js 10+ // finished can be promisified! const finished = util.promisify(stream.finished) const rs = fs.createReadStream('archive.tar') async function run() { await finished(rs) console.log('Stream is done reading.') } run().catch(console.error) rs.resume() // start & drain the stream @loige 88
127. stream.finished(streams, callback) Get notified when a stream is not readable, writable anymore - Node.js 10+ // finished can be promisified! const finished = util.promisify(stream.finished) const rs = fs.createReadStream('archive.tar') async function run() { await finished(rs) console.log('Stream is done reading.') } run().catch(console.error) rs.resume() // start & drain the stream @loige 88
128. stream.finished(streams, callback) Get notified when a stream is not readable, writable anymore - Node.js 10+ // finished can be promisified! const finished = util.promisify(stream.finished) const rs = fs.createReadStream('archive.tar') async function run() { await finished(rs) console.log('Stream is done reading.') } run().catch(console.error) rs.resume() // start & drain the stream @loige 88
129. readable-stream - Npm package that contains the latest version of Node.js stream library. It also makes Node.js streams compatible with the browser (can be used with Webpack and Broswserify) npm.im/readable-stream @loige * yeah, the name is misleading. The package offers all the functionalities in the official 'stream' package, not just readable streams. * 89
130. 04. WRITING CUSTOM 04. WRITING CUSTOM STREAMS STREAMS @loige 90
131. // emoji-stream.js (custom readable stream) const { EMOJI_MAP } = require('emoji') // from npm const { Readable } = require('readable-stream') // from npm const emojis = Object.keys(EMOJI_MAP) function getEmojiDescription (index) { return EMOJI_MAP[emojis[index]][1] } function getMessage (index) { return emojis[index] + ' ' + getEmojiDescription(index) } class EmojiStream extends Readable { constructor (options) { super(options) this._index = 0 } _read () { if (this._index >= emojis.length) { return this.push(null) } return this.push(getMessage(this._index++)) } } module.exports = EmojiStream @loige 91
132. // emoji-stream.js (custom readable stream) const { EMOJI_MAP } = require('emoji') // from npm const { Readable } = require('readable-stream') // from npm const emojis = Object.keys(EMOJI_MAP) function getEmojiDescription (index) { return EMOJI_MAP[emojis[index]][1] } function getMessage (index) { return emojis[index] + ' ' + getEmojiDescription(index) } class EmojiStream extends Readable { constructor (options) { super(options) this._index = 0 } _read () { if (this._index >= emojis.length) { return this.push(null) } return this.push(getMessage(this._index++)) } } module.exports = EmojiStream @loige 91
133. // emoji-stream.js (custom readable stream) const { EMOJI_MAP } = require('emoji') // from npm const { Readable } = require('readable-stream') // from npm const emojis = Object.keys(EMOJI_MAP) function getEmojiDescription (index) { return EMOJI_MAP[emojis[index]][1] } function getMessage (index) { return emojis[index] + ' ' + getEmojiDescription(index) } class EmojiStream extends Readable { constructor (options) { super(options) this._index = 0 } _read () { if (this._index >= emojis.length) { return this.push(null) } return this.push(getMessage(this._index++)) } } module.exports = EmojiStream @loige 91
134. // emoji-stream.js (custom readable stream) const { EMOJI_MAP } = require('emoji') // from npm const { Readable } = require('readable-stream') // from npm const emojis = Object.keys(EMOJI_MAP) function getEmojiDescription (index) { return EMOJI_MAP[emojis[index]][1] } function getMessage (index) { return emojis[index] + ' ' + getEmojiDescription(index) } class EmojiStream extends Readable { constructor (options) { super(options) this._index = 0 } _read () { if (this._index >= emojis.length) { return this.push(null) } return this.push(getMessage(this._index++)) } } module.exports = EmojiStream @loige 91
135. // emoji-stream.js (custom readable stream) const { EMOJI_MAP } = require('emoji') // from npm const { Readable } = require('readable-stream') // from npm const emojis = Object.keys(EMOJI_MAP) function getEmojiDescription (index) { return EMOJI_MAP[emojis[index]][1] } function getMessage (index) { return emojis[index] + ' ' + getEmojiDescription(index) } class EmojiStream extends Readable { constructor (options) { super(options) this._index = 0 } _read () { if (this._index >= emojis.length) { return this.push(null) } return this.push(getMessage(this._index++)) } } module.exports = EmojiStream @loige 91
136. // emoji-stream.js (custom readable stream) const { EMOJI_MAP } = require('emoji') // from npm const { Readable } = require('readable-stream') // from npm const emojis = Object.keys(EMOJI_MAP) function getEmojiDescription (index) { return EMOJI_MAP[emojis[index]][1] } function getMessage (index) { return emojis[index] + ' ' + getEmojiDescription(index) } class EmojiStream extends Readable { constructor (options) { super(options) this._index = 0 } _read () { if (this._index >= emojis.length) { return this.push(null) } return this.push(getMessage(this._index++)) } } module.exports = EmojiStream @loige 91
137. // uppercasify.js (custom transform stream) const { Transform } = require('readable-stream') class Uppercasify extends Transform { _transform (chunk, encoding, done) { this.push(chunk.toString().toUpperCase()) done() } _flush () { // in case there's buffered data // that still have to be pushed } } module.exports = Uppercasify @loige 92
138. // uppercasify.js (custom transform stream) const { Transform } = require('readable-stream') class Uppercasify extends Transform { _transform (chunk, encoding, done) { this.push(chunk.toString().toUpperCase()) done() } _flush () { // in case there's buffered data // that still have to be pushed } } module.exports = Uppercasify @loige 92
139. // uppercasify.js (custom transform stream) const { Transform } = require('readable-stream') class Uppercasify extends Transform { _transform (chunk, encoding, done) { this.push(chunk.toString().toUpperCase()) done() } _flush () { // in case there's buffered data // that still have to be pushed } } module.exports = Uppercasify @loige 92
140. // uppercasify.js (custom transform stream) const { Transform } = require('readable-stream') class Uppercasify extends Transform { _transform (chunk, encoding, done) { this.push(chunk.toString().toUpperCase()) done() } _flush () { // in case there's buffered data // that still have to be pushed } } module.exports = Uppercasify @loige 92
141. // uppercasify.js (custom transform stream) const { Transform } = require('readable-stream') class Uppercasify extends Transform { _transform (chunk, encoding, done) { this.push(chunk.toString().toUpperCase()) done() } _flush () { // in case there's buffered data // that still have to be pushed } } module.exports = Uppercasify @loige 92
142. // uppercasify.js (custom transform stream) const { Transform } = require('readable-stream') class Uppercasify extends Transform { _transform (chunk, encoding, done) { this.push(chunk.toString().toUpperCase()) done() } _flush () { // in case there's buffered data // that still have to be pushed } } module.exports = Uppercasify @loige 92
143. // uppercasify.js (custom transform stream) const { Transform } = require('readable-stream') class Uppercasify extends Transform { _transform (chunk, encoding, done) { this.push(chunk.toString().toUpperCase()) done() } _flush () { // in case there's buffered data // that still have to be pushed } } module.exports = Uppercasify @loige 92
144. // dom-append.js (custom writable stream) const { Writable } = require('readable-stream') class DOMAppend extends Writable { constructor (target, tag = 'p', options) { super(options) this._target = target this._tag = tag } _write (chunk, encoding, done) { const elem = document.createElement(this._tag) const content = document.createTextNode(chunk.toString()) elem.appendChild(content) this._target.appendChild(elem) done() } } module.exports = DOMAppend @loige 93
145. // dom-append.js (custom writable stream) const { Writable } = require('readable-stream') class DOMAppend extends Writable { constructor (target, tag = 'p', options) { super(options) this._target = target this._tag = tag } _write (chunk, encoding, done) { const elem = document.createElement(this._tag) const content = document.createTextNode(chunk.toString()) elem.appendChild(content) this._target.appendChild(elem) done() } } module.exports = DOMAppend @loige 93
146. // dom-append.js (custom writable stream) const { Writable } = require('readable-stream') class DOMAppend extends Writable { constructor (target, tag = 'p', options) { super(options) this._target = target this._tag = tag } _write (chunk, encoding, done) { const elem = document.createElement(this._tag) const content = document.createTextNode(chunk.toString()) elem.appendChild(content) this._target.appendChild(elem) done() } } module.exports = DOMAppend @loige 93
147. // dom-append.js (custom writable stream) const { Writable } = require('readable-stream') class DOMAppend extends Writable { constructor (target, tag = 'p', options) { super(options) this._target = target this._tag = tag } _write (chunk, encoding, done) { const elem = document.createElement(this._tag) const content = document.createTextNode(chunk.toString()) elem.appendChild(content) this._target.appendChild(elem) done() } } module.exports = DOMAppend @loige 93
148. // dom-append.js (custom writable stream) const { Writable } = require('readable-stream') class DOMAppend extends Writable { constructor (target, tag = 'p', options) { super(options) this._target = target this._tag = tag } _write (chunk, encoding, done) { const elem = document.createElement(this._tag) const content = document.createTextNode(chunk.toString()) elem.appendChild(content) this._target.appendChild(elem) done() } } module.exports = DOMAppend @loige 93
149. // dom-append.js (custom writable stream) const { Writable } = require('readable-stream') class DOMAppend extends Writable { constructor (target, tag = 'p', options) { super(options) this._target = target this._tag = tag } _write (chunk, encoding, done) { const elem = document.createElement(this._tag) const content = document.createTextNode(chunk.toString()) elem.appendChild(content) this._target.appendChild(elem) done() } } module.exports = DOMAppend @loige 93
150. // dom-append.js (custom writable stream) const { Writable } = require('readable-stream') class DOMAppend extends Writable { constructor (target, tag = 'p', options) { super(options) this._target = target this._tag = tag } _write (chunk, encoding, done) { const elem = document.createElement(this._tag) const content = document.createTextNode(chunk.toString()) elem.appendChild(content) this._target.appendChild(elem) done() } } module.exports = DOMAppend @loige 93
151. 05. STREAMS IN THE 05. STREAMS IN THE BROWSER BROWSER @loige 94
152. // browser/app.js const EmojiStream = require('../emoji-stream') const Uppercasify = require('../uppercasify') const DOMAppend = require('../dom-append') const list = document.getElementById('list') const emoji = new EmojiStream() const uppercasify = new Uppercasify() const append = new DOMAppend(list, 'li') emoji .pipe(uppercasify) .pipe(append) @loige 95
153. // browser/app.js const EmojiStream = require('../emoji-stream') const Uppercasify = require('../uppercasify') const DOMAppend = require('../dom-append') const list = document.getElementById('list') const emoji = new EmojiStream() const uppercasify = new Uppercasify() const append = new DOMAppend(list, 'li') emoji .pipe(uppercasify) .pipe(append) @loige 95
154. // browser/app.js const EmojiStream = require('../emoji-stream') const Uppercasify = require('../uppercasify') const DOMAppend = require('../dom-append') const list = document.getElementById('list') const emoji = new EmojiStream() const uppercasify = new Uppercasify() const append = new DOMAppend(list, 'li') emoji .pipe(uppercasify) .pipe(append) @loige 95
155. // browser/app.js const EmojiStream = require('../emoji-stream') const Uppercasify = require('../uppercasify') const DOMAppend = require('../dom-append') const list = document.getElementById('list') const emoji = new EmojiStream() const uppercasify = new Uppercasify() const append = new DOMAppend(list, 'li') emoji .pipe(uppercasify) .pipe(append) @loige 95
156. // browser/app.js const EmojiStream = require('../emoji-stream') const Uppercasify = require('../uppercasify') const DOMAppend = require('../dom-append') const list = document.getElementById('list') const emoji = new EmojiStream() const uppercasify = new Uppercasify() const append = new DOMAppend(list, 'li') emoji .pipe(uppercasify) .pipe(append) @loige 95
157. npm i --save-dev webpack webpack-cli node_modules/.bin/webpack src/browser/app.js # creates dist/main.js mv dist/main.js src/browser/app-bundle.js @loige Let's use webpack to build this app for the browser 96
158. <!DOCTYPE html> <html lang="en"> <head> <meta charset="utf-8" /> <meta name="viewport" content="width=device-width, initial-scale=1, shrink-to-fit=no" /> <title>Streams in the browser!</title> </head> <body> <ul id="list"></ul> <script src="app.bundle.js"></script> </body> </html> @loige Finally let's create an index.html 97
159. <!DOCTYPE html> <html lang="en"> <head> <meta charset="utf-8" /> <meta name="viewport" content="width=device-width, initial-scale=1, shrink-to-fit=no" /> <title>Streams in the browser!</title> </head> <body> <ul id="list"></ul> <script src="app.bundle.js"></script> </body> </html> @loige Finally let's create an index.html 97
160. <!DOCTYPE html> <html lang="en"> <head> <meta charset="utf-8" /> <meta name="viewport" content="width=device-width, initial-scale=1, shrink-to-fit=no" /> <title>Streams in the browser!</title> </head> <body> <ul id="list"></ul> <script src="app.bundle.js"></script> </body> </html> @loige Finally let's create an index.html 97
161. @loige 98
162. @loige 98
163. 06. CLOSING 06. CLOSING @loige 99
164. Streams have low memory footprint Process data as soon as it's available Composition through pipelines Define your logic as transform streams Readable and writable streams are good abstractions for input & output You can swap them as needed without having to change the logic @loige TLDR; TLDR; 100
165. IF YOU WANT TO LEARN (EVEN) MOAR IF YOU WANT TO LEARN (EVEN) MOAR ABOUT STREAMS... ABOUT STREAMS... nodejs.org/api/stream.html github.com/substack/stream-handbook @loige 101
166. @loige 102
167. CREDITS CREDITS Cover Photo by on for the amazing St. Patrick emoji art The internet for the memes! :D WeRoad Unsplash emojiart.org @loige SPECIAL THANKS SPECIAL THANKS , , , , @mariocasciaro @machine_person @Podgeypoos79 @katavic_d @UrsoLuca THANKS! THANKS! loige.link/streams-dub 103