RabbitMQ-fp
Lets feed our Rabbit' nicely
This repository contains a wrapper over amqplib written in Typescript with an accent of functionnal programming, using fp-ts. It will handle high-level features like RPC without hassle.
Warning This library is still heavily being worked on, but no breaking changes on the API are planned.
Feature highlights:
- Built-in RPC support
- Functional typings with
fp-ts - Automatic error recovery using
amqp-connection-manager
Getting started
Install the package from npm:
npm install --save-exact @mansagroup/rabbitmq-fp
This library has
fp-tsas peer dependency, to match your project's version.
Create a setup function
This library support automatic recovery on AMQP connection or channel error. However, a newly created channel will not inherit the configuration from the previous one. This means that each new channel must be reconfigured (asserting exchanges, queues, binding queues, etc...).
To solve this, when creating an adapter, you must pass a setup function which takes the created channel and returns this same channel. This can easily represented as a fp-ts's flow method:
import { flow } from 'fp-ts/function';
import * as TE from 'fp-ts/TaskEither';
import { SetupFn } from '@mansagroup/rabbitmq-fp';
const setupFn: SetupFn.Fn = flow(
SetupFn.assertExchange('my-exchange'),
SetupFn.assertQueue('my-queue'),
SetupFn.bindQueue('my-queue', 'my-exchange', 'my-routing-key'),
);
This function will be invoked every time a new channel is created.
Create an adapter
An adapter is the actual brain of this library. It is the high-level bridge between your code and the underlying amqplib library. It will requires your previously created setup function but also a logger:
import { Logger, createRabbitMQAdapter } from '@mansagroup/rabbitmq-fp';
const logger: Logger = {
info: (msg, extra) => {},
// This for every log level
};
const adapter = await createRabbitMQAdapter(
'amqp://username:password@host:port',
setupFn,
{
logger,
},
)();
Create your consumer
To keep this simple, we will setup a simple event consumer which will print hello {greetings} every time a message is published. A consumer is a function which takes a payload and returns a TaskEither.
Note If the
TaskEitheris aLeft, then the message will benack, otherwise it will beack.
import { EventHandler } from '@mansagroup/rabbitmq-fp';
import { pipe } from 'fp-ts/function';
import * as IO from 'fp-ts/IO';
interface Payload {
greetings: string;
}
const consumer: EventHandler<Payload> = (payload) =>
pipe(
`hello ${payload.greetings}`,
IO.of,
IO.map(console.log),
IO.map(TE.right),
)();
await adapter.consumeEvent('my-queue', consumer)();
Publish your message
Finally, after that your consumer is created and ready, you can publish your first message to see the consumer invoked:
await adapter.publish<Payload>('my-exchange', 'my-routing-key', {
greetings: 'Bob',
})();
Everything together
Now, if we pull everything together, we could have a flow like the one from the everything-together.ts example.
Examples
- Everything together: the previous section example in code
- RPC: consume and request over RPC methods
License
This project is MIT licensed.
Contributors
Thanks goes to these wonderful people (emoji key):
 Jérémy Levilain |
This project follows the all-contributors specification. Contributions of any kind welcome!
5 Sep 6, 2022
3 May 18, 2022
4 May 3, 2022
14 Jan 3, 2023
9 Dec 6, 2022
6 Jul 2, 2022
23 Sep 21, 2022
4 Sep 22, 2022
18.1k Jan 4, 2023
6 Feb 4, 2022
40 Dec 6, 2022
22 Dec 12, 2022
32 Dec 14, 2022
18 Dec 4, 2022
16 Nov 17, 2022
21 Oct 25, 2022
1.4k Dec 29, 2022
9.7k Jan 3, 2023
8 Sep 2, 2022