Added stream transform filter and reduce

Author: lmammino

06-coding-with-streams/13-transform-filter-reduce/README.md

@@ -0,0 +1,21 @@
+# 13-transform-filter-reduce
+
+This examples shows how to use Transform streams to implement streaming data filter and reduce blocks.
+
+
+## Dependencies
+
+Install all necessary dependencies with:
+
+```bash
+npm install
+```
+
+
+## Run
+
+To run the example:
+
+```bash
+node index.js
+```

06-coding-with-streams/13-transform-filter-reduce/data.csv.gz

@@ -0,0 +1,16 @@
+import { Transform } from 'stream'
+
+export class FilterByCountry extends Transform {
+  constructor (country, options = {}) {
+    options.objectMode = true
+    super(options)
+    this.country = country
+  }
+
+  _transform (record, enc, cb) {
+    if (record.country === this.country) {
+      this.push(record)
+    }
+    cb()
+  }
+}

06-coding-with-streams/13-transform-filter-reduce/filter-by-country.js

@@ -0,0 +1,18 @@
+import { createReadStream } from 'fs'
+import { createGunzip } from 'zlib'
+import parse from 'csv-parse'
+import { FilterByCountry } from './filter-by-country.js'
+import { SumProfit } from './sum-profit.js'
+
+const csvParser = parse({ columns: true })
+
+// A small difference from the code presented in the book is that
+// here we have gzipped the data to keep the download size of the repository
+// as small as possible. For this reason we added an extra step that decompresses
+// the data on the fly. The final result doesn't change
+createReadStream('data.csv.gz')
+  .pipe(createGunzip())
+  .pipe(csvParser)
+  .pipe(new FilterByCountry('Italy'))
+  .pipe(new SumProfit())
+  .pipe(process.stdout)

06-coding-with-streams/13-transform-filter-reduce/index.js

@@ -0,0 +1,16 @@
+{
+  "name": "13-transform-filter-reduce",
+  "version": "1.0.0",
+  "description": "",
+  "type": "module",
+  "main": "index.js",
+  "scripts": {
+    "test": "echo \"Error: no test specified\" && exit 1"
+  },
+  "keywords": [],
+  "author": "Luciano",
+  "license": "ISC",
+  "dependencies": {
+    "csv-parse": "^4.10.1"
+  }
+}

06-coding-with-streams/13-transform-filter-reduce/package-lock.json

@@ -0,0 +1,19 @@
+import { Transform } from 'stream'
+
+export class SumProfit extends Transform {
+  constructor (options = {}) {
+    options.objectMode = true
+    super(options)
+    this.total = 0
+  }
+
+  _transform (record, enc, cb) {
+    this.total += Number.parseFloat(record.profit)
+    cb()
+  }
+
+  _flush (cb) {
+    this.push(this.total.toString())
+    cb()
+  }
+}