This module implements an AMQP pub/sub + RPC scalable & fault tolerant interface for Nest.js.
- Configurable as a module + programatically add connections on the fly.
- 100% Rxjs based with typings.
- Supports multiple connections.
- Auto-magic reconnection. ๐
- Offline queueing support to handle disconnects.
- Handles exchange, queue, and routing key bindings for you.
- Pub/sub + RPC support.
- Shutdown signal trap for graceful shutdown. ๐ฃ
- Configurable, verbose, logging facility.
- Hackable and extensible. ๐งฐ
- Demo implementation.
npm install @nestjs.pro/amqp
import { Module } from '@nestjs/common';
import { AppService } from './AppService';
import { AMQPModule } from '@nestjs.pro/amqp/dist/AMQPModule';
import { AMQPLogLevel } from '@nestjs.pro/amqp/dist/AMQPLogLevel';
@Module({
imports: [
AMQPModule.forRoot({
//
// no secrets, log what is actually happening under the hood!
//
logLevel: AMQPLogLevel.DEBUG,
//
// add multiple connections if required..
//
connections: [
{
name: 'one',
uri: 'amqp://rabbitmq:rabbitmq@localhost:5672',
exchange: {
name: 'test-1',
type: 'topic',
options: {
durable: true
// all other amqplib options supported..
}
},
queues: [
{
name: '1',
routingKey: '111',
createBindings: true,
options: {
durable: false
// all other amqplib options supported..
}
}
// add more queues..
]
}
// add more connections to your hearts desire..
]
})
],
providers: [ AppService ]
})
export class AppModule {
}
See app/src/AppService.ts for a complete implementation.
import { Injectable } from '@nestjs/common';
import { AMQPService } from '@nestjs.pro/amqp/dist/AMQPService';
import { AMQPConnectionStatus } from '@nestjs.pro/amqp/dist/AMQPConnectionStatus';
@Injectable()
export class AppService {
//
// Inject in to your service(s)..
//
public constructor(private readonly amqpService: AMQPService) {
//
// Retrieve the connection named "one" and do some magic..
//
amqpService.getConnection('one').subscribe(connection => {
//
// Now that we're connected, publish a message..
//
connection.queue.publishJSON('test-1', 111, { date: new Date(), rand: Math.random() });
// ..other fancy work happenin here..
//
// Optionally remove the exchange, queue, and routing keys if you want..
//
amqpService.tearDown().subscribe(() => {
console.log('DEMO: Tear down complete, all remaining exchange(s) and queue(s) removed! ๐');
//
// Disconnect gracefully now that we've cleaned things up..
//
amqpService.disconnect();
});
});
}
}
Pull down this repo and install the dependencies using pnpm:
If you do not have pnpm installed simply run the following:
npm install -g pnpm
Next we can compile the library out to the dist
directory:
npx tsc -w
Finally, we just need to start the demo implementation:
npm run start:dev
git clone https://github.com/mateothegreat/nestjspro-amqp
cd nestjspro-amqp
pnpm install
Enjoy!
__ _
____ ___ _____/ /_ (_)____ ____ _________
/ __ \/ _ \/ ___/ __/ / / ___/ / __ \/ ___/ __ \
/ / / / __(__ ) /_ / (__ ) / /_/ / / / /_/ /
/_/ /_/\___/____/\__/_/ /____(_) .___/_/ \____/
/___/ /_/