A production-style RabbitMQ sample application demonstrating how to build a clean, layered message processing system using the core-ts ecosystem.
This sample is not just a RabbitMQ producer/consumer example.
It demonstrates how to separate:
- Message transport
- Message processing
- Validation
- Retry
- Persistence
into independent layers.
HTTP
│
▼
Application Context
│
▼
RabbitMQ Consumer
│
▼
Message Processor
│
▼
JSON Deserialization
│
▼
Message Validation
│ No
Valid ? ────────────────────────┐
│ │
│ Yes ▼
│ Error Handler
▼ (or Discard)
Business Logic
│
▼
Write to Database
│
┌─────────┴─────────┐
│ │
Success │ │ Failure
│ │
▼ ▼
Done Retry Strategy
│
┌─────────────┴─────────────┐
│ │
Retry Succeeds Retry Limit Reached
│ │
▼ ▼
Done Error Handler / Dead Letter Queue
The sample demonstrates a complete message processing pipeline rather than individual RabbitMQ operations.
- RabbitMQ producer
- RabbitMQ consumer
- Typed messages
- Message validation
- Automatic JSON deserialization
- Retry processing
- MySQL persistence
- Health check endpoint
- Layered architecture
- Production-oriented project structure
- TypeScript
- RabbitMQ
- MySQL
- Express
- amqplib
src/
├── config.ts
├── context.ts (acts as a composition root)
├── index.ts (start the application)
├── user/
│ ├── user.ts
│ ├── port.ts
│ ├── message-transport.ts
│ ├── processor.ts
│ ├── validator.ts
│ ├── writer.ts
│ ├── retry-writer.ts
│ └── error-handler.ts
└── ...
The exact structure may vary, but responsibilities remain separated.
The createContext() function acts as a composition root.
It wires together:
- RabbitMQ
- Health service
- Validation
- MySQL
- Retry
- Logging
- Processor
without mixing business logic.
That's exactly where dependency injection should happen.
This sample demonstrates how several core-ts libraries work together.
| Library | Purpose |
|---|---|
rabbitmq-transport |
Publish and subscribe to RabbitMQ |
message-processing |
Retry and error handling |
mysql2-core |
Write data into MySQL |
validation-core |
Validate incoming messages |
health-service |
Health endpoint |
logger-core |
Structured logging |
config-plus |
Configuration management |
Each library focuses on a single responsibility.
That demonstrates the intended layering very well.
- In the real application, we do not have message producer in this application.
- The message producer is in another microservice.
- This application consumes the message only.
HTTP Request
↓
RabbitMQ Sender
↓
RabbitMQ Queue
The sender serializes the message and publishes it to RabbitMQ.
RabbitMQ Queue
↓
Consumer
↓
Processor
The consumer only receives messages.
Business processing is delegated to the Message Processing library.
The processor executes the complete workflow.
Receive Message
↓
Deserialize JSON
↓
Validate
↓
Business Logic
↓
Write to Database
If processing succeeds, the message is acknowledged.
The sample demonstrates immediate retry for transient failures.
Write Database
↓
Failure
↓
Wait
↓
Retry
↓
Success
Typical retry scenarios include:
- temporary database outage
- deadlock
- network timeout
- transient infrastructure errors
Retry intervals are configurable.
Example:
const retries = [5000, 10000, 20000]Messages are validated by validation-core before reaching business logic.
Typical validation includes
- Required fields
- String length
- Email format
- Business constraints
Invalid messages are rejected before reaching the database.
RabbitMQ
↓
Processor
↓
Validator
↓
Business Logic
Messages are automatically converted into TypeScript objects.
RabbitMQ
↓
JSON
↓
Order
↓
Processor
Business logic receives strongly typed objects instead of raw strings.
Business logic writes data using mysql2-core.
Processor
↓
Repository
↓
MySQL
RabbitMQ code never interacts directly with the database.
The sample exposes a health endpoint.
GET /health
The endpoint checks:
- RabbitMQ connectivity
- MySQL connectivity
They can easily be extended to include additional infrastructure services.
Example response
{
"status": "UP",
"details": {
"rabbitmq": {
"status": "UP"
},
"mysql": {
"status": "UP"
}
}
}One of the goals of this sample is to demonstrate proper layering.
Responsible for:
- Producer
- Consumer
- Header mapping
- Health checking
Responsible for:
- Processing pipeline
- Validation
- Retry
- Error handling
- Logging
- JSON deserialization
Responsible only for business logic.
Process Order
Save Customer
Import Product
Send Notification
Business services remain independent of RabbitMQ.
Instead of embedding business logic inside RabbitMQ consumers,
Consumer
↓
Business Logic
↓
Database
↓
Retry
↓
Validation
the sample separates every concern.
Consumer
↓
Processor
↓
Validation
↓
Retry
↓
Business Logic
↓
Writer
Each layer has a single responsibility.
This makes the application:
- easier to maintain
- easier to test
- easier to replace infrastructure
- easier to extend
This sample demonstrates production-oriented practices including:
- Layered architecture
- Strong typing
- Validation
- Retry processing
- Logging
- Health monitoring
- Infrastructure abstraction
Additional production features such as dead-letter queues, delayed retry queues, metrics, and distributed tracing can be added without changing the application architecture.
MIT