📦 MuncherJs

Um projeto criado em TypeScript para facilitar a criação de consumidores RabbitMQ, com suporte a métricas Prometheus, tracing com Elastic APM e um sistema de plugins para consumidores.

📄 Objetivo

Simplificar a integração com consumo de stream de menssagens via RabbitMQ

✨ Funcionalidades

Métricas: Adicionado métricas de consumo (mensagens processadas, tempo médio, CPU, Memoria, etc.) via prometheus
Monitoramento: Suporte para integração com APM do Elastic Search
Resiliencia: Adicionado estrategias de retentativas e DLQ
Long Process: Mecanismos de fallback para operações longas
Plugin/Play: Adicionado suporte para criação de plugins para uso em geral
ManagerConsumer: Controle de Skip, Stop, Retryable consumo de mensagens
Logger: Suporte de de incoporação do logger de sua escolha

💡 Roadmap Futuro

Healcheck para consumidores ativos
Exemplos de idempotency key com redis

⚡ Pipeline inteligente de consumo RabbitMQ

mensagens fluem em um ciclo estruturado de pré-processamento, processamento e pós-processamento — com controle total do contexto, resiliência, garantindo consistência e extensibilidade de ponta a ponta.

📊 Metricas

Esta lib utiliza Prometheus para coletar, armazenar e alertar sobre métricas em tempo real, garantindo alta disponibilidade e performance.

🔍 Métricas Coletadas:

Infraestrutura: CPU, memória, uso de disco, tráfego de rede.
Consumidor RabbitMQ: Total de mensagens processadas, Tempo de processamento de mensagens, Mensagens atualmente sendo processadas, Total de novas tentativas de processamento, Total de mensagens enviadas para DLQ

🚀 Application Performance Monitoring (APM)

Este lib utiliza APM para monitorar desempenho, rastrear transações distribuídas e identificar gargalos em tempo real.

🔍 Métricas Coletadas:

Erros corridos em tempo de execução
Latencia media do consumidor
Throughput por consumo de mensagens
CPU e Memoria utilizada

⚙️ Instalação

npm install muncherjs -S

🤓 Exemplo de uso

Simple

import { MuncherConsumerBase } from 'muncher'

class ExampleConsumer extends MuncherConsumerBase<Example> {

    protected async processMessage(ctx: IMessageContext<Example>): Promise<void> {
        console.log(ctx)
    }

}

const consumer = new ExampleConsumer({
    connection: {
        username: 'admin',
        password: 'admin'
    },
    queues: [{
        topic: 'Example.Queue'
    }]
})

await consumer.start()

process.on('SIGTERM', async () => {
    await consumer.stop();
    process.exit(0);
});

Simple Metadata shared

import { MuncherConsumerBase } from 'muncher'

class ExampleConsumer extends MuncherConsumerBase<Example> {

    protected async preProcess(ctx: IMessageContext<Example>): Promise<void> {
        ctx.metadata.new = "New Value in ctx" // Data shared in context
    }

    protected async processMessage(ctx: IMessageContext<Example>): Promise<void> {
        console.log(ctx.metadata.new)
    }

}

const consumer = new ExampleConsumer({
    connection: {
        username: 'admin',
        password: 'admin'
    },
    queues: [{
        topic: 'Example.Queue'
    }]
})

await consumer.start()

process.on('SIGTERM', async () => {
    await consumer.stop();
    process.exit(0);
});

Saidas de Error´s

import { MuncherConsumerBase } from 'muncher'

class ExampleConsumer extends MuncherConsumerBase<Example> {
    
    protected async processMessage(ctx: IMessageContext<Example>): Promise<void> {
        console.log(ctx)
    }

    // In Class Domain
    protected async onErrorProcess(ctx: IMessageContext<Example>, error: unknown): Promise<void> {
        console.log(ctx, error)
    }

}

const consumer = new ExampleConsumer({
    connection: {
        username: 'admin',
        password: 'admin'
    },
    queues: [{
        topic: 'Example.Queue'
    }]
})

// Out Class Domain
consumer.onError((err) => {
    console.log(err)
})

await consumer.start()

process.on('SIGTERM', async () => {
    await consumer.stop();
    process.exit(0);
});

Suporte ao APM

import { MuncherConsumerBase } from 'muncher'

class ExampleConsumer extends MuncherConsumerBase<Example> {
    
    protected async processMessage(ctx: IMessageContext<Example>): Promise<void> {
        console.log(ctx)
    }

}

const consumer = new ExampleConsumer({
    connection: {
        username: 'admin',
        password: 'admin'
    },
    queues: [{
        topic: 'Example.Queue'
    }],
    apm: {
        serverUrl: 'http://localhost:8200'
    }
})

await consumer.start()

process.on('SIGTERM', async () => {
    await consumer.stop();
    process.exit(0);
});

Suporte ao Prometheus

import { MuncherConsumerBase } from 'muncher'
import http, { Server } from 'http';

class ExampleConsumer extends MuncherConsumerBase<Example> {
    
    protected async processMessage(ctx: IMessageContext<Example>): Promise<void> {
        console.log(ctx)
    }

}

const consumer = new ExampleConsumer({
    connection: {
        username: 'admin',
        password: 'admin'
    },
    queues: [{
        topic: 'Example.Queue'
    }]
})

const server = http.createServer(async (req, res) => {
    if (req.url === '/metrics') {
        const { metrics, contentType } = await consumer.getMetrics();
        res.writeHead(200, { 'Content-Type': contentType });
        res.end(metrics);
    } else {
        res.writeHead(404);
        res.end();
    }
});

await consumer.start()

server.listen(9091, () => {
    console.log('🚀 Metrics server rodando em http://localhost:9091/metrics');
})

process.on('SIGTERM', async () => {
    await consumer.stop();
    process.exit(0);
});

Suporte para Plugins

import { MuncherConsumerBase, SkipMessageError } from 'muncher'

class PluginValidade extends PluginBase<unknown> {

    private schema: AnySchema

    constructor(schema: AnySchema) {
        super('PluginValidade');
        this.schema = schema
    }

    async preProcess(ctx: IMessageContext<unknown>): Promise<void> {
        const validate = this.schema.validate(ctx.)
        if (validate.error) {
            throw new SkipMessageError(this.name, validate.error.details)
        }
    }

    async initialize(consumer: IConsumer): Promise<void> {
        consumer.onError((err) => {
            console.log("plugin", err)
        })
    }
}

class ExampleConsumer extends MuncherConsumerBase<Example> {
    
    protected async processMessage(ctx: IMessageContext<Example>): Promise<void> {
        console.log(ctx)
    }

}

const consumer = new ExampleConsumer({
    connection: {
        username: 'admin',
        password: 'admin'
    },
    queues: [{
        topic: 'Example.Queue'
    }],
    plugins: [
        new PluginValidade(joi.object({
            name: joi.string()
        }))
    ]
})

consumer.withPlugin(new PluginValidade(joi.object({
    name: joi.string()
})))

await consumer.start()

process.on('SIGTERM', async () => {
    await consumer.stop();
    process.exit(0);
});

Error Types

Parâmetro	Descrição	Example
StopProcessingError	Consumo da menssagem e parado é marcado como erro	new StopProcessingError('msg', {...params})
SkipMessageError	Consumo da menssagem e finalizado é marcado como sucesso	new SkipMessageError('msg', {...params})
RetryableError	Consumo da menssagem e marcado para reprocessamento	new RetryableError('msg', {...params})

Configs

Parâmetro	Tipo	Obrigatório	Descrição
connection	`object`	✅	Conexão com RabbitMQ Valores
exchanges	`Array`	❌	Configuração de consumo Via Exchange Valores
queues	`Array`	❌	Configuração de consumo Via Queue Valores
binds	`Array`	❌	Configuração de consumo Via Binding Valores
plugins	`Array`	❌	Configuração de Plugins
serviceName	`string`	❌	Nome do serviço
prefetch	`string`	❌	Quantidade mensagens um consumidor pode processar ao mesmo tempo
logger	`number`	❌	Logger customizado Default value: `console`
DLQ	`object`	❌	Configuração de `DLQ` Valores
apm	`object`	❌	Configuração do apm Valores
retryPolicy	`object`	❌	Configuração de retryPolicy Valores

Connection

Parâmetro	Descrição	Exemplo
protocol	O protocolo a ser utilizado	Default value: 'amqp'
hostname	Nome do host usado para conexão ao servidor.	Default value: 'localhost'
port	Port used for connecting to the server.	Default value: 5672
username	Nome de usuário usado para autenticação no servidor.	Default value: 'guest'
password	Senha usada para autenticação no servidor.	Default value: 'guest'
locale	O local desejado para mensagens de erro. O RabbitMQ usa apenas en_US	Default value: 'en_US'
frameMax	O tamanho em bytes do quadro máximo permitido na conexão. 0 significa sem limite (mas como os quadros têm um campo de tamanho que é um inteiro sem sinal de 32 bits, ele é forçosamente 2^32 - 1).	Default value: 0x1000 (4kb)
heartbeat	O período do `heartbeat` da conexão em segundos.	Default value: 0
vhost	VHost usado.	Default value: '/'

DLQ

Parâmetro	Tipo	Obrigatório	Descrição
routingKey	`string`	✅	Nome da fila que receberá a mensagem.
exchange	`string`	❌	Nome da exchange que envia a mensagem.

Importante!: Caso não seja passado uma exchange o envio de dados ao DLQ será feito via sendToQueue

Cenário	Use `sendToQueue`	Use `publish`
Enviar direto pra uma fila específica?	✅	⚠️
Quer usar Exchange + RoutingKey?	⚠️	✅
Precisa de controle com DLX, fanout, topic routing?	❌	✅
Protótipo simples/teste?	✅	⚠️

Queues

Parâmetro	Tipo	Obrigatório	Descrição
topic	`string`	✅	Nome da fila que receberá a mensagem.
options	`object`	❌	Opções de Assert Valores
optionsConsumer	`object`	❌	Opções do consumidor Valores

Exchange

Parâmetro	Tipo	Obrigatório	Descrição
topic	`string`	✅	Nome da fila que receberá a mensagem.
type	`string`	✅	Tipo de roteamento da exchange. Valores
options	`object`	❌	Opções de Assert Valores
optionsConsumer	`object`	❌	Opções do consumidor Valores

Binding

Parâmetro	Tipo	Obrigatório	Descrição
`queue`	`string`	✅	Nome da fila que receberá a mensagem.
`exchange`	`string`	✅	Nome da exchange que envia a mensagem.
`routingKey`	`string`	✅	Chave usada pra rotear a mensagem. Pode ser `''`, `#`, `*.log`, etc.
`args`	`any`	❌	Argumentos adicionais (raramente usados; para casos avançados).

Exchange Type

Tipo	Comportamento
`direct`	Roteamento exato via `routingKey`.
`topic`	Roteamento por padrões (`*.log`, `#.error`).
`fanout`	Ignora `routingKey`, envia para todas as filas ligadas.
`headers`	Roteia por headers, não por `routingKey`. Mais raro e avançado.

Options.AssertQueue

Parâmetro	Tipo	Descrição
`durable`	`boolean`	Se `true`, a fila sobrevive a reinicializações do servidor.
`exclusive`	`boolean`	Se `true`, a fila é acessível apenas por esta conexão e será deletada ao desconectar.
`autoDelete`	`boolean`	Se `true`, a fila será removida automaticamente quando não tiver mais consumidores.
`messageTtl`	`number`	Tempo (ms) que a mensagem pode ficar na fila antes de ser descartada.
`expires`	`number`	Tempo (ms) que a fila pode ficar inativa antes de ser deletada.
`maxLength`	`number`	Máximo de mensagens que a fila pode conter.
`maxPriority`	`number`	Define níveis de prioridade de mensagens (quando usado).

Options.AssertExchange

Parâmetro	Tipo	Descrição
`durable`	`boolean`	Se `true`, a exchange sobrevive a reinicializações do servidor.
`internal`	`boolean`	Se `true`, a exchange não pode ser publicada diretamente, apenas usada por outras exchanges.
`autoDelete`	`boolean`	Se `true`, a exchange é deletada quando não houver mais filas vinculadas.
`arguments`	`any`	Usado para configurações adicionais específicas de tipo de exchange.

Options.Consume

Parâmetro	Tipo	Descrição
`consumerTag`	`string`	Identificador único do consumidor. Útil para cancelar ou rastrear consumo.
`noAck`	`boolean`	Se `true`, não exige ack manual. A mensagem é considerada entregue assim que recebida. ⚠️ Use com cuidado.
`exclusive`	`boolean`	Se `true`, somente esta conexão pode consumir da fila.
`priority`	`number`	Prioridade do consumidor. Maior valor consome primeiro (requer suporte da fila).
`arguments`	`any`	Argumentos adicionais (ex: filtros, extensões customizadas).

RetryConfig

Parâmetro	Tipo	Descrição
`maxRetries`	`number`	Número de retentativas
`backoff`	`number`	Tempo de espera entre tentativas

APM

Parâmetro	Tipo	Descrição
`serverUrl`	`string`	URL de conexão com APM
`apiKey`	`string`	A chave de API. Esta é uma alternativa ao secretToken. (Mais Informações)[https://www.elastic.co/docs/reference/apm/agents/nodejs/configuration#api-key]
`secretToken`	`string`	O token secreto. (Mais Informações)[https://www.elastic.co/docs/reference/apm/agents/nodejs/configuration#secret-token]
`cloudProvider`	`string`	Auto configuração para provedores Clound (Mais Informações)[https://www.elastic.co/docs/reference/apm/agents/nodejs/configuration#cloud-provider]

📫 Contribuições

Sinta-se à vontade para abrir issues ou enviar pull requests. Sugestões são sempre bem-vindas!

🤓 Contato

Desenvolvido por: Ismael Alves 🤓🤓🤓

Email: cearaismael1997@gmail.com
Github: github.com/ismaelalvesgit
Linkedin: linkedin.com/in/ismael-alves-6945531a0/

Name		Name	Last commit message	Last commit date
Latest commit History 20 Commits
examples		examples
src		src
.gitignore		.gitignore
eslint.config.js		eslint.config.js
jest.config.ts		jest.config.ts
jest.setup.ts		jest.setup.ts
package.json		package.json
readme.md		readme.md
tsconfig.json		tsconfig.json

ismaelalvesgit/muncher

Folders and files

Latest commit

History

Repository files navigation