Ever wondered how streams are working in Node.js? I had. So far I know that using streams will most certainly reduce memory usage on the server when processing large files. Instead of reading whole file into memory we stream it in chunks to whoever requested it and apply transformations to that stream if needed. That's huge benefit as it allows to avoid vertical scaling. Processing files is common task but not as common as interacting with databases and that's going to be my focus now. I'll build simple API with 2 endpoints. Both will be returning large amount of records from Postgres DB. One endpoint will stream data to the client and another will read whole data into memory and return in one chunk. If I work in Node.js environment, there's usually Express.js framework in front of it with Sequelize orm. For this experiment I decided to try something different and choose Koa.js framework and Knex orm. Let's start. I'll setup the project first: mkdir streamer && streamer npm init -y # let's name our project "streamer" cd # I'm happy with npm defaults for this project so: Now time to add app dependencies. You can see all packages . To setup my development environment I'll use and Let's see how looks like: here Docker docker-compose. Dockerfile.dev node: . API_PORT ${API_PORT} FROM 12.16 1 # create new working directory WORKDIR /src # expecting to receive API_PORT as argument ARG # make sure to run latest version of npm RUN npm i npm@latest -g # fetch dependencies on a separate layer first as it's not changing that often, # will be cached and speed up the image build process COPY ./package.json ./package-lock.json ./ RUN npm i # copy the rest of the source files into working directory COPY . . EXPOSE CMD [ , , ] "npm" "run" "start:dev" Pretty standard right? Now to add DB dependency and bootstrap whole thing: docker-compose.yml version: "3.7" volumes: streamervolume: name: streamer-volume networks: streamernetwork: name: streamer-network services: pg: image: postgres:12.0 restart: on-failure env_file: - .env ports: - "${POSTGRES_PORT}:${POSTGRES_PORT}" volumes: - streamervolume: /var/lib/postgresql/data networks: - streamernetwork streamer_api: build: context: . dockerfile: Dockerfile.dev args: API_PORT: ${API_PORT} restart: on-failure depends_on: - pg volumes: - ./:/src ports: - "${API_PORT}:${API_PORT}" networks: - streamernetwork env_file: - .env Now that environment is pretty much ready, I'll continue with application logic itself. Let's look at : app.js ( ).config(); Koa = ( ); errorHandler = ( ); router = ( ); app = Koa();

app
  .use(errorHandler)
  .use(router.routes())
  .use(router.allowedMethods()); .exports = app; // src/app.js require 'dotenv' const require 'koa' const require './middleware/errorHandler' const require './routes' const new module I first load vars, import Koa, my custom error handling middleware and router. The error handling middleware looks like this: .env .exports = (ctx, next) => { { next();
  } (err) { .error(err);
    ctx.status = err.status || ;
    ctx.body = { : err.message || ,
    };
  }
} // src/middleware/errorHandler.js module async try await catch console 500 msg 'Oops. Something went wrong. Please try again later' According to documentation, error handling middleware has to be added first in Koa app in order to capture all exceptions. Next stop is router: Router = ( ); router = Router();

router.get( , ( ));
router.get( , ( )); .exports = router; // src/routes/index.js const require '@koa/router' const new '/nostream' require './users/nostream' '/stream' require './users/stream' module Interesting thing to mention is that in I use middleware that handles unsupported methods with for me. Lets look at the handlers next. app.js router.allowedMethods() 405 Method Not Allowed nostream: db = ( ); .exports = (ctx) => { { users = db.select( ).from( );
    ctx.status = ;
    ctx.body = users;
  } (err) {
    ctx.throw( , err);
  }
}; // src/routes/users/nostream/index.js const require '../../../services/db' module async try const await '*' 'users' 200 catch 500 Pretty sure you're wondering how that looks like: db config = ( ); knex = ( )({ : , : { : config.dbHost, : config.dbUser, : config.dbPwd, : config.db,
  },
}); .exports = knex; // src/services/db.js const require './config' const require 'knex' client 'pg' connection host user password database module is just simple object literal that exposes some env vars. You can see it . I require knex, establish db connection and export instance of it. Since all node modules are singletons, no matter how many times I'll require this package, initiation will be performed only once. Everything else in handler is pretty much straightforward. Please refer to Koa or Knex documentation for more details. Next let's look at handler: config here nostream stream Stringify = ( ) db = ( ); .exports = (ctx) => {
  ctx.type = ;
  ctx.set( , ); { stream = db
      .select( )
      .from( )
      .stream();

    ctx.status = ; pipe(stream, ctx.res, { : });
  } (err) {
    ctx.throw( , err);
  }
}; { ( { .pipe(Stringify())
      .pipe(to, options); .on( , reject); .on( , resolve);
  })
} // src/routes/users/stream/index.js const require 'streaming-json-stringify' const require '../../../services/db' module async 'application/json; charset=utf-8' 'Connection' 'keep-alive' try const '*' 'users' 200 await end false catch 500 ( ) function pipe from, to, options return new Promise ( ) => resolve, reject from from 'error' from 'end' Here I call on my query to get stream instance. I then pipe this query stream to client. In order for that to happen I have function that returns a promise and we don't exit the handler instantly but rather wait until streaming is done or error occurs. I added stringify package here because response (writeable stream) expects an input of a type string or an instance of Buffer and DB stream operates with Object types. .stream() pipe By the way, all will be captured by middleware we created previously. Now before I jump into testing my server I need some data in DB. Since I have Knex already installed locally, I need a config file for it to be able to run migrations, seeds etc: ctx.throw errorHandler ./node_modules/.bin/knex init This will generate . I have slightly changed it to use my env vars. You can see it . Now I'll generate the migration: knexfile.js here ./node_modules/.bin/knex migrate:make users This will create a migration file in You can review it . Before I run it, start the application: migrations/{timestapm}_users.js here docker-compose up --build Now since app is up and running, time to run migration: POSTGRES_HOST=localhost npm run migrate:up I specify postgres host here since I run migration from host machine for DB  that runs in docker environment under pg host name and if I don't specify it, pg name will be used from env vars which will end up with failed connection attempt. Now what I need is some records in users table for my API to serve. Knex cli can ease this process. It has command to generate a seed file: ./node_modules/.bin/knex seed:make create_users This will generate a seed file in directory. You can see what I did with it . If you'll run into issues running it, try reducing amount of records you try to create. Currently it'll try to create 60k records. Lets run it: seeds here POSTGRES_HOST=localhost npm run db:seed OK, with all that in place, I can now test both endpoints and see how well they perform. Let's start with nostream endpoint: And now stream endpoint: So, streaming might be less memory intensive operation but it takes a bit longer to retrieve all data (60k users in this case). That certainly works when sending/processing large amounts of data and could be perfect fit for file processing or BI where we have to work with large data sets. When it comes to common tasks though, say admin panel that retrieves 10-100 records per page, most likely streaming wouldn't be the best fit. Not only because it takes a bit more time, but also due to code complexity as it requires more code to achieve same output. It all boils down to good old saying: I hope you have learned something useful. You can find the whole project . "choose the right tool for the job". here

Fetch

Instantly

2021 - HackerNoon Contributor of the Year - GO

2022 - HackerNoon Contributor of the Year - Aws

2022 - HackerNoon Contributor of the Year - Rest Api

HELLO WORLD_

Nominated for 2022 - HackerNoon Contributor of the Year - Rest Api

Nominated for 2022 - HackerNoon Contributor of the Year - Aws

Too Long; Didn't Read

Cassandra Forward a free Cassandra-focused community event

Node.js Streams in Action

Node.js Streams in Action

Too Long; Didn't Read

Companies Mentioned

@danstenger

RELATED STORIES