Add more transport strategies (microservices)

[kamilmysliwiec]

Feature Request

There are many requests for creating more transport strategies for the @nestjs/microservices package and it would be nice to track them in one single place. Since we'll very likely never work on more than 1 strategy at once, it doesn't make sense to have +5 issues opened instead.

Is your feature request related to a problem? Please describe.

Describe the solution you'd like

• Apache Thrift Apache thrift support #740
• AWS SQS/SNS - Kinesis/Lambda Support for AWS SQS/SNS #870 & Proposal: AWS Specific Microservices (Lambda/Kinesis) #2442
• ioredis Support ioredis with microservices (Redis) #1048
• Redis Streams Redis Streams support #3061
• NATS streaming NATS Streaming (STAN) #3955
• Cloud Pub/Sub (GCP)
• Azure Service Bus

Teachability, Documentation, Adoption, Migration Strategy

What is the motivation / use case for changing the behavior?

[TreeMan360]

+1 for AWS EventBridge support

[duro]

I'm curious for the AWS based transports, how many are using these running inside containers or VMs, vs processing these event sources in Lambda?

We are using Lambda as our deployment model, with Nest, and it has been working well. In that approach, processing Kinesis, SNS, SQS, EventBridge, has become trivial. We simply use NestFactory.createApplicationContext and then pass the event and context into a service class that handles the logic.

So in my case, I'm less interested in a Kinesis, SQS, SNS, EventBridge specific transport, and more interested in creating a better way for Nest to handle incoming API Gateway events in a way that does nto require spinning up an Express server in the background, and passing it over a socket (al. a aws-serverless-express). It's unnecessary overhead (which also has odd side effects), and if our experiance with using Nest to handle stream/queue/transport processing has taught us anything, Nest should be flexible enough to also handle incoming API Gateway events without Express.

[TreeMan360]

@duro this is a very good point. I am already using EventBridge without any issues and when I call a micro service I instead use createApplicationContext and that is working well for us at the moment. It would however be nice for nest microservices to support the transport but it’s not a showstopper as you point out. Using these transports would mean wiring received events to services becomes a lot easier than manually mapping the incoming payload to the appropriate event handler. And it keeps things consistent.

It would be very nice to have a solution that when deployed did not use an instance of express inside lambda to proxy the API gateway request. I totally agree it would be more beneficial to remove this overhead with respect to new AWS related functionality vs the transports.

I do however feel that perhaps the express provided middleware functionality may be an issue here in certain scenarios but that’s just me thinking aloud.

[kamilmysliwiec]

It would be very nice to have a solution that when deployed did not use an instance of express inside lambda to proxy the API gateway request.

@TreeMan360 @duro we have this implemented for Azure already. I suppose we should be able to deliver AWS version this year too

[duro]

@kamilmysliwiec I believe I looked into the Azure implementation and it was still using an express server under the hood, similar to aws-serverless-express. Please correct me if I'm wrong.

[kamilmysliwiec]

There are 2 modes available, with and without express. @duro

[BrutalSimplicity]

Ok, just a rando that stumbled onto this framework, and have been toying around with the idea of a lambda/serverless-esque framework for a while now, and wanted to give my thoughts...

I've been using the AWS CDK for the past 6 months and feel like this might be useful here. Perhaps some new decorators that define the lambda execution environments and notification triggers that is then scanned using the CDK to generate the necessary cloudformation behind the scenes. This would make the declaration of your lambda ecosystem completely code-driven.

Not sure if it makes sense for this project, but when I came across this framework I got excited by the possibilities.

Python has a framework that somewhat follows the convention I had in mind.
https://github.com/aws/chalice

[TreeMan360]

@BrutalSimplicity I am using nest with the serverless framework and in this case it’s simple to add the appropriate cloudformation to the serverless.yml. Not quite as nice as the CDK but it works very well with the severless variable model etc.

The decorators you mention would be perfect, I wouldn’t need any kind of CDK scanning to generate the infrastructure but that does also sound 😎.

[shlomiassaf]

I'm not sure supporting all of the above is the right thing to do...
It might be a big issue to support it in the long run.

Instead, I think it will be better to refactor the microservice package a bit, so it will be easier to add your own transport.

It is mostly built for that, as there are a lot of transport already, so just exposing the right stuff, documenting and maybe a little tutorial will do.

I'm sure that some transports can even be created/managed by the authors... I believe microsoft will be interested in creating and maintaining the azure service bus transport...

[kamilmysliwiec]

Instead, I think it will be better to refactor the microservice package a bit, so it will be easier to add your own transport.

Actually, it's possible already. We are tracking the docs here nestjs/docs.nestjs.com#113

[shlomiassaf]

Yep, I saw the code and I saw it's possible but there's a lot of interface to implement with not documentation so I wouldn't know where to start and what there is or isn't

Great job on the docs (and on the project in general!)

[ceoro9]

A lot of custom transport strategies is of course a good thing. But I don't think we should keep them all here in main repo.

At first because soon we just are not gonna be able to support all this bloated codebase. And its size is not even the main reason. The main reason, we should care about is code. If you take a look on current existing custom transports, for example kafka, code looks not really debuggable, 'cause typings are lost, we are loading package dynamically, 'cause kafkajs is not explicitly listed in the dependencies.

let kafkaPackage: any = {};

class ServerKafka {

  public constructor(...) {
    // ...
    kafkaPackage = this.loadPackage('kafkajs', ServerKafka.name, () =>
      require('kafkajs'),
    );
  }
}

Although there is a @nestjs/microservices/external/kafka.interface.ts file with external typings, mostly for configuration, so the consumer of this transport at least gets type-safety on initialization. But anyway supporting all this external typings and dynamically loaded packages with no typings is unnecessary overhead.

So my point is, that we should focus on refactoring and implementing more flexible and transparent interface for custom transports and their usage, instead of trying to integrate every single type of transport strategy into the exiting codebase.

[vanishdark]

it would be nace a transporter to KubeMQ. since kubernets is coming bigger day by day, and kubeMQ it's awesome solution for kubernets use

[johnbiundo]

Just a quick note re: documenting custom transporters. I just published (the first 3 parts of) an in-depth tutorial on building custom transporters. https://dev.to/nestjs/part-1-introduction-and-setup-1a2l

Will be working on some official docs soon.

[zastrozzi]

We've been building a reusable SawtoothModule for building Hyperledger Sawtooth apps using NestJS. Sawtooth uses ZeroMQ with Protobuf for internal communication between components. @kamilmysliwiec do you want to add ZeroMQ to that list? At this stage our custom transporter only supports a Dealer/Router pattern but I'm happy to put together a PR at some point.

[jtomaszewski]

Not sure if this "feature request" fits here but just want to add my 3 cents about "what I'd like to have in NestJS":

I'd like to have some kind of async jobs module like the one in rails: https://edgeguides.rubyonrails.org/active_job_basics.html . So you could schedule a job (add and receive a message) very easily with API like JobsService.perform(...) JobsService.performLater(...) ; and with the transport mechanism being abstracted away.

I know there's @nestjs/bull package already, but it's coupled with bull. What if I want to keep my queue in SQS, Postgres, or just in-memory of the server? Also, what if I don't have an ongoing server instance (and instead, have a AWS Lambda that is launched whenever SQS event appears)?

I'm writing now some code by myself that:

1. Lets me handle messages from SQS queue:

# src/jobs-aws-lambda.ts
import 'source-map-support/register';
import { INestApplication } from '@nestjs/common';
import { NestFactory } from '@nestjs/core';
import { SQSHandler } from 'aws-lambda';
import { AppModule } from './app.module';
import { JobsQueueMessageHandlerService } from './jobs/jobs-queue-message-handler/jobs-queue-message-handler.service';

let cachedApp: INestApplication;
async function initApp(): Promise<INestApplication> {
  if (cachedApp) {
    // eslint-disable-next-line no-console
    console.log('Using cached server');
    return cachedApp;
  }

  // eslint-disable-next-line no-console
  console.log('Bootstraping server');
  cachedApp = await NestFactory.create(AppModule);

  return cachedApp;
}

export const handler: SQSHandler = async (event, context) => {
  const app = await initApp();
  const handlerService = app.get(JobsQueueMessageHandlerService);
  await handlerService.handleSqsEvent(event, context);
};

2. Lets me easily to run now/later an async job:

import { Injectable } from '@nestjs/common';
import { SQS } from 'aws-sdk';
import { Job } from './job.interface';

@Injectable()
export class JobsService {
  private _sqs?: SQS;

  private readonly queueUrl = process.env.JOBS_SQS_QUEUE_URL!;

  get sqs(): SQS {
    if (!this._sqs) {
      this._sqs = new SQS();
    }
    return this._sqs;
  }

  async perform(job: Job): Promise<void> {
    switch (job.type) {
      case 'LOG_TEST_MESSAGE':
        return new Promise<void>(resolve => {
          setTimeout(() => {
            // eslint-disable-next-line no-console
            console.log('LOG_TEST_MESSAGE job has been done!');
            resolve();
          }, 3000);
        });

      default:
        throw new TypeError(`Unknown job: ${JSON.stringify(job)}`);
    }
  }

  async performLater(job: Job): Promise<void> {
    // TODO on development environment, just run `.perform()` locally
    await this.sqs
      .sendMessage({
        QueueUrl: this.queueUrl,
        MessageBody: JSON.stringify(job),
      })
      .promise();
  }
}

Have you considered on the nestjs roadmap, to somehow make it possible in the future to wire up those different transport mechanisms with queues and task scheduling?

I imagine that in some future, you could choose and setup your queue transport mechanism AND add messages to it just by downloading some nest adapter module and configuring it, without the need of writing that custom code which I showed above 😃

[panoti]

I have tried implement custom strategy for NATS streaming. You can review my work here https://github.com/nestjs-ex/stan-strategy. I hope it can be added to @nestjs/microservice package.

[damianesteban]

Hello folks. So I originally opened the issue about SQS and I'd like to help out.

[damianesteban]

I'm curious for the AWS based transports, how many are using these running inside containers or VMs, vs processing these event sources in Lambda?

We are using Lambda as our deployment model, with Nest, and it has been working well. In that approach, processing Kinesis, SNS, SQS, EventBridge, has become trivial. We simply use NestFactory.createApplicationContext and then pass the event and context into a service class that handles the logic.

So in my case, I'm less interested in a Kinesis, SQS, SNS, EventBridge specific transport, and more interested in creating a better way for Nest to handle incoming API Gateway events in a way that does nto require spinning up an Express server in the background, and passing it over a socket (al. a aws-serverless-express). It's unnecessary overhead (which also has odd side effects), and if our experiance with using Nest to handle stream/queue/transport processing has taught us anything, Nest should be flexible enough to also handle incoming API Gateway events without Express.

Something I'd like to work on is an AWS Transports package for NestJS. Originally I was going to add SQS first but I think I'll try out EventBridge or SNS. Unless anyone has any other suggestions.

[uri-chandler]

For anyone interested:
I couldn't find a full start to finish guide on how to add my own custom transport layer strategy - so I did some digging in the source code, and ended up writing this guide:

Custom Transport Layers in NestJS
https://medium.com/@uri_chandler/custom-transport-layers-in-nestjs-5a913d9a383f

#hope_this_helps_someone (:

[sebastian-schlecht]

Any updates on this?

[dls314]

[edit: Just realized that I've duplicated what @jtomaszewski asked above, apologies]

Is there a transport strategy that uses an in memory transport; perhaps a stream?

I've successfully used the TCP transport in an SQS lambda handler, but I'd love to avoid the TCP and serialization overhead between the lambda entry point and the microservice controller.

[p-fedyukovich]

I have implemented Google PubSub transport #6200, could somebody have a look?

[kamilmysliwiec]

I gave this idea a second thought and I agree with @ceoro9 on this. Supporting external typings and dynamically load packages is simply unnecessary overhead. Maintaining all the existing transporters has been a pain already, and adding more right into the @nestjs/microservices package will make it even more bloated. Anyone can actually see from this thread that there are literally dozens of different tools/services (Kinesis, SQS, SNS, Thrift, GCP PubSub, Azure Service Bus, Redis Streams, KubeMQ ZeroMQ, the sky is the limit here..) we could integrate and that would make all these strategies almost impossible to maintain in the future (it would be way too much effort to even keep up with potential breaking changes & supporting platform-specific features, not even mentioning a regular development & replying to developers issues - this isn't something we could afford).

Instead, I've added a new Custom transporters chapter to the docs https://docs.nestjs.com/microservices/custom-transport where you can find brief instructions on how you can integrate any tool/external service with NestJS microservices. If you want to dive deeper, you can also check out @johnbiundo article series:

1. NestJS Microservices in Action
2. Advanced NestJS Microservice

For most transporters (especially those that don't require request-response communication support), what's in the docs should be sufficient (https://docs.nestjs.com/microservices/custom-transport#creating-a-strategy). If you need a custom transporter (a strategy not provided ootb by the framework), you should be able to do this by following the docs instructions and interacting with the official SDK (of the tool/service you want to use).

I would love to see any integrations created & open-sourced by the community so if you have anything to share, please do so here!

[p-fedyukovich]

Hi all, I have implemented GCP Pub/Sub microservice transport, feel free to test it out

[TrejGun]

example of getting events from Etherium blockchain is here
https://trejgun.github.io/articles/ethereum-server-for-nestjs

[TrejGun]

just in case. I have updated SQS transport
https://github.com/GemunIon/nestjs-sqs

[ali-master]

We are running the Redis by Sentinel for HA and control the failover by two servers and we can't connect to multi Redis connections now. What is your suggestion?

As I realized, the node-redis now supports the multi-cluster configuration by the @nestjs/microservice package doesn't support this option.

https://github.com/redis/node-redis/blob/master/docs/clustering.md#createcluster-configuration

[exsesx]

Any updates on AWS SQS transport integration?

[micalevisk]

@exsesx not under nestjs scope, see: #3960 (comment)