npm package: @onlineapps/conn-infra-mq
Description: A promise-based, broker-agnostic client for sending and receiving messages via RabbitMQ
Author: OnlineApps Team
Latest version: 1.1.69

# @onlineapps/conn-infra-mq

[![Build Status](https://img.shields.io/github/actions/workflow/status/onlineapps/conn-infra-mq/nodejs.yml?branch=main)](https://github.com/onlineapps/conn-infra-mq/actions)
[![Coverage Status](https://codecov.io/gh/onlineapps/conn-infra-mq/branch/main/graph/badge.svg)](https://codecov.io/gh/onlineapps/conn-infra-mq)
[![npm version](https://img.shields.io/npm/v/@onlineapps/conn-infra-mq)](https://www.npmjs.com/package/@onlineapps/conn-infra-mq)

> Message queue connector with **layered architecture** for workflow orchestration, fork-join, and retry patterns. Built on top of RabbitMQ with clean separation of concerns. **Asynchronous workflow pattern only** - synchronous RPC patterns are not supported and not aligned with our architecture philosophy.

---

## 🚀 Features

- **Layered Architecture**: Clean separation into specialized layers (WorkflowRouter, QueueManager, ForkJoinHandler, RetryHandler)
- **Workflow Orchestration**: Decentralized workflow routing without central orchestrator
- **Fork-Join Pattern**: Parallel processing with result aggregation and built-in join strategies
- **Asynchronous First**: All communication is asynchronous (fire-and-forget), no synchronous blocking patterns
- **Automatic Retry**: Exponential backoff, dead letter queue management, configurable retry policies
- **Queue Management**: TTL, DLQ, auto-delete, temporary queues, exchange bindings
- **Promise-based API**: All operations return promises for clean async/await usage
- **Built-in Serialization**: JSON with custom error handling
- **Config Validation**: Strict schema validation via Ajv
- **Extensible Transport**: Clear separation between core logic and transport layer

---

## 📦 Installation

```bash
npm install @onlineapps/conn-infra-mq
# or
yarn add @onlineapps/conn-infra-mq
````

> Requires Node.js ≥12. For RabbitMQ usage, ensure an accessible AMQP server.

---

## 🏗️ Architecture

```
ConnectorMQClient (main orchestrator - for business services only)
    ├── BaseClient (core AMQP operations)
    ├── WorkflowRouter (workflow orchestration)
    ├── QueueManager (queue lifecycle management)
    ├── ForkJoinHandler (parallel processing)
    └── RetryHandler (error recovery & DLQ)
```

### WorkflowRouter - How It Works

**Purpose**: Handles workflow routing between services in a decentralized architecture.

**Key Methods**:
- `publishWorkflowInit(workflow, options)` - Publishes workflow to `workflow.init` queue (entry point)
- `publishToServiceWorkflow(serviceName, message, options)` - Routes to specific service's workflow queue
- `publishWorkflowCompleted(result, options)` - Publishes completed workflow to `workflow.completed` queue
- `consumeWorkflowInit(handler, options)` - Consumes from `workflow.init` (competing consumers pattern)
- `consumeServiceWorkflow(serviceName, handler, options)` - Consumes from service-specific workflow queue

**How It Works**:
1. **Gateway** publishes to `workflow.init` via `publishWorkflowInit()`
2. **Business services** (competing consumers) consume from `workflow.init` via `consumeWorkflowInit()`
3. **Services** route to next service via `publishToServiceWorkflow()`
4. **Final service** publishes completion via `publishWorkflowCompleted()`

**Note**: WorkflowRouter is part of `conn-infra-mq` connector (for business services). Infrastructure services (gateway) should use the underlying MQ client library directly, not the connector.

## 🔧 Quick Start

```js
'use strict';

const ConnectorMQClient = require('@onlineapps/conn-infra-mq');

(async () => {
  // 1. Create client with configuration
  const client = new ConnectorMQClient({
    host: 'amqp://localhost:5672',
    serviceName: 'my-service',
    queue: 'default-queue',
    durable: true,
    prefetch: 5,                        // Default prefetch count for consumers
    noAck: false,                       // Default auto-acknowledge = false
    retryPolicy: {                      // Optional reconnection policy (not enforced in v1.0.0)
      retries: 5,
      initialDelayMs: 1000,
      maxDelayMs: 30000,
      factor: 2
    }
  });

  // 2. Register a global error handler
  client.onError(err => {
    console.error('[AgentMQClient] Error:', err);
  });

  // 3. Connect to RabbitMQ
  try {
    await client.connect();
    console.log('Connected to broker');
  } catch (err) {
    console.error('Connection failed:', err);
    process.exit(1);
  }

  // 4. Publish a sample message
  const samplePayload = { taskId: 'abc123', action: 'processData', timestamp: Date.now() };
  try {
    await client.publish('job_queue', samplePayload, {
      persistent: true,
      headers: { origin: 'quickStart' }
    });
    console.log('Message published:', samplePayload);
  } catch (err) {
    console.error('Publish error:', err);
  }

  // 5. Consume messages
  try {
    await client.consume(
      'job_queue',
      async (msg) => {
        const data = JSON.parse(msg.content.toString('utf8'));
        console.log('Received:', data);
        // Process message...
        await client.ack(msg);
      },
      { prefetch: 5, noAck: false }
    );
    console.log('Consuming from "job_queue"...');
  } catch (err) {
    console.error('Consume error:', err);
  }

  // 6. Graceful shutdown on SIGINT
  process.on('SIGINT', async () => {
    console.log('Shutting down...');
    try {
      await client.disconnect();
      console.log('Disconnected, exiting.');
      process.exit(0);
    } catch (discErr) {
      console.error('Error during disconnect:', discErr);
      process.exit(1);
    }
  });
})();
```

---

## 📄 Configuration

Configuration can be provided to the `AgentMQClient` constructor or as overrides to `connect()`. Below is a summary of supported fields (see `docs/api.md` for full details):

| Field         | Type      | Description                                                                                                                                                        | Default                                                              |
| ------------- | --------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------ | -------------------------------------------------------------------- |
| `type`        | `string`  | Transport type: `'rabbitmq'`                                                                                     | `'rabbitmq'`                                                         |
| `host`        | `string`  | Connection URI or hostname. For RabbitMQ: e.g. `'amqp://user:pass@localhost:5672'`.                                                                                | *Required*                                                           |
| `queue`       | `string`  | Default queue name for publish/consume if not overridden per call.                                                                                                 | `''`                                                                 |
| `exchange`    | `string`  | Default exchange name. Empty string uses the default direct exchange.                                                                                              | `''`                                                                 |
| `durable`     | `boolean` | Declare queues/exchanges as durable.                                                                                                                               | `true`                                                               |
| `prefetch`    | `integer` | Default prefetch count for consumers.                                                                                                                              | `1`                                                                  |
| `noAck`       | `boolean` | Default auto-acknowledge setting for consumers. If `true`, messages will be auto-acked.                                                                            | `false`                                                              |
| `logger`      | `object`  | Custom logger with methods: `info()`, `warn()`, `error()`, `debug()`. If omitted, `console` is used.                                                               | `null`                                                               |
| `retryPolicy` | `object`  | Reconnection policy with properties:<br>‒ `retries` (number)<br>‒ `initialDelayMs` (ms)<br>‒ `maxDelayMs` (ms)<br>‒ `factor` (multiplier). Not enforced in v1.0.0. | `{ retries: 5, initialDelayMs: 1000, maxDelayMs: 30000, factor: 2 }` |

---

## 🛠️ API Reference

For full class and method documentation, including parameter descriptions, return values, and error details, see [docs/api.md](https://github.com/onlineapps/agent-mq-client/blob/main/docs/api.md).

---

## ✅ Testing

```bash
npm test                  # All tests
npm run test:unit         # Unit tests only
npm run test:component    # Component tests
npm run test:integration  # Integration tests
```

### Test Coverage Status
- **Overall Coverage**: 24.52% (improving after refactoring)
- **Passing Tests**: 75/104 (72%)
- **Test Suites**: 10/14 passing
- **Well Tested**: Config, Transports, Error handling (100%)
- **Needs Testing**: New layers (1-5%)

See [Test Report](docs/TEST_REPORT.md) for detailed coverage analysis.

---

## 🎨 Coding Standards

* **Linting**: ESLint (`eslint:recommended` + Prettier).
* **Formatting**: Prettier — check with `npm run prettier:check`, fix with `npm run prettier:fix`.
* **Testing**: Jest, aiming for ≥90% coverage.

---

## 🎯 Refactoring Benefits

### ✅ What Was Achieved
1. **Clean Layered Architecture** - Separated into specialized layers
2. **Removed Technical Debt** - Replaced MQWrapper with cleaner design
3. **Improved Extensibility** - Easy to add new patterns
4. **Better Developer Experience** - Cleaner API and documentation

### 📊 Quality Improvements
- **Separation of Concerns** - Each layer has single responsibility
- **Modular Design** - Use individual layers independently
- **Testability** - Each layer can be tested in isolation
- **Maintainability** - Easier to understand and modify
- **Backwards Compatibility** - MQWrapper alias maintained

---

## 🤝 Contributing

Contributions welcome! Please see [CONTRIBUTING.md](https://github.com/onlineapps/agent-mq-client/blob/main/CONTRIBUTING.md) for guidelines:

1. Fork the repo.
2. Create a feature branch: `git checkout -b feature/your-feature`.
3. Run tests locally and ensure linting passes.
4. Commit your changes and push to your branch.
5. Open a Pull Request against `main`.

---

## 📜 License

This project is licensed under the MIT License. See [LICENSE](https://github.com/onlineapps/agent-mq-client/blob/main/LICENSE) for details.

## 📚 Documentation

- [Complete Connectors Documentation](../../../docs/modules/connector.md)
- [Testing Standards](../../../docs/modules/connector.md#testing-standards)