Feat/record with pipeline (#318)

* use pipeline for nodejs streams

* use pipeline for nodejs streams

lib/record/upload.js

@@ -3,6 +3,7 @@ const Websocket = require('ws');
 const PCMToMP3Encoder = require('./encoder');
 const wav = require('wav');
 const { getUploader } = require('./utils');
+const { pipeline } = require('stream');
 
 async function upload(logger, socket) {
   socket._recvInitialMetadata = false;
@@ -60,22 +61,19 @@ async function upload(logger, socket) {
               bitrate: 128
             }, logger);
           }
-          const handleError = (err, streamType) => {
-            logger.error(
-              { err },
-              `Error while streaming for vendor: ${obj.vendor}, pipe: ${streamType}: ${err.message}`
-            );
-          };
 
           /* start streaming data */
-          const duplex = Websocket.createWebSocketStream(socket);
-          duplex
-            .on('error', (err) => handleError(err, 'duplex'))
-            .pipe(encoder)
-            .on('error', (err) => handleError(err, 'encoder'))
-            .pipe(uploadStream)
-            .on('error', (err) => handleError(err, 'uploadStream'));
-
+          pipeline(
+            Websocket.createWebSocketStream(socket),
+            encoder,
+            uploadStream,
+            (error) => {
+              if (error) {
+                logger.error({ error }, 'pipeline error, cannot upload data to storage');
+                socket.close();
+              }
+            }
+          );
         } else {
           logger.info(`account ${accountSid} does not have any bucket credential, close the socket`);
           socket.close();