message-processing
v0.0.6
Published
message processing
Maintainers
minhduc140583Readme
message-processing
A lightweight, transport-independent framework for building reliable message processing pipelines in TypeScript.
message-processing separates message transport from message processing, allowing you to build clean, testable, and production-ready event-driven applications.
The library works with any message broker, including:
- ActiveMQ: activemq, to wrap and simplify rhea
- RabbitMQ: rabbitmq-transport, to wrap and simplify amqplib
- Apache Kafka: kafka-plus, to wrap and simplify kafkajs
- Google Pub/Sub: google-pubsub, to wrap and simplify @google-cloud/pubsub
- NATS: nats-plus, to wrap and simplify nats
- IBM MQ: ibmmq-plus, to wrap and simplify ibmmq
- Redis: redis-messaging, to wrap and simplify redis
- Amazon SQS
- Azure Service Bus
- Custom message transports
The transport library is responsible for receiving and sending messages. This library is responsible for processing them.
Why message-processing?
Most messaging libraries only provide APIs to publish and consume messages.
Production applications usually need much more:
- Deserialize messages
- Validate data
- Execute business logic
- Retry transient failures
- Retry after a delay
- Handle permanent failures
- Log processing
- Dead Letter Queue (DLQ)
Instead of implementing these concerns repeatedly in every consumer, message-processing provides reusable processing components.
Examples:
- redis-messaging-sample: An example to consume message from Redis.
- rabbitmq-sample: An example to consume message from rabbitmq.
- activemq-sample: An example to consume message from activemq.
- pubsub-sample: An example to consume message from pubsub.
- ibmmq-sample: An example to consume message from ibmmq.
- kafka-sample: An example to consume message from kafka.
- nats-sample: An example to consume message from nats.
- sqs-sample: An example to consume message from AWS sqs.
Architecture
Processor (Immediate Retry)
Processor performs retries inside the current process. It is suitable for transient failures such as temporary database errors or short network interruptions.
Message Broker
│
▼
Message Consumer
│
▼
Processor
│
▼
JSON Deserialization
│
▼
Message Validation
│
Valid? ────────────────────┐
│ No │
│ Yes ▼
│ Error Handler
▼
Business Logic
│
▼
Write to Database
│
┌─────────────┴─────────────┐
│ │
Success Failure
│ │
▼ ▼
Done Immediate Retry
│
┌──────────────┴──────────────┐
│ │
Success Retry Exhausted
│ │
▼ ▼
Done Error HandlerRetryProcessor (Retry Queue)
RetryProcessor republishes failed messages to a retry queue. It is designed for failures that are expected to last longer, such as cloud service outages or unavailable external systems.
Message Broker
│
▼
Message Consumer
│
▼
RetryProcessor
│
▼
JSON Deserialization
│
▼
Message Validation
│
Valid? ──────────No──────────┐
│ │
│ Yes ▼
│ Error Handler
▼
Business Logic
│
▼
Write to destination (For example: Database)
│
┌─────────────┴─────────────┐
│ │
Success Failure
│ │
▼ ▼
Done Increase Retry Count
│
▼
Retry Count < Limit?
│ │
│ Yes │ No
│ │
▼ ▼
Publish to Retry Queue Error Handler / Dead Letter Queue
│
│
▼
Delayed Consumer
│
└───────────────► RetryProcessorFeatures
- Transport independent
- Automatic JSON deserialization
- Message validation
- Generic message processing
- Immediate retry
- Delayed retry using retry queues
- Dead Letter Queue support
- Configurable retry limits
- Retry count management
- Error handling
- Logging
- Strongly typed APIs
Installation
npm install message-processingor
yarn add message-processingConcepts
The library separates responsibilities into independent components.
| Component | Responsibility | |----------------|--------------------------------------------------------------------------------------------| | Processor | Coordinates the processing pipeline, performs immediate retries inside the current process | | RetryProcessor | Coordinates the processing pipeline, republishes failed messages to a retry queue | | validate | Validates incoming messages | | write | Executes business logic | | retry | Sends failed messages to a retry queue | | handleError | Handles permanently failed messages | | RetryWriter | Retries immediately with delays | | RetrySender | Retries sending operations |
Two Retry Strategies
The library supports two different retry models.
1. Processor
Processor performs immediate retries inside the current process.
This strategy is suitable for temporary failures such as:
- MySQL deadlock
- Temporary network timeout (For example: cloud services, such as Google Firestore)
- Database connection reset
- Short infrastructure interruptions
Consumer
↓
Processor
↓
Business Logic
↓
Database
↓
Failure
↓
Retry immediately
↓
SuccessExample:
const processor = new Processor(
writer,
validator,
[1000, 5000, 10000]
);Immediate Retry Configuration
Immediate retry is useful for temporary failures such as:
- Database unavailable
- Temporary network issues
- HTTP timeout
- Cloud services, such as Google Firestore
- When Application cannot write data to Google Firestore, normally the network between Application and Google Cloud is down
- If Application retry after 1 second, normally it is still down
- The best approach is to retry after 30 seconds, then 60 seconds, then 180 seconds
const delays = [
30000,
60000,
180000
];
await writeWithRetry(
order,
saveOrder,
delays
);Retry sequence:
Try
↓
Wait 30 seconds
↓
Retry
↓
Wait 60 seconds
↓
Retry
↓
Wait 180 seconds
↓
RetryDelayed Retry Configuration
Delayed retries are usually much longer.
Example:
const retries = createRetry({
1: 60000,
2: 180000,
3: 360000
});This strategy avoids repeatedly retrying when an external service is unavailable.
2. RetryProcessor
RetryProcessor republishes failed messages to a retry queue.
This strategy is recommended when failures are expected to last longer.
Typical scenarios include:
- Google Firestore unavailable
- External REST API unavailable
- Third-party service outage
- Cloud infrastructure failure
Instead of retrying immediately, the message is published to a retry queue.
Consumer
↓
RetryProcessor
↓
Business Logic
↓
Failure
↓
Retry Queue
↓
Delayed Consumer
↓
Business LogicRetry counts are stored inside message headers.
Once the retry limit is exceeded, the message can be forwarded to a Dead Letter Queue.
Example:
const processor = new RetryProcessor(
writer,
retryService.retry,
validator,
errorHandler.error,
console.error,
console.log,
5
);Retry Count
Retry queues automatically maintain retry counts.
Message
↓
retryCount = 1
↓
retryCount = 2
↓
retryCount = 3
↓
Dead Letter QueueThe retry header name is configurable.
Default:
retryCountValidation
Messages are validated before business logic executes.
Receive Message
↓
Deserialize JSON
↓
Validate
↓
Business LogicInvalid messages never reach the database.
async function validateOrder(order: Order) {
const errors = [];
if (!order.id) {
errors.push({
field: "id",
code: "required"
});
}
return errors;
}Error Handling
Permanent failures can be handled by a custom error handler.
Typical implementations include:
- Dead Letter Queue
- Error database
- Audit log
- Notification service
Example:
const errorHandler = new ErrorHandler(console.error);Logging
Optional logging hooks are available throughout the processing pipeline.
console.log
console.erroror your own logging framework.
Other Functions
RetryWriter
RetryWriter wraps any write operation with retry logic.
const writer = new RetryWriter(
saveOrder,
[30000, 60000, 180000]
);
await writer.write(order);RetrySender
RetrySender wraps any send operation.
const sender = new RetrySender(
publishOrder,
[30000, 60000]
);
await sender.send(order);Dead Letter Handler
After the retry limit is exceeded, the library invokes the configured error handler.
async function saveDeadLetter(
order: Order,
headers?: StringMap
) {
await repository.save(order);
}Possible implementations include:
- Dead Letter Queue
- Database
- Log file
- Email notification
- Monitoring system
Helper Functions
write()
Executes business logic and optionally sends failed messages to a retry queue.
await write(
processOrder,
order,
headers,
deadLetter,
retrySender,
5
);writeWithRetry()
Retries immediately using configurable delays.
await writeWithRetry(
order,
processOrder,
[30000, 60000, 180000]
);createRetry()
Creates a retry delay array from a numbered configuration object.
const delays = createRetry({
1: 30000,
2: 60000,
3: 180000
});Produces:
[30000, 60000, 180000]Transport Independence
The library has no dependency on any messaging system.
It can be used with:
- RabbitMQ
- Kafka
- ActiveMQ
- NATS
- Redis Streams
- Amazon SQS
- Google Pub/Sub
- Azure Service Bus
or any custom message transport.
Typical Architecture
RabbitMQ Transport
↓
Consumer
↓
Processor
↓
Business Logic
↓
Writer
↓
Destination (For example: Database)or
Kafka Transport
↓
Consumer
↓
RetryProcessor
↓
Business Logic
↓
Retry Queue
↓
Destination (For example: Database)The processing layer remains identical regardless of the transport.
Related Projects
| Library | Responsibility |
|----------|------------------------------------------|
| health-service | Health checks |
| config-plus | Configuration |
| logger-core | Structured logging |
| validation-core | Data validation |
| rabbitmq-transport | RabbitMQ transport and Health Check |
| activemq | ActiveMQ transport and Health Check |
| kafka-plus | Kafka transport and Health Check |
| google-pubsub | Google Pubsub transport and Health Check |
| nats-plus | NATS transport and Health Check |
| ibmmq-plus | IBM MQ transport and Health Check |
| redis-messaging | Redis Pubsub transport and Health Check |
| mysql2-core | MySQL access and Health Check |
| mongodb-kit | MongoDB access and Health Check |
License
MIT