carotte-amqp

npm install carotte-amqp

A tiny amqp.node wrapper to be used when you don't want to deal with the low-level RabbitMQ handling.

It is part of a lightweight microservice framework that we are cooking here at Cubyn. carotte-amqp requires node.js >= 6

Features:

• Built-in RPC (single and multiple answers)
• Automatic backoff/retry and dead-letter handling
• Compatible with direct/topic/fanout exchanges
• Can aggregate statistics about your queues
• Tiny (depends on amqplib, debug, and puid)
• Built-in distributed tracing, see doc
• Provide your own transport to log every microservice message
• Allow to share your environments and services between developers without launching the whole stack

Sample usage

Only basic usage is covered here, you can browse the examples folder on the repo for more use-cases.

Module configuration

Here is a basic sample configuration using default interval values for reference (each one is optional):

const carotte = require('carotte-amqp')({
    serviceName: require('package.json').name, // your service name
    host: 'amqp://localhost:5672', // amqp host name
    enableAutodoc: false, // answer to autodocumentation fanout requests
    enableDeadLetter: false, // should failing messages be put in dead-letter?
    deadLetterQualifier: 'direct/dead-letter', // dead-lettered messages will be sent here
    autoDescribe: false // provide a :describe queue for your function
});

// SSL connection
// amqplib example: https://github.com/squaremo/amqp.node/blob/main/examples/ssl.js#L37-L44
const carotte = require('carotte-amqp')({
    // ...
    host: 'amqps://localhost:5672',
    connexion: {
        cert: fs.readFileSync('../etc/client/cert.pem'),
        key: fs.readFileSync('../etc/client/key.pem'),
        // cert and key or
        // pfx: fs.readFileSync('../etc/client/keycert.p12'),
        passphrase: 'MySecretPassword',
        ca: [fs.readFileSync('../etc/testca/cacert.pem')]
    }
});

Direct exchange

Without consumer answer (publish)

This is the most basic usage of AMQP, a publisher emit a message in the broker and there is (at least) one consumer to consume it:

carotte.subscribe('direct/hello.world', ({ data }) => {
    console.log(`hello ${data.firstname}`);
});

carotte.publish('direct/hello.world', { firstname: 'Gabe' });

With consumer answer (invoke)

The lib provides you a way to wait for an answer with the invoke method:

carotte.subscribe('direct/hello.world', ({ data }) => {
    return `hello ${data.firstname}`;
});

carotte.invoke('direct/hello.world', { firstname: 'Gabe' })
    .then(data => {
        console.log(data);
    });

Topic

Topic allows RabbitMQ to route a message to multiple consumers:

carotte.subscribe('topic/hello.*/function1', ({ data }) => {
    console.log(`function 1 got it!`);
});

carotte.subscribe('topic/hello.world/function2', ({ data }) => {
    console.log(`function 2 got it!`);
});

carotte.publish('topic/hello.world', { firstname: 'Gabe' });

Fanout

Fanout will broadcast a message on all consumers:

carotte.subscribe('fanout/a', ({ data }) => {
    console.log(`function 1 got it!`);
});

carotte.subscribe('fanout/b', ({ data }) => {
    console.log(`function 2 got it!`);
});

carotte.publish('fanout', { firstname: 'Gabe' });

Handling multiple answers (topic/fanout with RPC)

This is done using the parallel function:

carotte.subscribe('fanout/a', ({ data }) => {
    return `function 1 answer`;
});

carotte.subscribe('fanout/b', ({ data }) => {
    return `function 2 answer`;
});

carotte.parallel('fanout', { firstname: 'Gabe' }, (err, data) => {
    if (!err) {
        console.log(data);
    }
});

Custom transport

To log every incoming and outgoing message, as well as execution time, you can provide your own custom transport to carotte-amqp

const winston = require('winston');
const carotte = require('carotte-amqp')({
    transport: winston
})

Distributed tracing

Propagating a context is easy with carotte-amqp, the lib will take care of it for you.

carotte.subscribe('direct/second-route', ({ data, context }) => {
    console.log(context.requestId); // prints the requestId
    context.count++;
});

carotte.subscribe('direct/first-route', ({ data, context, invoke }) => {
    context.requestId = randomString(10).toString('hex');
    context.count = 0;
    return invoke('direct/second-route', {});
});


carotte.invokeWithFullResponse('direct/first-route', {})

    .then(({ data, context }) => {
        console.log(context.requestId); // prints the request ID
        console.log(context.count); // prints 1
    });

The context will be logged in every log of the wrapper, if you provide a custom transport.

Automatic description

In your microservice architecture, you sometimes want to implement automatic discovery and get a service check for one of your functions or get some metrics of a service. This is done using the meta parameter in your subscribe functions:

const carotte = require('carotte-amqp')({ autoDescribe: true });

carotte.subscribe('direct/increment', ({ data } => {
    return data + 1;
}), {
    description: 'awesome function that increments',
    version: 1,
    requestSchema: {
        // JSON SCHEMA HERE
    },
    responseSchema: {
        // JSON SCHEMA HERE
    }
});

carotte.invoke('direct/increment:describe')
    .then(data => {
        console.log(data.description);
    });

This structure is also used in carotte-dashboard to auto-document your microservices architecture. You can find more information about how it works on the dashboard repository.

Automatic Retry

You can set a retry strategy to automatically catch exceptions and execute remote executions again. This is done using the meta parameter in your subscribe functions.

| property | type | required | description | | -------- | ------ | -------- | ----------- | | max | int | y | maximum number of attempt | | interval | int | n | interval between attempts (depends on strategy) | | strategy | string | n | either direct, fixed or exponential | | jitter | int | n | maximum random delay to add between attempts |

If you don't set any (no presence of retry property in meta), the default one will be taken into account:

• max: 5
• strategy: 'direct'
• interval: 0
• jitter: 0

const carotte = require('carotte-amqp')({});

carotte.subscribe(
    'direct/increment',
    ({ data } => data + 1),
    {
        retry: {
            max: 2,
            strategy: 'exponential',
            interval: 5,
            jitter: 10
        }
    }
);

Working together

When multiple devs are working on multiple microservices, you can use environment variables to be able to communicate with each other. To do so, the developers must set the CAROTTE_DEBUG_TOKEN env variable to a shared string before launching their services.

carotte will then automatically reach each-others services.

Typing

The typing file was generated by tsconfig.json. It should not longer be used because the types are not exactly correct (the library is too complex to have a correct typing). If the public API changes, directly update the index.d.ts file.