Add solutions for 6th chapter

Author: kasymbayaman

.gitignore

@@ -0,0 +1 @@
+node_modules

06-streams/6.1-data-compression-efficiency.js

@@ -0,0 +1,85 @@
+/*
+6.1 Data compression efficiency: Write a command-line script that takes a
+file as input and compresses it using the different algorithms available in the
+zlib module (Brotli, Deflate, Gzip). You want to produce a summary table
+that compares the algorithm's compression time and compression efficiency
+on the given file. Hint: This could be a good use case for the fork pattern, but
+remember that we made some important performance considerations when
+we discussed it earlier in this chapter.
+*/
+
+import { PassThrough, pipeline } from 'stream';
+import { createReadStream, createWriteStream } from 'fs';
+import { createBrotliCompress, createGzip, createDeflate } from 'zlib';
+import path from 'path';
+import { hrtime } from 'process';
+
+class Profiler {
+  constructor(label) {
+    this.label = label
+  }
+  start = () => {
+    this.startTime = hrtime.bigint();
+  }
+  end = () => {
+    const endTime = hrtime.bigint();
+    const diff = endTime - this.startTime;
+    console.log(`${this.label} took ${diff / BigInt(1000000)} ms`);
+    return diff;
+  }
+}
+
+const sourceFilePath = process.argv[2];
+const destinationPath = path.dirname(sourceFilePath);
+
+const compress = (algoritm) => {
+  let label;
+  let createCompressStream;
+  if (algoritm === 'gzip') {
+    [label, createCompressStream] = ['Gzip', createGzip];
+  } else if (algoritm === 'brotli') {
+    [label, createCompressStream] = ['Brotli', createBrotliCompress];
+  } else if (algoritm === 'deflate') {
+    [label, createCompressStream] = ['Deflate', createDeflate];
+  } else {
+    throw new Error('Invalid algorithm', algoritm);
+  }
+  const profiler = new Profiler(label);
+
+  const readStream = createReadStream(sourceFilePath);
+  const compressStream = createCompressStream();
+  const writeStream = createWriteStream(path.join(destinationPath, `book${algoritm}.gz`));
+
+  let size = 0;
+  const passThrough = new PassThrough();
+  passThrough.on('data', (chunk) => {
+    size += chunk.length;
+  })
+  const cb = (err) => {
+    if (err) throw err;
+    console.log('====================================')
+    profiler.end();
+    console.log(`Total byte size is ${size}`);
+  }
+
+  profiler.start();
+  pipeline(readStream, compressStream, passThrough, writeStream, cb);
+}
+
+// Algorithms data compression test should be done isolated
+// In fork mode slowest algorithm slows down others
+compress('gzip');
+compress('brotli');
+compress('deflate');
+
+/* Sample output
+====================================
+Deflate took 288 ms
+Total byte size is 4540894
+====================================
+Gzip took 295 ms
+Total byte size is 4540906
+====================================
+Brotli took 12788 ms
+Total byte size is 4208085
+*/

06-streams/6.2-stream-data-processing/CountCrimesByParam.js

@@ -0,0 +1,32 @@
+import { Transform } from 'stream';
+
+export class CountCrimesByParam extends Transform {
+  constructor(options = {}) {
+    options.objectMode = true
+    super(options)
+    this.param = options.param;
+    this.crimesByParam = {};
+  }
+
+  _transform(row, enc, cb) {
+    try {
+      if (!row || row.year === 'year') return cb(); // skip header
+      const paramVal = row[this.param];
+      if (!this.crimesByParam[paramVal]) this.crimesByParam[paramVal] = 0;
+      this.crimesByParam[paramVal] += Number(row.value);
+      cb()
+    } catch (e) {
+      cb(e);
+    }
+
+  }
+
+  _flush(cb) {
+    try {
+      this.push(Object.entries(this.crimesByParam));
+      cb()
+    } catch (e) {
+      cb(e);
+    }
+  }
+}

06-streams/6.2-stream-data-processing/ResultToString.js

@@ -0,0 +1,18 @@
+import { Transform } from 'stream';
+
+export class ResultToString extends Transform {
+  constructor(options = {}) {
+    options.objectMode = true
+    super(options)
+    this.label = options.label || '';
+  }
+
+  _transform(arr, enc, cb) {
+    let text = `\n${this.label}\n`;
+    for (const entry of arr) {
+      text = text + `${entry[0]}: ${entry[1]}\n`;
+    }
+    this.push(text);
+    cb();
+  }
+}

06-streams/6.2-stream-data-processing/SortAndLimit.js

@@ -0,0 +1,22 @@
+import { Transform } from 'stream';
+
+export class SortAndLimit extends Transform {
+  constructor(options = {}) {
+    options.objectMode = true;
+    super(options);
+
+    this.sortDirection = options.sortDirection || 'desc';
+    this.limit = options.limit;
+  }
+
+  _transform(arr, enc, cb) {
+    try {
+      arr.sort((a, b) => this.sortDirection === 'desc' ? b[1] - a[1] : a[1] - b[1]);
+      const result = this.limit ? arr.slice(0, this.limit) : arr;
+      this.push(result);
+      cb();
+    } catch (e) {
+      cb(e);
+    }
+  }
+}

06-streams/6.2-stream-data-processing/TopCategoryPerArea.js

@@ -0,0 +1,41 @@
+import { Transform } from 'stream';
+
+export class TopCategoryPerArea extends Transform {
+  constructor(options = {}) {
+    options.objectMode = true
+    super(options)
+    this.countCategoriesByArea = {};
+  }
+
+  _transform(row, enc, cb) {
+    if (!row || row.year === 'year') return cb(); // skip header
+    try {
+      const area = this.countCategoriesByArea[row.borough];
+      if (area) {
+        if (!area[row.major_category]) area[row.major_category] = 0;
+        area[row.major_category] += Number(row.value);
+      } else {
+        this.countCategoriesByArea[row.borough] = {
+          [row.major_category]: Number(row.value),
+        }
+      }
+      cb();
+    } catch (e) {
+      cb(e);
+    }
+  }
+
+  _flush(cb) {
+    try {
+      const res = Object.entries(this.countCategoriesByArea).map(([key, val]) => {
+        const maxVal = Object.entries(val).sort(([, a], [, b]) => b - a)[0];
+        return [`${key}, ${maxVal[0]}`, maxVal[1]]
+      });
+  
+      this.push(res);
+      cb();
+    } catch (e) {
+      cb(e);
+    }
+  }
+}

06-streams/6.2-stream-data-processing/index.js

@@ -0,0 +1,70 @@
+/*
+  6.2 Stream data processing: On Kaggle, you can find a lot of interesting data
+sets, such as the London Crime Data ( nodejsdp.link/london-crime ). You can
+download the data in CSV format and build a stream processing script that
+analyzes the data and tries to answer the following questions:
+• Did the number of crimes go up or down over the years?
+• What are the most dangerous areas of London?
+• What is the most common crime per area?
+• What is the least common crime?
+*/
+
+import { createReadStream } from 'fs';
+import { parse } from 'csv-parse'
+import { pipeline } from 'stream';
+import { CountCrimesByParam  } from './CountCrimesByParam.js';
+import { ResultToString  } from './ResultToString.js';
+import { SortAndLimit } from './SortAndLimit.js';
+import { TopCategoryPerArea } from './TopCategoryPerArea.js';
+
+
+const fileDest = '/home/baiaman_dev/Documents/short.csv';
+
+const readStream = createReadStream(fileDest);
+const csvParser = parse({ skip_records_with_error: true, columns: true });
+
+const cb = (err) => {
+  console.log('Error occurred in pipeline', err);
+}
+
+// Did the number of crimes go up or down over the years?
+pipeline(
+  readStream,
+  csvParser,
+  new CountCrimesByParam({ param: 'year' }),
+  new ResultToString({ label: 'Number of crimes by year' }),
+  process.stdout,
+  cb
+);
+
+// What are the most dangerous areas of London?
+pipeline(
+  readStream,
+  csvParser,
+  new CountCrimesByParam({ param: 'borough' }),
+  new SortAndLimit({ sortDirection: 'desc', limit: 20 }),
+  new ResultToString({ label: 'Most dangerous areas in London' }),
+  process.stdout,
+  cb
+);
+
+// What is the least common crime?
+pipeline(
+  readStream,
+  csvParser,
+  new CountCrimesByParam({ param: 'major_category' }),
+  new SortAndLimit({ sortDirection: 'asc', limit: 20 }),
+  new ResultToString({ label: 'Top least common crimes' }),
+  process.stdout,
+  cb
+);
+
+// What is the most common crime per area?
+pipeline(
+  readStream,
+  csvParser,
+  new TopCategoryPerArea(),
+  new ResultToString({ label: 'Most common crime per area' }),
+  process.stdout,
+  cb
+);

06-streams/6.3-file-share-over-tcp/client.js

@@ -0,0 +1,20 @@
+import { createReadStream } from 'fs';
+import net from 'net';
+import { pipeline } from 'stream';
+import path from 'path';
+
+const filePath = process.argv[2] || '/home/baiaman_dev/Documents/spider.jpg';
+const filename = path.basename(filePath);
+
+const client = net.createConnection({ port: 3000 }, () => {
+  console.log('Connection listener');
+})
+
+client.on('connect', () => {
+  client.write(filename);
+  pipeline(createReadStream(filePath), client, (err) => {
+    if (err) console.log(err);
+
+    client.end();
+  })
+})

06-streams/6.3-file-share-over-tcp/server.js

@@ -0,0 +1,34 @@
+import net from 'net';
+import { createWriteStream, write } from 'fs';
+import { dirname, join } from 'path';
+import { fileURLToPath } from 'url';
+
+const __dirname = dirname(fileURLToPath(import.meta.url));
+let writeStream;
+
+const server = net.createServer((socket) => {
+  socket.on('data', (data) => {
+    // first chunk is filename
+    // ideally fixed byte length should be allocated
+    if (!writeStream) {
+      const file = data.toString();
+      writeStream = createWriteStream(join(__dirname, file));
+      console.log('Started accepting file', file);
+      return;
+    }
+
+    writeStream.write(data);
+  });
+
+  socket.on('close', () => {
+    writeStream.destroy();
+    writeStream = null;
+    console.log('closed')
+  })
+
+  socket.on('error', (e) => {
+    console.log('Error', e)
+  })
+})
+
+server.listen(3000);