DEV Community: Wesley Miranda

Creating your own ExpressJS from scratch (Part 6) - Creating a body-parser middleware

Wesley Miranda — Mon, 09 Oct 2023 17:01:10 +0000

I have not created tutorials for our Framework for a while, but It's time to return.

Understanding and Creating a Custom Body-Parser Middleware

In this tutorial, we'll dive into the creation of a custom body-parser middleware for our web framework. Before we get into the code, let's understand why this middleware is essential.

Why a Body-Parser Middleware?

The body-parser middleware is a crucial component in web applications as it allows us to transform incoming data streams from HTTP requests into accessible JavaScript objects. Understanding and creating a custom body-parser middleware is fundamental to handling incoming data effectively.

Creating the Middleware

src/body-parser.js

const BodyParser = async (req, res, next) => {
  // (1)
  let body = []

  for await (const chunk of req) {
    body.push(chunk)
  }

// (2)
  body = Buffer.concat(body).toString()

// (3)
  if (req.headers['content-type'] === 'application/json') {
    req.body = JSON.parse(body)
  }
  else if (req.headers['content-type'] === 'application/x-www-form-urlencoded') {
    let params = new URLSearchParams(body)
    let entries = params.entries();
    req.body = Object.fromEntries(entries)
  }

  next()
}

module.exports = BodyParser

From the code sessions above

We already know how to create middleware for our Framework, It has the same middleware definition. if you don't know, please take a look at the previous tutorials.

1 - We are reading and storing the Stream object from the request.

2 - Converting the Stream object to a String

3 - Depending on the content-type we have to convert the String to a Javascript Object.

Using the Custom Middleware

Now we have to import our middleware and use it for all requests.

index.js

const BodyParser = require('./src/body-parser')

app.useAll(BodyParser)

app.post('/body-parser', (req, res) => {
    res.send(`name: ${req.body.name}, age: ${req.body.age}`)
})

Testing the Middleware

If you make a POST request to the endpoint /body-parser with a JSON body like this:

{
    "name": "Wesley Miranda",
    "age": 28
}

You will receive the following response:

name: Wesley Miranda, age: 28

You can see here how the code is looking.

In the next tutorials, we are going to create the cors and multer middleware for our framework.

I hope you like it!

See you soon!

Creating your own ExpressJS from scratch (Part 5) - Serving static files

Wesley Miranda — Sat, 06 May 2023 18:57:25 +0000

Our Web Framework so far can create a REST API, but cannot serve files yet, we need to implement it to make the framework complete.

To serve files we are going to use the Streams concept to read the files and serve to the client.

If you want to know more about NodeJS Streams take a look at it here

Let's go!

Reading files and folders

src/app.js

// (1)
const { readdir } = require('fs/promises')
const { statSync, createReadStream } = require('fs')
const path = require('path')
const { pipeline } = require('stream/promises')

// (2)
async function* getAllStaticFiles(folder) {

    const files = await readdir(folder)
    for (const file of files) {
        const absolutePath = path.join(folder, file)
        if (statSync(absolutePath).isDirectory()) {
            yield* getAllStaticFiles(absolutePath)
        }
        else {
            yield absolutePath
        }
    }
}

// (3)
const static = async (folderPath) => {
    let folderRelative = folderPath.replace('./', '')
    for await (const file of getAllStaticFiles(folderPath)) {
        const pathWithoutBase = file.replace(folderRelative, '')
        get(pathWithoutBase, async (req, res) => {
            const relativePath = path.join(__dirname, '..', file)
            const fileStream = createReadStream(relativePath)
            res.setHeader('Content-Type', file.split('.').filter(Boolean).slice(1).join('.'))
            return await pipeline(fileStream, res)
        })
    }
}

// (4)
return {
    run,
    get,
    post,
    patch,
    put,
    del,
    use,
    useAll,
    useRouter,
    static
}

From the code sessions above

1 - To read files and folders and deal with Streams we need to import fs and stream libraries.

2 - Here we are reading files and folders' paths recursively using a generator pattern to make the reader be ordered.

3 - For each file path found we are going create a new GET route, read the file path as a Stream, and pass the Stream to the response object. If you didn't know the response object is a Writable Stream.

4 - Exporting the static function.

Testing

Now we can call the static function passing the folder path that contains the files we want to serve.

index.js

app.static('./files')

const start = async () => {
    app.run(3000)
}

start()

To test I will create a folder structure with a file like that.

files/folder1/file.json

{
  "test": "test"
}

If I run the app and try to access the path localhost:3000/folder1/file.json, I can see the content inside it.

You can see herehow the code is looking.

See you in the next tutorial to improve our Web Framework, we are going to create a body-parser middleware.

I hope you like it!

Creating your own ExpressJS from scratch (Part 4) - Modular router and global middlewares

Wesley Miranda — Sun, 23 Apr 2023 22:45:02 +0000

It's time to deal deeper with middleware and create a way to allow the users from our framework to create their routes separately from the main app object.

Let's get started!

Global Middlewares

src/app.js

// (1)
const middlewaresForAll = []

// (2)
const use = (path, ...middlewares) => {
    const possiblePaths = [path + '/GET', path + '/POST', path + '/PUT', path + '/PATCH', path + '/DELETE']
    possiblePaths.forEach(route => {
        const middlewaresAndControllers = routes.get(route) || []

        if (middlewaresAndControllers.length) {
            routes.set(route, [...middlewares, ...middlewaresAndControllers])
        }
    })
}

// (3)
const useAll = (...middlewares) => {
    middlewaresForAll.push(...middlewares)
}

const serverHandler = async (request, response) => {
    const sanitizedUrl = sanitizeUrl(request.url, request.method)

    const match = matchUrl(sanitizedUrl)

    if (match) {
        const middlewaresAndControllers = routes.get(match)
// (4)
        await dispatchChain(request, response,
            [requestDecorator.bind(null, routes.keys()), responseDecorator, ...middlewaresForAll, ...middlewaresAndControllers])
    } else {
        response.statusCode = 404
        response.end('Not found')
    }
}

From the code sessions above

1 - We need an array to store the middlewares that should be applied for all routes.

2 - The use function is responsible to add middleware for a specified path independently of the used method.

3 - The useAll is a way to add middleware to the middleware store.

4 - Now we can modify our serverHandler function to apply the array of middlewares for all routes.

Modular Router

It is a mini-app that we can include in our main application.

src/router.js

const Router = () => {
  const routes = new Map()
  const middlewaresForAll = []

  const getRoutes = () => {
    return routes
  }

  const getMiddlewaresForAll = () => {
    return middlewaresForAll
  }

  const useAll = (...middlewares) => {
    middlewaresForAll.push(...middlewares)
  }

  const use = (path, ...middlewares) => {
    const possiblePaths = [path + '/GET', path + '/POST', path + '/PUT', path + '/PATCH', path + '/DELETE']
    possiblePaths.forEach(route => {
      const middlewaresAndControllers = routes.get(route) || []

      if (middlewaresAndControllers.length) {
        routes.set(route, [...middlewares, ...middlewaresAndControllers])
      }
    })
  }

  const get = (path, ...handlers) => {
    const middlewaresAndControllers = routes.get(`${path}/GET`) || []
    routes.set(`${path}/GET`, [...middlewaresAndControllers, ...handlers])
  }

  const post = (path, ...handlers) => {
    const middlewaresAndControllers = routes.get(`${path}/POST`) || []
    routes.set(`${path}/POST`, [...middlewaresAndControllers, ...handlers])
  }

  const put = (path, ...handlers) => {
    const middlewaresAndControllers = routes.get(`${path}/PUT`) || []
    routes.set(`${path}/PUT`, [...middlewaresAndControllers, ...handlers])
  }

  const patch = (path, ...handlers) => {
    const middlewaresAndControllers = routes.get(`${path}/PATCH`) || []
    routes.set(`${path}/PATCH`, [...middlewaresAndControllers, ...handlers])
  }

  const del = (path, ...handlers) => {
    const middlewaresAndControllers = routes.get(`${path}/DELETE`) || []
    routes.set(`${path}/DELETE`, [...middlewaresAndControllers, ...handlers])
  }

  return {
    get,
    post,
    put,
    patch,
    del,
    use,
    useAll,
    getRoutes,
    getMiddlewaresForAll
  }

}

module.exports = Router

That is all we have in the main app, the only difference we need to return routes and middlewaresForAll too.

To use our modular router we need to get the router handlers and combine them with existent routes.

src/app.js

const useRouter = (path, router) => {
        const routerRoutes = router.getRoutes()
        const middlewaresFromRouter = router.getMiddlewaresForAll()
        const existentHandlers = routes.get(path) || []
        routerRoutes.forEach((middlewares, key) => {
            routes.set(`${path + key}`, [...existentHandlers, ...middlewaresFromRouter, ...middlewares])
        })
    }

Testing

router.get('/users', (req, res) => res.end('User route from router instance'))
router.get('/admins', (req, res) => res.end('Admins route'))

router.useAll((req, res, next) => {
    console.log('middleware for router instance /admins and /users')
    next()
})

router.use('/users', (req, res, next) => {
    console.log('middleware for /users')
    next()
})

router.use('/admins', (req, res, next) => {
    console.log('middleware for /admins')
    next()
})


app.useRouter('', router)

app.use('/admins', (req, res, next) => {
    console.log('middleware for all admins routes')
    next()
})

app.useAll((req, res, next) => {
    console.log('middleware for all routes')
    next()
})

If I try to access admins route running in a terminal this command:

curl --location 'localhost:3000/admins'

You must see in the console something like that:

middleware for all routes
middleware for all admins routes
middleware for router instance /admins and /users
middleware for /admins

That's all for now. You can see the entire code here

In the next tutorial, I will show how to serve static files.

See you soon!

Creating your own ExpressJS from scratch (Part 3) - Handling Request and Response objects

Wesley Miranda — Sat, 15 Apr 2023 23:56:43 +0000

In part 2, We applied the middleware pattern for our web framework, which make us able to execute how many middlewares that we want for a route.

Today We are going to create Decorators for our Request and Response objects, applying the decorator and middleware pattern together to get parameters and query strings from the URL, also create helper functions to respond to the requests as a JSON transformer, for example.

Let's go!

Request Decorator

src/request.js

const { match } = require('path-to-regexp')

const RequestDecorator = (routes, request, response, next) => {

// (1)
    const getParams = () => {
        const urlParams = request.url.split('/').slice(1)

        // remove querystrings from the last parameter
        const [lastParam] = urlParams[urlParams.length - 1].split('?')
        urlParams.splice(urlParams.length - 1, 1)

        // joining all params without querystrings
        const allParams = [...urlParams, lastParam].join('/')

        // matching url pattern to get the params
        for (const path of routes) {
            const urlMatch = match(path, {
                decode: decodeURIComponent,
            });
            const url = `/${allParams}/${request.method.toUpperCase()}`
            const found = urlMatch(url)
            if (found) {
                Object.keys(found.params).forEach(key => {
                    request.params = {
                        ...request.params,
                        [key]: found.params[key]
                    }
                })
                break
            }
        }
    }

// (2)
    const getQuery = () => {
        const urlParams = request.url.split('/').slice(1)

        // Isolating query strings
        const [lastParam, queryString] = urlParams[urlParams.length - 1].split('?')
        let params = new URLSearchParams(queryString);
        let entries = params.entries();

        request.query = {
            ...request.query,
            ...Object.fromEntries(entries)
        }
    }

    getParams()
    getQuery()
    next()
}

module.exports = RequestDecorator

From the code sessions above:

1 - Treating the URL to compare with the existent routes and get the parameters and set the request object.

2 - Transforming the query string into an object and setting the request object.

Response Decorator

src/response.js

const ResponseDecorator = (req, res, next) => {
// (1)
    res.status = (status) => {
        res.statusCode = status
        return res
    }
// (2)
    res.json = (data) => {
        res.setHeader('Content-type', 'application/json')
        res.end(JSON.stringify(data))
    }
// (3)
    res.send = async (data) => {
        res.end(data)
    }

    res.render = async (templatePath, data) => {
        // We are going to implement it later
    }

    next()

}

module.exports = ResponseDecorator

From the code sessions above:

1 - Changing the status code and returning the response object, to set the status and response in one line.

2 - Transforming the response into JSON.

3 - Send a text to the client.

Applying Decorators as Middlewares

src/app.js

We need to import our two decorators:

const requestDecorator = require('./request')
const responseDecorator = require('./response')

Now, We have to modify our serverHandler function to apply the decorators as middlewares for all routes.


const serverHandler = async (request, response) => {
        const sanitizedUrl = sanitizeUrl(request.url, request.method)

        const match = matchUrl(sanitizedUrl)

        if (match) {
            const middlewaresAndControllers = routes.get(match)
            await dispatchChain(request, response,
                [requestDecorator.bind(null, routes.keys()), responseDecorator, ...middlewaresAndControllers])
        } else {
            response.statusCode = 404
            response.end('Not found')
        }
    }

Testing

index.js


// (1)
app.get('/params/:id/:name', (req, res) => {
    res.end(JSON.stringify({ params: req.params, query: req.query }, null, 2))
})

// (2)
app.get('/response/:id', (req, res) => {
    if (req.params.id === '123') {
        res.status(200).json(req.params)
        return
    }

    res.status(400).json({ message: 'Invalid id' })
})

From the code sessions above:

1 - Returning the parameters and query strings from the URL as text.

2 - Here is an example of how you can return JSON and change the status code at the same time.

Conclusion

Using the same that We use for middlewares, we can treat the resquest and response and make them fancier.

You can take a look at the entire code here

I hope you like it.

Creating your own ExpressJS from scratch (Part 2) - Middlewares and Controllers

Wesley Miranda — Sun, 09 Apr 2023 19:13:02 +0000

Today I want to show you more about how to create a Web Framework like Express.js. This is part 2 of a series of a tutorial that I am creating. You missed part 1, check it out here

We are going to apply middleware and controllers for our routes, for it We need to use the middleware pattern. If you want to know more about it, I will let a link at the end of this tutorial.

Let's get started.

Middlewares

src/app.js

// (1)
const dispatchChain = (request, response, middlewares) => {
    return invokeMiddlewares(request, response, middlewares)
}

// (2)
const invokeMiddlewares = async (request, response, middlewares) => {

    if (!middlewares.length) return;

    const currentMiddleware = middlewares[0];

    return currentMiddleware(request, response, async () => {
            await invokeMiddlewares(request, response, middlewares.slice(1));
    })
}
// (3) 
const serverHandler = async (request, response) => {
    const sanitizedUrl = sanitizeUrl(request.url, request.method)

    const match = matchUrl(sanitizedUrl)

    if (match) {
            const middlewaresAndControllers = routes.get(match)
            await dispatchChain(request, response, [...middlewaresAndControllers])
    } else {
            response.statusCode = 404
            response.end('Not found')
    }
}

From the code sections above:

The dispatchChain as the name suggests, Will dispatch a chain of functions.
This is a middleware handler, responsible to execute and remove the function from the array of middleware, till there is no more function to execute.
Now we can modify our serverHandler function, to execute the middlewares for the current route.

Testing

index.js

const App = require('./src/app')

const app = App()


app.get('/test/test2', function test() { }, function test2() { })
app.post('/test', (req, res) => console.log('test'))
app.patch('/test', (req, res) => console.log('test'))
app.put('/test', (req, res) => console.log('test'))
app.del('/test', (req, res) => console.log('test'))

// (1)
const mid1 = (req, res, next) => {
    console.log('mid1')
    next()
}
// (2)
const mid2 = (req, res, next) => {
    console.log('mid2')
    next()
}
// (3)
const controller = (req, res) => {
    console.log('controller')
    res.end('controller')
}
// (4)
app.get('/middleware', mid1, mid2, controller)



const start = async () => {
    app.run(3000)
}

start()

From the code sections above:

First middleware function using the next function to pass to the next middleware.
Second middleware function using the next function to pass to the next middleware.
Controller to route without next function. Here we can return something to the client, as a message for example.
Applying the middleware and controller to a route.

Middleware pattern is not so difficult to apply, in a few code lines you can use it in your application.

At this link you can see more about the middleware pattern.

Also, you can take a look at the code here

Don't hesitate to follow me to receive more content.

Thanks!

Creating your own ExpressJS from scratch (Part 1) - Basics, Methods, and Routing

Wesley Miranda — Sun, 02 Apr 2023 20:25:00 +0000

Hey, have you ever thought about creating your web framework using NodeJS, to understand what is happening under the hood, the methods, the middlewares, controllers, and so on...

I know we have a lot of great frameworks to create awesome applications, like ExpressJS, NestJS, Fastify, etc. But you know what are they doing, I think the majority of the developers even don't care about it, only want to create their applications, and that's ok, but might they have a better experience creating applications and understanding what the framework is doing, allowing extends the framework's functionalities, fix some unexpected errors, improve performance issues, in general, make their life easier.

Along the way to creating our Web Framework, We are going to work with several concepts related to Javascript, NodeJS, and programming in general, like design patterns and regular expressions.

Application: Today we are going to create the first part of our Web Framework, to make us able to create routes and handle these routes.

Requirements:

NodeJS 16.16 version

Setup

We are going to create a basic setup here, only to be able to continue working.

mkdir web-framework

cd web-framework

npm init -y

Open the terminal you prefer
Create a folder with the name that you want.
Enter in the folder
Start a new npm project

Imports

src/app.js

const { createServer } = require('http')
const { match } = require('path-to-regexp')

The first line imports the createServer function from Node.js's built-in http module, which allows us to create an HTTP server. The second line imports the match function from the path-to-regexp package, which allows us to match URLs to specific routes.

To install path-to-regex package:

npm install path-to-regex

Application

src/app.js

const App = () => {
    const routes = new Map()
    const createMyServer = () => createServer(serverHandler.bind(this))

The App function is the main entry point for our web application framework. It creates a new Map object called routes to store all our defined routes. It also defines a helper function called createMyServer that creates an HTTP server using the serverHandler function, a function that We will create soon, will be the main handler.

Route Definition

It is not my intention to cover all the HTTP methods, but only show some of them to serve as an example.

src/app.js

    const get = (path, ...handlers) => {
        const currentHandlers = routes.get(`${path}/GET`) || []
        routes.set(`${path}/GET`, [...currentHandlers, ...handlers])
    }

    const post = (path, ...handlers) => {
        const currentHandlers = routes.get(`${path}/POST`) || []
        routes.set(`${path}/POST`, [...currentHandlers, ...handlers])
    }

    const put = (path, ...handlers) => {
        const currentHandlers = routes.get(`${path}/PUT`) || []
        routes.set(`${path}/PUT`, [...currentHandlers, ...handlers])
    }

    const patch = (path, ...handlers) => {
        const currentHandlers = routes.get(`${path}/PATCH`) || []
        routes.set(`${path}/PATCH`, [...currentHandlers, ...handlers])
    }

    const del = (path, ...handlers) => {
        const currentHandlers = routes.get(`${path}/DELETE`) || []
        routes.set(`${path}/DELETE`, [...currentHandlers, ...handlers])
    }

These are the route definition functions, which define a new route and specify the handlers for each HTTP method (GET, POST, PUT, PATCH, DELETE). They take a path argument and any number of handlers. Each function retrieves the current handlers for the specified method and path from the routes object, adds any new handlers, and sets the updated handlers back into the routes object.

the pattern to match with the route and find the right handler must contain the route and at the end the name of the method. like the examples below:

For /test/1/GET Path is /test/1 and the method is GET

For /test1/2/test2/POST Path is /test1/2/test2 and the method is POST

The Map is a good way to store the handlers because here we can use some useful functions like get and keys.

Treating the URLs

For our routing system works well We need to treat the URLs, because they don't come as we expect.

src/app.js

    const sanitizeUrl = (url, method) => {
        // Get the last part of the URL, removing the domain
        const urlParams = url.split('/').slice(1)

        // Remove querystrings from the last parameter
        const [lastParam] = urlParams[urlParams.length - 1].split('?')
        urlParams.splice(urlParams.length - 1, 1)

        // Create the URL with our pattern
        const allParams = [...urlParams, lastParam].join('/')
        const sanitizedUrl = `/${allParams}/${method.toUpperCase()}`

        return sanitizedUrl
    }

    const matchUrl = (sanitizedUrl) => {
        for (const path of routes.keys()) {
            const urlMatch = match(path, {
                decode: decodeURIComponent,
            })

            const found = urlMatch(sanitizedUrl)

            if (found) {
                return path
            }
        }
        return false
    }

These are helper functions used to sanitize and match URLs to their corresponding routes. sanitizeUrl takes a URL and an HTTP method and removes any query strings, then concatenates the parameters into a pattern matching the structure of the route. matchUrl iterates through the routes checking if the URL matches with one of our current routes.

Handling the Server

src/app.js

    const serverHandler = async (request, response) => {
        const sanitizedUrl = sanitizeUrl(request.url, request.method)

        const match = matchUrl(sanitizedUrl)

        if (match) {
            const middlewaresAndControllers = routes.get(match)
            console.log(middlewaresAndControllers)
            response.statusCode = 200
            response.end('Found')
        } else {
            response.statusCode = 404
            response.end('Not found')
        }
    }

The function first calls sanitizeUrl() with the request.url and request.method as arguments to create a sanitized URL, which is used to check against the routes set up in the framework.

The function then calls matchUrl() with the sanitized URL to try and find a matching route set up in the framework. If a match is found, it retrieves the list of middlewares and controllers associated with that route from the routes Map.

If there are middlewares and/or controllers associated with the route, the function logs them to the console and sets the response status code to 200, indicating that the request was successful. Finally, it ends the response with the message 'Found'.

If a match is not found, the function sets the response status code to 404, indicating that the requested resource was not found. Finally, it ends the response with the message 'Not found'.

The Run function and Exports

src/app.js

    const run = (port) => {
        const server = createMyServer()
        server.listen(port)
    }


    return {
        run,
        get,
        post,
        patch,
        put,
        del
    }

The run function is responsible for starting the server by calling the listen method of the server instance. The listen method accepts a port number as an argument, which is the port on which the server will listen for incoming requests.

In this function, a new instance of the server is created by calling the createMyServer function, which is defined earlier in the code. Then, the listen method is called on the server instance, passing in the port argument provided to the run function.

Finally, we can export all the public functions as a closure returning them.

Entire code

src/app.js

const { createServer } = require('http')
const { match } = require('path-to-regexp')

const App = () => {
    const routes = new Map()
    const createMyServer = () => createServer(serverHandler.bind(this))


    const get = (path, ...handlers) => {
        const currentHandlers = routes.get(`${path}/GET`) || []
        routes.set(`${path}/GET`, [...currentHandlers, ...handlers])
    }

    const post = (path, ...handlers) => {
        const currentHandlers = routes.get(`${path}/POST`) || []
        routes.set(`${path}/POST`, [...currentHandlers, ...handlers])
    }

    const put = (path, ...handlers) => {
        const currentHandlers = routes.get(`${path}/PUT`) || []
        routes.set(`${path}/PUT`, [...currentHandlers, ...handlers])
    }

    const patch = (path, ...handlers) => {
        const currentHandlers = routes.get(`${path}/PATCH`) || []
        routes.set(`${path}/PATCH`, [...currentHandlers, ...handlers])
    }

    const del = (path, ...handlers) => {
        const currentHandlers = routes.get(`${path}/DELETE`) || []
        routes.set(`${path}/DELETE`, [...currentHandlers, ...handlers])
    }

    const sanitizeUrl = (url, method) => {
        const urlParams = url.split('/').slice(1)

        // remove querystrings from the last parameter
        const [lastParam] = urlParams[urlParams.length - 1].split('?')
        urlParams.splice(urlParams.length - 1, 1)

        // create the URL with our pattern
        const allParams = [...urlParams, lastParam].join('/')
        const sanitizedUrl = `/${allParams}/${method.toUpperCase()}`

        return sanitizedUrl
    }

    const matchUrl = (sanitizedUrl) => {
        for (const path of routes.keys()) {
            const urlMatch = match(path, {
                decode: decodeURIComponent,
            })

            const found = urlMatch(sanitizedUrl)

            if (found) {
                return path
            }
        }
        return false
    }


    const serverHandler = async (request, response) => {
        const sanitizedUrl = sanitizeUrl(request.url, request.method)

        const match = matchUrl(sanitizedUrl)

        if (match) {
            const middlewaresAndControllers = routes.get(match)
            console.log(middlewaresAndControllers)
            response.statusCode = 200
            response.end('Found')
        } else {
            response.statusCode = 404
            response.end('Not found')
        }
    }

    const run = (port) => {
        const server = createMyServer()
        server.listen(port)
    }


    return {
        run,
        get,
        post,
        patch,
        put,
        del
    }
}

module.exports = App

Using and Testing

I created a separate file called index.js outside the src folder to test our framework.

index.js

const App = require('./src/app')

const app = App()


app.get('/test/test2', function test() { }, function test2() { })
app.post('/test', (req, res) => console.log('test'))
app.patch('/test', (req, res) => console.log('test'))
app.put('/test', (req, res) => console.log('test'))
app.del('/test', (req, res) => console.log('test'))



const start = async () => {
    app.run(3000)
}

start()

You can execute this file and test the application using your favorite HTTP client.

node index.js

Next Steps

In the next tutorial, we are working with middlewares and controllers. If you liked this tutorial don't hesitate to give your reaction and follow me to see at the first hand my next tutorials. I am trying to write at least one tutorial per week.

Thanks, see you soon!

Processing 13 million rows from a CSV file in the Browser (Without freezing the screen)

Wesley Miranda — Sun, 26 Mar 2023 20:07:43 +0000

Have you ever thought that you could process huge files in the browser? and even better, without freezing the browser screen. Browsers are well-known only for showing things on the screen, and all the hard processing is a server's responsibility. But, nowadays the browsers are much more powerful, and new great APIs have been implemented, which open many different possibilities to implement and improve our applications.

Application: We're going to create a pure Javascript application able to load many big CSV files and process them separately in different threads, transforming the CSV rows into a Javascript object and sending mocked requests for all these objects.

Main Principles:

Streams: Interface to load and process file chunks instead of the entire buffer.
Web Workers: Interface to deal with multithreading in the browser.

If you want to know about multithreading and Streams in NodeJS, you can access my tutorials through the links below:

Steps to Reproduce:

Create a thread worker to handle all the file processing
Inside our thread worker we are going to process the file using Streams and treat each chunk of the file.
At the end we are going to deal with the interface, to load the files and show the outcomes.

Requirements

I am using version 111.0.5563.65 of Google Chrome to test the application.
You will need to serve the application in your localhost because Chrome blocks the creation of threads if you are in a local path. I am using the vs code live server plugin to do that.
The CSV file with 13 million rows you can download here

Thread Worker

The thread worker must be responsible for the file processing, extracting the rows as Javascript objects, and in the end, simulating the requests for all the Javascript objects to a server.

To measure the file loading progress and transform CSV rows into a Javascript object just like we did at my first tutorial, We are going to use Transform Streams, and a Writable Stream to store the objects.

threadWorker.js


// (1)
let readableStream = null
let fileIndex = null
let bytesLoaded = 0
let linesSent = 0
const objectsToSend = []
let fileCompletelyLoaded = false
const readyEvent = new Event('ready')


// (2)
const ObjectTranform = {
    headerLine: true,
    keys: [],
    tailChunk: '',

    start() {
        this.decoder = new TextDecoder('utf-8');
    },

    transform(chunk, controller) {
        const stringChunks = this.decoder.decode(chunk, { stream: true })
        const lines = stringChunks.split('\n')

        for (const line of lines) {
            const lineString = (this.tailChunk + line)
            let values = lineString.split(',')

            if (this.headerLine) {
                this.keys = values
                this.headerLine = false
                continue
            }


            if (values.length !== this.keys.length || lineString[lineString.length - 1] === ',') {
                this.tailChunk = line
            } else {
                const chunkObject = {}

                this.keys.forEach((element, index) => {
                    chunkObject[element] = values[index]
                })

                this.tailChunk = ''

                controller.enqueue(`${JSON.stringify(chunkObject)}`)
            }
        }
    },

}

// (3)
const ProgressTransform = {
    transform(chunk, controller) {
        bytesLoaded += chunk.length
        controller.enqueue(chunk)
        postMessage({ progressLoaded: bytesLoaded, progressSent: linesSent, index: fileIndex, totalToSend: 0 })
    },
    flush() {
        fileCompletelyLoaded = true
    }
}

// (4)
const MyWritable = {
    write(chunk) {
        objectsToSend.push(postRequest(JSON.parse(chunk)))
    },
    close() {
        if (fileCompletelyLoaded) {
            postMessage({ totalToSend: objectsToSend.length, index: fileIndex, progressLoaded: bytesLoaded, progressSent: linesSent })
            dispatchEvent(readyEvent)
        }
    },
    abort(err) {
        console.log("Sink error:", err);
    },
}

// (5)
const postRequest = async data => {
    return new Promise((resolve, reject) => {
        setTimeout(() => {
            linesSent++
            postMessage({ totalToSend: objectsToSend.length, progressSent: linesSent, progressLoaded: bytesLoaded, index: fileIndex })
            resolve(data)
        }, 3000)
    })
}

// (6)
addEventListener('ready', async () => {
    await Promise.all(objectsToSend)
})

// (7)
addEventListener("message", event => {
    fileIndex = event.data?.index
    readableStream = event.data?.file?.stream()
    readableStream
        .pipeThrough(new TransformStream(ProgressTransform))
        .pipeThrough(new TransformStream(ObjectTranform))
        .pipeTo(new WritableStream(MyWritable))
})

From the code sections above:

Controller variables:
- readableStream: to store the file passed by the main thread.
- fileIndex: to handle which file is being processed.
- bytesLoaded: to sum the number of bytes processed.
- linesSent: to sum the number of objects that were sent.
- objectsToSend: array to store all the objects extracted.
- fileCompletelyLoaded: flag to know if the file was loaded.
- readyEvent: event to dispatch the requests to the server
Transform Stream to convert the rows into Javascript Objects. There is a session in one of the previous tutorials that I explain better how I do that. here
Transform Stream is responsible to control the progress, incrementing the bytesLoaded variables, and setting the flag fileCompletelyLoaded when the file is loaded. The enqueue function We apply to pass the chunk to the rest of the Stream pipeline.
Writable Stream is responsible to store all our converted objects and dispatch the ready event to start the requests.
Function used to simulate a request behavior with a 3 seconds delay.
Listener is responsible to listen when the ready is dispatched and starting the mocked requests.
Another listener, but it is waiting for messages from the main thread with the file index and the file object.

OBS: The postMessage function is used to send messages to the main thread that created the worker.

UI behavior

Our focus is not the interface here, so we have a simple input that accepts multiple files and a div to show the progress and information related to the files.

main.js


// (1)
const input = document.getElementById('files')
const progress = document.getElementById('progress')


// (2)
const formatBytes = (bytes, decimals = 2) => {
    if (!+bytes) return '0 Bytes'

    const k = 1024
    const dm = decimals < 0 ? 0 : decimals
    const sizes = ['Bytes', 'KiB', 'MiB', 'GiB', 'TiB', 'PiB', 'EiB', 'ZiB', 'YiB']

    const i = Math.floor(Math.log(bytes) / Math.log(k))

    return `${parseFloat((bytes / Math.pow(k, i)).toFixed(dm))} ${sizes[i]}`
}

// (3)
input.addEventListener('change', async (e) => {
    const files = e.target.files
    const workersInfo = []

    for (const i in files) {
        const worker = new Worker("threadWorker.js")
        if (files[i].name) {
            worker.postMessage({ index: i, name: files[i].name, file: files[i] })

            worker.addEventListener("message", event => {
                if (event.data) {
                    const infos = {
                        progressSent: event.data.progressSent, progressLoaded: event.data.progressLoaded, index: event.data.index, totalToSend: event.data.totalToSend, fileSize: files[i].size, fileName: files[i].name
                    }
                    workersInfo[i] = infos
                }

                progress.innerHTML = `
                    <table align="center" border cellspacing="1">
                        <thead>
                            <tr>
                                <th>File</th>
                                <th>File Size</th>
                                <th>Loaded</th>
                                <th></th>
                                <th>Total Rows</th>
                                <th>Rows Sent</th>
                                <th></th>
                            </tr>    
                        </thead>
                        <tbody>
                            ${workersInfo.map(info =>
                    `<tr>
                                    <td>${info.fileName}</td>
                                    <td>${formatBytes(info.fileSize)}</td>
                                    <td>${formatBytes(info.progressLoaded)}</td>
                                    <td><progress value="${Math.ceil(info.progressLoaded / info.fileSize * 100)}" max="100"> 32% </progress></td>
                                    <td>${info.totalToSend}</td>
                                    <td>${info.progressSent}</td>
                                    <td><progress value="${Math.ceil(info.progressSent / info.totalToSend * 100)}" max="100"> 32% </progress></td>
                                </tr>`
                )}
                        </tbody>
                    </table>
                    `

            })
        }
    }
})

From the code sections above:

Getting the references to manipulate the DOM.
This function I got on Stackoverflow elegantly shows the bytes.
The listener is responsible to listen when the user selects the files to process. When this event is dispatched we are going to create one thread for each, using postMessage function to pass the file to the thread created. Also, the thread should listen to messages to update the results on the screen.

Joining Everything

Now we can import all the scripts we created in an HTML file and put the necessary tags.

index.html

<!DOCTYPE html>
<html>

<head>
    <meta charset='utf-8'>
    <meta http-equiv='X-UA-Compatible' content='IE=edge'>
    <title>Multithreading browser</title>
    <meta name='viewport' content='width=device-width, initial-scale=1'>
    <script src='threadWorker.js' async></script>
    <script src='main.js' async></script>
</head>

<body>
    <input type="file" multiple id="files" accept=".csv" /><br><br>
    <div id="progress"></div>
</body>

</html>

Takeaways

Nowadays modern browsers make available awesome functionalities to improve performance and create nice things.
We can process hard jobs in the browser like manipulating files and taking off some server's responsibilities.

You can take a look at the entire code here

Uploading multiple files at the same time using multithreading in NodeJS

Wesley Miranda — Sun, 19 Mar 2023 17:44:11 +0000

NodeJS is well known for being single-threading, but it is not true, because only the event-loop is handled by a single thread. NodeJS gives to us the possibility to use 2 approaches for multithreading, worker_threads and child_process.

worker_threads are controlled by a process and have shared memory, which makes communication between them easier.
child_process are processes generated from the main thread, it's useful when we need direct communication with the Operational System, but we need more memory to create them.

Application: The purpose of the application we are going to create is to load all the folder's content and upload every file to Google Cloud Storage, but the most interesting part is that we will decide how many threads should do this operation, speeding up the upload process.

Main Principles:

Worker threads: NodeJS module to create threads.
Streams: if you don't know NodeJS streams I suggest you take a look at the tutorial that I've created here.
File System: NodeJS provides us with a simple way to access the OS and manipulate files and folders.

Steps to reproduce:

Load the folder's content
Create the threads
Create the upload worker
Assign the threads to upload worker

Requirements:

We are going to use NodeJS 16.16 version
You need to have a Google account to access Google Cloud Services.

Cloud Storage Service

We need to install cloud storage library to deal with google cloud service.

npm install @google-cloud/storage

If you need help with configuring your Cloud Storage service on your Google account, many good tutorials can help you with that, it's not the purpose of this tutorial.

Let's create our first file cloudStorageFileService.jsto work with our storage.

cloudStorageFileService.js

const { Storage } = require('@google-cloud/storage')
const path = require('path')
const serviceKey = path.join(__dirname, '../gkeys.json')


class CloudStorageFileService {

// (1)
    constructor() {
        this.storage = new Storage({
            projectId: 'my-project-id',
            keyFilename: serviceKey
        })
    }

// (2)
    async uploadFile(bucketName, destFileName) {
        return await this.storage
            .bucket(bucketName)
            .file(destFileName)
            .createWriteStream()
    }
}

module.exports = CloudStorageFileService

From the code sections above:

Basic configurations to use Cloud Storage service, as the project id and the path with your Google Cloud credentials.
Google Cloud Storage provides us a Writable Stream for uploading files.

Thread Controller

The thread controller will handle the thread distribution, we want to give a thread for each file, and upload them separately.

threadController.js

const {
    Worker
} = require('node:worker_threads');
const { readdir } = require('fs/promises')
const path = require('path')


class ThreadController {

// (1)
    constructor(threadsNumber) {
        this.files = []
        this.threadsNumber = threadsNumber
        this.count = 0
    }

// (2)
    async loadFiles() {
        this.files = await readdir(path.join(__dirname, '/content'))
    }

// (3)
    async uploadThread(filePath) {
        return new Promise((resolve, reject) => {
            const worker = new Worker('./fileUploadWorker.js', {
                workerData: {
                    file: filePath
                }
            });
            worker.once('error', reject);
            worker.on('exit', (code) => {
                resolve(filePath)
            });
        })
    }

// (4)
    async execute() {
        const init = performance.now()

        await this.loadFiles()

        let promises = []

        while (this.count < this.files.length) {

            for (let i = this.count; i < this.count + this.threadsNumber; i++) {
                if (this.files[i]) {
                    promises.push(this.uploadThread(this.files[i]))
                }
            }

            const result = await Promise.all(promises)

            promises = []
            this.count += this.threadsNumber

            console.log(result)
        }

        const end = performance.now()
        console.log(end - init)
    }
}

module.exports = ThreadController

From the code sections above:

Initializing our three main parameters, the number of threads that we want, the files we want to upload, and the counter for created threads.
Load all the files contained into the folder we want to process.
Here we are sending a message with the right file path to the worker thread and waiting until the thread finishes its process using a Worker object.
Running everything together. Now we are giving a thread for each file until there are no files to process. for example if there are 5 files and we pass 3 threads to process, at the first time It will process the first 3 files and at the second time will process the 2 files remaining. Also, I put a performance meter to test the behavior with a different number of threads.

File Upload Worker

The upload worker is the representation of the thread as a code, here we are going to put all we want that the thread does.

fileUploadWorker.js

const {
  isMainThread, parentPort, workerData
} = require('node:worker_threads');
const path = require('path')
const { pipeline } = require('stream/promises')
const { createReadStream } = require('fs')
const CloudStorageFileService = require('./cloudStorageFileService');

class FileUploadWorker {

// (1)
  constructor() {
    this.storage = new CloudStorageFileService()
    this.filePath = path.join(__dirname, '/content/', workerData.file)
    this.fileName = workerData.file
  }

// (2)
  async upload() {
    if (!isMainThread) {
      await pipeline(createReadStream(this.filePath), await this.storage.uploadFile('myfileuploads', this.fileName))
    }
  }
}

// (3)
;
(async () => {
  const fileUploader = new FileUploadWorker()
  await fileUploader.upload()
})()

From the code sections above:

In the constructor we need to initialize the Storage service or we could receive it as a parameter. Also, we need to get the file path from the parent thread through the workerData.
Here we check if we are in a thread dynamically created by us or in the NodeJS main thread. If we are not in the main thread we create a Readable Stream object from the file and upload it.
This anonymous function is responsible for executing our created thread.

Executing Everything

To test our application I will put 9 threads, one for each file in my folder. You can experiment with other values to measure the performance.

fileUploadWorker.js

const ThreadController = require('./threadController');

const controller = new ThreadController(9)
    ;
(async () => {
    await controller.execute()
})()

Takeaways

NodeJS is not single threading.
Threads are handy when you need to process a heavy job and don't want to crash the NodeJS main thread.
We also can use multithreading for batch jobs.

You can take a look at the entire code here

Processing massive amounts of data on demand without crashing nodejs main thread

Wesley Miranda — Wed, 01 Mar 2023 14:15:51 +0000

Sometimes we can have some problems processing big files using Node.js because the main thread(responsible for the event loop) can stop. For this reason, I decided to write about how to use streams in Node.js showing a funny example. Let's go!

Application: Node.js application that downloads a huge CSV file from Google Cloud Storage, processes the data, transforms it into a JSON file, extracts some data insights, and finally compresses the file and uploads it again to the Cloud Storage. (We are going to create a terminal progress bar to visualize the process progress)

Main principles:

Streams: Interface to work with chunks of data.
Readable Streams: Interface to consume data streams.
Writable Streams: Interface to provide data and source to the data streams.
Duplex Streams: Implement both interfaces, Readable and Writable.
Transform Streams: It's a Duplex Stream used to modify the data during the process.
PassThrough: It's a Transform Stream useful when we need to check or test something during the stream pipeline process.
Pipeline: It's useful to handle all the stream process.

Steps to reproduce:

Download file stream
Create a progress bar
Transform each CSV file line into javascript objects
Get information from the file
Convert the data into JSON format
Compress the data
Upload the new file converted

Requirements

The CSV file you can find here
We are going to use Node 16.16 version
You need to have a Google account to access Google Cloud Services.

Cloud Storage Service

We need to install cloud storage library to deal with google cloud service.

npm install @google-cloud/storage

If you need help with configuring your Cloud Storage service on your Google account, many good tutorials can help you with that, it's not the purpose of this tutorial.

Let's create our first file cloudStorageFileService.js in src folder to work with our storage.

src/cloudStorageFileService.js

const { Storage } = require('@google-cloud/storage')
const path = require('path')
const serviceKey = path.join(__dirname, '../gkeys.json')


class CloudStorageFileService {

// (1)
    constructor() {
        this.storage = new Storage({
            projectId: 'my-project-id',
            keyFilename: serviceKey
        })
    }

// (2)
    async downloadFile(bucketName, fileName) {
        return await this.storage
            .bucket(bucketName)
            .file(fileName)
            .createReadStream()
    }

// (3)
    async uploadFile(bucketName, destFileName) {
        return await this.storage
            .bucket(bucketName)
            .file(destFileName)
            .createWriteStream()
    }

// (4)
    async getFileSize(bucketName, fileName) {
        const [metadata] = await this.storage
                        .bucket(bucketName)
                        .file(fileName)
                        .getMetadata();
        return metadata.size
    }
}

module.exports = CloudStorageFileService

Basic configurations to use Cloud Storage service, as the project id and the path with your Google Cloud credentials.
Google Cloud Storage provides us a Readable Stream for downloading files, we can use it to download our huge file as a stream and not stuck main Node.js thread.
Google Cloud Storage also provides a Writable Stream to upload files.
The last function we are going to use to create our progress bar.

Progress Bar

Here we are extending PassThrough because we don't need to apply any modification to the stream, only to get some data to create our progress bar.

Important to know that Streams implement Event Emitter interface, and we can work with emit to emit events and on to listen to them.

src/progressPass.js

const { PassThrough } = require('stream')

class ProgressPass extends PassThrough {

// (1)
    constructor(fileSize, options = {}) {
        super({ ...options })
        this.on('data', this.processData)
        this.on('progress', this.showResult)
        this.on('close', this.finishProgress)
        this.bytesRead = 0
        this.progress = 0
        this.fileSize = fileSize
        this.createProgressBar()
    }

// (2)
    processData(data) {
        this.bytesRead += data.length
        this.progress = (this.bytesRead / this.fileSize) * 100
        this.emit('progress', Math.floor(this.progress))
    }

// (3)
    createProgressBar() {
        process.stdout.write("\x1B[?25l")
        process.stdout.write('[')
        for(let i = 1; i <= 101; i++) {
            process.stdout.write('-')
        }
        process.stdout.write(']')
    }

// (4)
    showResult(progress) {
        process.stdout.cursorTo(progress+1)
        process.stdout.write('=')


        process.stdout.cursorTo(105)
        process.stdout.write(`${progress}%`)

    }

    finishProgress() {
        process.stdout.write("\x1B[?25h")
        process.stdout.write("\n")
    }
}


module.exports = ProgressPass

Initializing our progress bar:
- The file size parameter added to the constructor will be useful to create the progress percentage.
- data event is emitted when comes a new chunk to the current stream.
- progress is our custom event that we emit to update the progress bar
- close is the event emitted when there is no more data to pass through the stream.
Every time that comes to a new data we get the chunk and add to the sum the chunk's length and emitting progress event to update the progress bar.
Creating a unfilled progress bar like:

[------------------------------------------------]

4 When progress event happens we update the progress bar with the current percentage:

[===========-------------------------------------]  25%

Transforming CSV line into a javascript object

Now It's time to get every line of our CSV file and convert it to JS object, the intention of it, is to manipulate the data easier and at the final of the process, convert it to a JSON file.

The strategy is to convert the binary chunk into text, and to read every line of the it.

To get the values we need to split the lines using ,.

When we are reading using streams, by default each chunk has 16kb (We can modify it), and at the moment of the separation could occur something like that:

chunk 1:

value1, value2, val

chunk 2:

ue3, value4, value5

Keep in mind we need to treat this kind of thing and save temporarily the previous chunk before splitting the text.

To treat the transformation we are going to use Transform Stream, as I mentioned It's useful when we want to transform our data chunks.

src/objectTransform.js

const { Transform } = require('stream')

class ObjectTranform extends Transform {

// (1)
    constructor(options = {}) {
        super({ ...options })
        this.headerLine = true
        this.keys = []
        this.tailChunk = ''
    }

// (2) 
    _transform(chunk, encoding, callback) {
        const stringChunks = chunk.toString("utf8")
        const lines = stringChunks.split('\n')

        for (const line of lines) {
            const lineString = (this.tailChunk + line)
            let values = lineString.split(',')

            if (this.headerLine) {
                this.keys = values
                this.headerLine = false
                continue
            }


            if (values.length !== this.keys.length || lineString[lineString.length - 1] === ',') {
                this.tailChunk = line
            } else {
                const chunkObject = {}

                this.keys.forEach((element, index) => {
                    chunkObject[element] = values[index]
                })

                this.tailChunk = ''
                this.push(`${JSON.stringify(chunkObject)}`)
            }
        }
        callback()
    }

// (3)
    _flush(callback) {
        callback()
    }

}

module.exports = ObjectTranform

Let's save the keys to create our JS objects using a flag. The tailChunk is for saving the incompleted CSV lines.
The magic happens here. We read every line, split the text wether the lines are completed, and convert it to JS object.
flush is called when there is no more data to be processed.

Extracting information

Now that we have JS objects we can extract some data from the chunks.

Here again, we are going to use PassThrough Stream because we only want to check some information.

src/monitorTransform.js

const { PassThrough } = require('stream')

class MonitorTransform extends PassThrough {
    constructor(options = {}) {
        super({ ...options })
        this.on('data', this.processData)
        this.on('close', this.showResult)
        this.totalCrimes = 0
        this.boroughTotal = new Map()
        this.monthTotal = new Map()
        this.yearTotal = new Map()
    }

// (1)
    processData(data) {
        const row = JSON.parse(data.toString())
        const rowCrimeQuantity = Number(row.value) || 0
        const currentBoroughTotal = Number(this.boroughTotal.get(row.borough)) || 0
        const currentMonthTotal = Number(this.monthTotal.get(row.month)) || 0
        const currentYearTotal = Number(this.yearTotal.get(row.year)) || 0

        this.totalCrimes += rowCrimeQuantity
        this.boroughTotal.set(row.borough, currentBoroughTotal + rowCrimeQuantity)
        this.monthTotal.set(row.month, currentMonthTotal + rowCrimeQuantity)
        this.yearTotal.set(row.year, currentYearTotal + rowCrimeQuantity)
    }

// (2)
    showResult() {
        console.log(this.totalCrimes)
        console.log(this.boroughTotal)
        console.log(this.monthTotal)
        console.log(this.yearTotal)
    }
}


module.exports = MonitorTransform

When a data event happens we can process the chunk and get whatever we want from it.
If there is no more data to process we show the extracted information.

Transforming the file to a JSON file

All the chunks are in a JS object format, but when we save the file, won't work properly, we need to treat it, transforming it into an array of objects using Transform Stream.

src/jsonTransform.js

const { Transform } = require('stream')

class JsonTransform extends Transform {
    constructor (options = {}) {
        super({ ...options })
        this.once('data', this.startJson)
        this.firstLine = true
    }

// (1)
    startJson() {
        this.push('[')
    }

// (2)
    _transform (chunk, encoding, callback) {
        const row = JSON.parse(chunk.toString())
        const newChunk = this.firstLine ? `${JSON.stringify(row)}` : `,${JSON.stringify(row)}`
        this.push(newChunk)
        if(this.firstLine) {
            this.firstLine = false
        }
        callback()
    }

// (3)
    _flush(callback) {
        this.push(']')
        callback()
    }

}

module.exports = JsonTransform

Our array should starts with [, it happens only in the first chunk. Here we are using once instead of on, to listen only one time the event data.
Here we are putting , to separate our JS objects.
At the end of the file we need to close the array with ]

Compressing the chunks

For it we don't need so much efforts, Node.js has a core library to do it.

The zlib from Node.js is a Transform Stream used to compress chunks of data during the streaming process. We only need to add this to our stream pipeline.

Creating Stream pipeline process

The pipeline process needs to have at least the Readable and the Writable Stream, and it can have how many Tranform Stream and PassThrough Stream we want.

To handle the pipeline process we are going to use the Builder design pattern, to control the creation process of our pipeline and throw some errors.

src/fileProcessor.js

const { pipeline } = require('stream/promises')

class FileProcessor {
    constructor() {
        this.readableStream = null
        this.transforms = []
        this.writableStream = null
    }

// (1)
    setReadable(readableStream) {
        this.readableStream = readableStream
        return this
    }

// (2)
    addTransforms(transformsStream) {
        this.transforms = transformsStream
        return this
    }

// (3)
    setWritable(writableStream) {
        this.writableStream = writableStream
        return this
    }

// (4)
    async execute() {
        try {
            if(!this.readableStream) {
                throw Error('Readable stream not implemented')
            }
            if(!this.writableStream) {
                throw Error('Writable stream not implemented')
            }
            await pipeline(this.readableStream, ...this.transforms, this.writableStream)
        } catch (error) {
            console.log(error)
        }
    }
}

module.exports = FileProcessor

Setting our Readable Stream, in this case should be the donwload function from Google Cloud Storage.
Adding Tranforms to our pipeline. All the Transform and PassThrough should be added as an array.
The Writable Stream is the upload function from Google Cloud Storage.
To execute the pipeline if there are the Writable and Readable Stream. The pipeline function must have at least these two functions.

Joining Everything

It's time to import everything we've created, define the constants and execute the the entire app.

src/index.js

// (1)
const FileProcessor = require('./fileProcessor')
const JsonTransform = require('./jsonTransform')
const MonitorTransform = require('./monitorTransform')
const ObjectTransform = require('./objectTransform')
const { createGzip } = require('node:zlib')
const CloudStorageFileService = require('./cloudStorageFileService')
const ProgressPass = require('./progressPass')
const fileProcessor = new FileProcessor()
const cloudFileService = new CloudStorageFileService()

// (2)
const gzip = createGzip()

// (3)
const bucketName = 'myfileuploads'
const fileName = 'london_crime_by_lsoa2.csv'
const destFileName = 'london_crime_by_lsoa2.tar.gz'

// (4)
    ;
(async () => {
    try {
        const fileSize = await cloudFileService.getFileSize(bucketName, fileName)
        await fileProcessor
            .setReadable(await cloudFileService.downloadFile(bucketName, fileName))
            .addTransforms([new ProgressPass(fileSize), new ObjectTransform(), new MonitorTransform(), new JsonTransform(), gzip])
            .setWritable(await cloudFileService.uploadFile(bucketName, destFileName))
            .execute()
    } catch (e) {
        console.log(e)
    }
})()

Here are all the things we've created, services, Transform Streams and PassThrough Streams.
As I mentioned, this is the Node.js library to compress data.
We need to define our Cloud Storage bucket and file.
This is our final function to execute everything together, defining the Readable, Transforms, and the Writable.

Takeaways

You can use this approach to import and export data between databases or generate reports.
Streams are useful to process and treat audio and video files and for file conversion.

You can take a look at the entire code here