-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat(pipeline): readable, transform & writable with async pipeline
- Loading branch information
1 parent
7f3c720
commit f6ea094
Showing
2 changed files
with
80 additions
and
0 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 1,49 @@ | ||
| import { pipeline, Readable, Transform } from 'stream'; | ||
| import { promisify } from 'util'; | ||
| import { createWriteStream } from 'fs'; | ||
|
|
||
| /* | ||
| o pipeline funciona do mesmo modo que o pipe, | ||
| porém ele resolve problemas de vazamento de memória | ||
| e possui um método de callback */ | ||
| /* o promisify permite dar um await na pipeline */ | ||
| const pipelineAsync = promisify(pipeline); | ||
|
|
||
| const readableStream = Readable({ | ||
| read () { | ||
| for ( let index = 0; index < 1e2; index ) { | ||
| const person = { id: Date.now() index, name: `Ricky-${index}` }; | ||
| const payload = JSON.stringify(person); | ||
| this.push(payload); | ||
| } | ||
| this.push(null); | ||
| } | ||
| }) | ||
|
|
||
| const writableMapToCSV = Transform({ | ||
| transform(chunk, enconding, callback) { | ||
| const payload = JSON.parse(chunk); | ||
| const result = `${payload.id},${payload.name.toUpperCase()}\n`; | ||
| callback(null, result); | ||
| } | ||
| }) | ||
|
|
||
| const setHeader = Transform({ | ||
| transform(chunk, enconding, callback) { | ||
| this.counter = this.counter ?? 0; | ||
| if (this.counter) { | ||
| return callback(null, chunk); | ||
| } | ||
|
|
||
| this.counter = 1; | ||
| callback(null, "id, name\n".concat(chunk)); | ||
| } | ||
| }) | ||
|
|
||
| await pipelineAsync( | ||
| readableStream, | ||
| writableMapToCSV, | ||
| setHeader, | ||
| // process.stdout, | ||
| createWriteStream('my.csv') | ||
| ) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 1,31 @@ | ||
| import { pipeline, Readable, Writable } from 'stream'; | ||
| import { promisify } from 'util'; | ||
|
|
||
| /* | ||
| o pipeline funciona do mesmo modo que o pipe, | ||
| porém ele resolve problemas de vazamento de memória | ||
| e possui um método de callback */ | ||
| /* o promisify permite dar um await na pipeline */ | ||
| const pipelineAsync = promisify(pipeline); | ||
|
|
||
| const readableStream = Readable({ | ||
| read: function () { | ||
| this.push("Hello world! 0"); | ||
| this.push("Hello world! 1"); | ||
| this.push("Hello world! 2"); | ||
| this.push(null); | ||
| } | ||
| }) | ||
|
|
||
| const writableStream = Writable({ | ||
| write(chunk, enconding, callback) { | ||
| console.log('msg', chunk.toString()); | ||
| callback(); | ||
| } | ||
| }) | ||
|
|
||
| await pipelineAsync( | ||
| readableStream, | ||
| // process.stdout, | ||
| writableStream, | ||
| ) |