Initial commit

.editorconfig

@@ -0,0 +1,18 @@
+# http://editorconfig.org
+root = true
+
+[*]
+charset = utf-8
+end_of_line = lf
+indent_size = 2
+indent_style = space
+insert_final_newline = true
+trim_trailing_whitespace = true
+max_line_length = 120
+
+[*.md]
+max_line_length = 0
+trim_trailing_whitespace = false
+
+[COMMIT_EDITMSG]
+max_line_length = 0

.gitignore

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

package.json

@@ -0,0 +1,17 @@
+{
+  "name": "node-streams",
+  "version": "1.0.0",
+  "main": "index.js",
+  "license": "MIT",
+  "type": "module",
+  "scripts": {
+    "client": "nodemon src/client.js",
+    "server": "nodemon src/server.js"
+  },
+  "devDependencies": {
+    "nodemon": "^2.0.20"
+  },
+  "dependencies": {
+    "axios": "^1.2.2"
+  }
+}

src/client.js

@@ -0,0 +1,46 @@
+import axios from 'axios'
+import { Transform, Writable } from 'stream'
+
+const url = 'http://localhost:3300'
+
+async function consume() {
+  const response = await axios({
+    url,
+    method: 'GET',
+    responseType: 'stream',
+  })
+
+  return response.data
+}
+
+const stream = await consume()
+
+stream
+  .pipe(
+    new Transform({
+      // Não é legal usar esta função como async, pois pode fechar
+      // a stream sem ter terminado o processamento
+      // - chunk é o pedaço recebido pela stream
+      transform(chunk, enc, cb) {
+        // Como o server manda o json como resposta é preciso fazer o parser dele
+        const item = JSON.parse(chunk)
+        const { name } = item
+
+        const itemNumber = /\d+/.exec(name)[0]
+
+        item.name = (itemNumber % 2 === 0) ? `${name} é par` : `${name} é ímpar`
+
+        // o callback informa o fim do processo
+        cb(null, JSON.stringify(item))
+      }
+    })
+  )
+  .pipe(
+    new Writable({
+      write(chunk, enc, cb) {
+        console.info('The End', chunk.toString())
+
+        cb()
+      }
+    })
+  )

src/server.js

@@ -0,0 +1,42 @@
+import http from 'http'
+import { Readable } from 'stream'
+import { randomUUID } from 'crypto'
+
+const PORT = process.env.PORT || 3300
+
+// Generator
+function* run() {
+  for(let i = 0; i < 100; i++) {
+    const data = {
+      id: randomUUID(),
+      name: `David-${i}`,
+    }
+
+    // Retorna o dado que já foi processado mesmo antes do loop terminar
+    yield data
+  }
+}
+
+async function handler(request, response) {
+  const readable = new Readable({
+    read() {
+      for (const data of run()) {
+        console.log('🚀 ~ sending', data);
+        // Transforma em string pois as streams só trabalham com este tipo de dado
+        // por causa do buffer
+        this.push(JSON.stringify(data) + "\n");
+      }
+
+      // Informa que os dados acabaram
+      this.push(null)
+    }
+  })
+
+  readable
+    // O pipe vai gerenciar o dado recebido
+    .pipe(response)
+}
+
+http.createServer(handler)
+  .listen(PORT)
+  .on('listening', () => console.info(`listening on localhost:${PORT}`))