From 5aaba3b1a11ba5082c3073bc8b6a3362a4a3dd8b Mon Sep 17 00:00:00 2001 From: Pedro Correa Date: Mon, 19 Aug 2024 19:42:48 -0300 Subject: [PATCH] :sparkles: (Node) stream --- Node/examples/stream/create-big-file.mjs | 9 + .../stream/implement-duplex-stream.mjs | 21 ++ .../stream/implement-object-mode-stream.mjs | 41 ++++ .../stream/implement-readable-stream.mjs | 14 ++ .../stream/implement-transform-stream.mjs | 12 + .../stream/implement-writable-stream.mjs | 10 + Node/examples/stream/server-with-stream.mjs | 12 + .../examples/stream/server-without-stream.mjs | 14 ++ Node/streams.org | 217 ++++++++++++++++++ 9 files changed, 350 insertions(+) create mode 100644 Node/examples/stream/create-big-file.mjs create mode 100644 Node/examples/stream/implement-duplex-stream.mjs create mode 100644 Node/examples/stream/implement-object-mode-stream.mjs create mode 100644 Node/examples/stream/implement-readable-stream.mjs create mode 100644 Node/examples/stream/implement-transform-stream.mjs create mode 100644 Node/examples/stream/implement-writable-stream.mjs create mode 100644 Node/examples/stream/server-with-stream.mjs create mode 100644 Node/examples/stream/server-without-stream.mjs create mode 100644 Node/streams.org diff --git a/Node/examples/stream/create-big-file.mjs b/Node/examples/stream/create-big-file.mjs new file mode 100644 index 0000000..6b4448d --- /dev/null +++ b/Node/examples/stream/create-big-file.mjs @@ -0,0 +1,9 @@ +import fs from 'node:fs' + +const file = fs.createWriteStream('./big.file') + +for (let i = 0; i <= 1e6; i++) { + file.write('Lorem ipsum dolor sit amet, consectetur adipisicing elit, sed do eiusmod tempor incididunt ut labore et dolore magna aliqua. Ut enim ad minim veniam, quis nostrud exercitation ullamco laboris nisi ut aliquip ex ea commodo consequat. Duis aute irure dolor in reprehenderit in voluptate velit esse cillum dolore eu fugiat nulla pariatur. Excepteur sint occaecat cupidatat non proident, sunt in culpa qui officia deserunt mollit anim id est laborum.\n') +} + +file.end() diff --git a/Node/examples/stream/implement-duplex-stream.mjs b/Node/examples/stream/implement-duplex-stream.mjs new file mode 100644 index 0000000..9ae07fe --- /dev/null +++ b/Node/examples/stream/implement-duplex-stream.mjs @@ -0,0 +1,21 @@ +import { Duplex } from 'node:stream' + +const inoutStream = new Duplex({ + write(chunk, encoding, callback) { + console.log(chunk.toString()) + callback() + } + + read(size) { + this.push(String.fromCharCode(this.currentCharCode++)) + if (this.currentCharCode > 90 ) { + this.push(null) + } + } +}) + +inoutStream.currentCharCode = 65 + +process.stding + .pipe(inountStream) + .pipe(process.stdout) diff --git a/Node/examples/stream/implement-object-mode-stream.mjs b/Node/examples/stream/implement-object-mode-stream.mjs new file mode 100644 index 0000000..f514295 --- /dev/null +++ b/Node/examples/stream/implement-object-mode-stream.mjs @@ -0,0 +1,41 @@ +import { Transform } from 'node:stream' + +const commaSplitter = new Transform({ + readableObjectMode: true, + + transform(chunk, encoding, callback) { + this.push(chunk.toString().trim().split(',')) + callback() + } +}) + +const arrayToObject = new Transform({ + readableObjectMode: true, + writableObjectMode: true, + + transform(chunk, encoding, callback) { + const obj = {} + for (let i = 0; i < chunk.length; i += 2) { + obj[chunk[i]] = chunk[i + 1] + } + this.push(obj) + + callback() + } +}) + +const objectToString = new Transform({ + readableObjectMode: true, + writableObjectMode: true, + + transform(chunk, encoding, callback) { + this.push(JSON.stringify(chunk) + '\n') + callback() + } +}) + +process.stdin + .pipe(commaSplitter) + .pipe(arrayToObject) + .pipe(objectToString) + .pipe(process.stdout) diff --git a/Node/examples/stream/implement-readable-stream.mjs b/Node/examples/stream/implement-readable-stream.mjs new file mode 100644 index 0000000..0c0ad60 --- /dev/null +++ b/Node/examples/stream/implement-readable-stream.mjs @@ -0,0 +1,14 @@ +import { Readable } from 'node:stream' + +const inStream = new Readable({ + read(size) { + this.push(String.fromCharCode(this.currentCharCode++)) + if (this.currentCharCode > 90 ) { + this.push(null) + } + } +}) + +inStream.currentCharCode = 65 + +inStream.pipe(process.stdout) diff --git a/Node/examples/stream/implement-transform-stream.mjs b/Node/examples/stream/implement-transform-stream.mjs new file mode 100644 index 0000000..6f56082 --- /dev/null +++ b/Node/examples/stream/implement-transform-stream.mjs @@ -0,0 +1,12 @@ +import { Transform } from 'node:stream' + +const upperCaseTr = new Transform({ + transform(chunk, encoding, callback) { + this.push(chunk.toString().toUpperCase()) + callback() + } +}) + +process.stding + .pipe(upperCaseTr) + .pipe(process.stdout) diff --git a/Node/examples/stream/implement-writable-stream.mjs b/Node/examples/stream/implement-writable-stream.mjs new file mode 100644 index 0000000..e1cd879 --- /dev/null +++ b/Node/examples/stream/implement-writable-stream.mjs @@ -0,0 +1,10 @@ +import { Writable } from 'node:stream' + +const outStream = new Writable({ + write(chunk, encoding, callback) { + console.log(chunk.toString()) + callback() + } +}) + +process.stdin.pipe(outStream) diff --git a/Node/examples/stream/server-with-stream.mjs b/Node/examples/stream/server-with-stream.mjs new file mode 100644 index 0000000..d0a55d7 --- /dev/null +++ b/Node/examples/stream/server-with-stream.mjs @@ -0,0 +1,12 @@ +import fs from 'node:fs' +import { createServer } from 'node:http' + +const server = createServer() + +server.on('request', (req, res) => { + const src = fs.createReadStream('./big.file') + + src.pipe(res) +}) + +server.listen(8000) diff --git a/Node/examples/stream/server-without-stream.mjs b/Node/examples/stream/server-without-stream.mjs new file mode 100644 index 0000000..cc190ad --- /dev/null +++ b/Node/examples/stream/server-without-stream.mjs @@ -0,0 +1,14 @@ +import fs from 'node:fs' +import { createServer } from 'node:http' + +const server = createServer() + +server.on('request', (req, res) => { + fs.readFile('./big.file', (err, data) => { + if (err) throw err; + + res.end(data) + }) +}) + +server.listen(8000) diff --git a/Node/streams.org b/Node/streams.org new file mode 100644 index 0000000..65db418 --- /dev/null +++ b/Node/streams.org @@ -0,0 +1,217 @@ +* Streams + +- don't need to fit in memory all at once +- data not available all at once +- instance of =EventEmitter= +- is usually consumed using =pipe= + +** Pratical +Create a huge file + +- creates a file with 400MB using a =writable= stream + +#+begin_src js :tangle examples/stream/create-big-file.mjs + import fs from 'node:fs' + + const file = fs.createWriteStream('./big.file') + + for (let i = 0; i <= 1e6; i++) { + file.write('Lorem ipsum dolor sit amet, consectetur adipisicing elit, sed do eiusmod tempor incididunt ut labore et dolore magna aliqua. Ut enim ad minim veniam, quis nostrud exercitation ullamco laboris nisi ut aliquip ex ea commodo consequat. Duis aute irure dolor in reprehenderit in voluptate velit esse cillum dolore eu fugiat nulla pariatur. Excepteur sint occaecat cupidatat non proident, sunt in culpa qui officia deserunt mollit anim id est laborum.\n') + } + + file.end() +#+end_src + +Let's serve this files with a running server + +- sends the file without blocking the main thread +- reads all the file in memory and then send it for the client + +#+begin_src js :tangle examples/stream/server-without-stream.mjs + import fs from 'node:fs' + import { createServer } from 'node:http' + + const server = createServer() + + server.on('request', (req, res) => { + fs.readFile('./big.file', (err, data) => { + if (err) throw err; + + res.end(data) + }) + }) + + server.listen(8000) +#+end_src + +- HTTP response object is also a writable stream + +#+begin_src js :tangle examples/stream/server-with-stream.mjs + import fs from 'node:fs' + import { createServer } from 'node:http' + + const server = createServer() + + server.on('request', (req, res) => { + const src = fs.createReadStream('./big.file') + + src.pipe(res) + }) + + server.listen(8000) +#+end_src + +** Types of streams + +- Readable stream :: source in which data can be consumed +- Writable stream :: destination to which data can be written +- Duplex stream :: is both Readable and Writable stream +- Transform stream :: is a duplex that is used to transform data when data is read and written + +** Pipe + +- pipes the output of a stream into the input of the next +- source has to be a readable stream and the destination has to be a writable + +** Stream events + +- streams can also be consumed with events + +** Paused and Flowing Modes of readable streams + +- also referred as =pull= and =push= modes +- all readable streams starts =paused=, only after the =read()= is that it changes to =flowing= +- in flowing mode, data can be lost if no consumers are available +- when consuming readable streams using =pipe=, we don't nedd to worry about this as it is managed automatically + +** Implement Writable Stream + +#+begin_src js :tangle examples/stream/implement-writable-stream.mjs + import { Writable } from 'node:stream' + + const outStream = new Writable({ + write(chunk, encoding, callback) { + console.log(chunk.toString()) + callback() + } + }) + + process.stdin.pipe(outStream) +#+end_src + +- chunk :: usually a buffer, unless specified +- encoding :: is needed, but usually ignored +- callback :: it signals whether the write was successful or not. To signal a failure call the callback with an error object + +** Implement a Readable Stream + +#+begin_src js :tangle examples/stream/implement-readable-stream.mjs + import { Readable } from 'node:stream' + + const inStream = new Readable({ + read(size) { + this.push(String.fromCharCode(this.currentCharCode++)) + if (this.currentCharCode > 90 ) { + this.push(null) + } + } + }) + + inStream.currentCharCode = 65 + + inStream.pipe(process.stdout) +#+end_src + +- pushing data into the stream when the consumer asks + +** Implement Duplex/Transform Stream + +#+begin_src js :tangle examples/stream/implement-duplex-stream.mjs + import { Duplex } from 'node:stream' + + const inoutStream = new Duplex({ + write(chunk, encoding, callback) { + console.log(chunk.toString()) + callback() + } + + read(size) { + this.push(String.fromCharCode(this.currentCharCode++)) + if (this.currentCharCode > 90 ) { + this.push(null) + } + } + }) + + inoutStream.currentCharCode = 65 + + process.stding + .pipe(inountStream) + .pipe(process.stdout) +#+end_src + +- readable and writable sides of a duplex operate independently from one another + +#+begin_src js :tangle examples/stream/implement-transform-stream.mjs + import { Transform } from 'node:stream' + + const upperCaseTr = new Transform({ + transform(chunk, encoding, callback) { + this.push(chunk.toString().toUpperCase()) + callback() + } + }) + + process.stding + .pipe(upperCaseTr) + .pipe(process.stdout) +#+end_src + +** Streams Object Mode + +- by default, streams only accepts Buffer or String +- flag =objectMode= to set stream to accept any js object + +#+begin_src js :tangle examples/stream/implement-object-mode-stream.mjs + import { Transform } from 'node:stream' + + const commaSplitter = new Transform({ + readableObjectMode: true, + + transform(chunk, encoding, callback) { + this.push(chunk.toString().trim().split(',')) + callback() + } + }) + + const arrayToObject = new Transform({ + readableObjectMode: true, + writableObjectMode: true, + + transform(chunk, encoding, callback) { + const obj = {} + for (let i = 0; i < chunk.length; i += 2) { + obj[chunk[i]] = chunk[i + 1] + } + this.push(obj) + + callback() + } + }) + + const objectToString = new Transform({ + readableObjectMode: true, + writableObjectMode: true, + + transform(chunk, encoding, callback) { + this.push(JSON.stringify(chunk) + '\n') + callback() + } + }) + + process.stdin + .pipe(commaSplitter) + .pipe(arrayToObject) + .pipe(objectToString) + .pipe(process.stdout) +#+end_src -- 2.51.2