@onlineapps/mq-client-core

Core MQ client library for RabbitMQ - shared by infrastructure services and connectors.

Overview

This is the core library extracted from @onlineapps/conn-infra-mq to provide basic MQ functionality for infrastructure services (gateway, registry, validator) without business-specific features.

Architecture Principle: Connectors are exclusively for business services. If a connector contains functionality that should also serve infrastructure services, it must be extracted into a shared library.

Installation

npm install @onlineapps/mq-client-core

Usage

For Infrastructure Services

const BaseClient = require('@onlineapps/mq-client-core');

const mqClient = new BaseClient({
  type: 'rabbitmq',
  host: 'amqp://localhost:5672',
  queue: 'workflow.init' // Optional default queue
});

await mqClient.connect();
await mqClient.publish('workflow.init', { workflowId: '123', data: '...' });
await mqClient.publish('workflow.control', { workflowId: '123', data: '...' });

Configuration

Flexible Schema - Only type and host are required:

{
  type: 'rabbitmq',        // Required
  host: 'amqp://...',      // Required
  queue: 'optional',        // Optional default queue
  exchange: '',             // Optional exchange
  durable: true,            // Optional (default: true)
  prefetch: 1,              // Optional (default: 1)
  noAck: false,             // Optional (default: false)
  logger: null              // Optional custom logger
}

API

BaseClient

• connect(options?) - Connect to RabbitMQ
• disconnect() - Disconnect from RabbitMQ
• publish(queue, message, options?) - Publish message to queue
• consume(queue, handler, options?) - Consume messages from queue
• ack(msg) - Acknowledge message
• nack(msg, options?) - Negative acknowledge message
• isConnected() - Check connection status
• onError(callback) - Register error handler

publish() and the workflow helpers built on it resolve to undefined; failure is signalled by a thrown PublishError, never by a falsy return.

Queue ownership — publishing never creates an owned queue

A missing queue is answered by who owns the name, not by the client's scope:

| Name | Owner | Missing at publish time | |---|---|---| | workflow.*, registry.*, infrastructure.*, validation.*, monitoring.*, telemetry.*, delivery.* | the infrastructure service that declares it | QueueNotFoundError (kind: 'infrastructure') | | {service}.workflow, {service}.queue, {service}.dlq | the owning service, via setupServiceQueues() after registration | QueueNotFoundError (kind: 'business') | | anything else | nobody | created with default options, if recoveryScope allows it |

Publishing must never create an owned queue: sendToQueue()/assertQueue() would declare it with default arguments — no TTL, no DLQ — and the owner's later setupServiceQueues() then fails with 406 PRECONDITION_FAILED because the arguments no longer match. The rule is enforced in two places, and both are required: the 404 branch of _publishOnce() refuses the publish, and RecoveryWorker.handleQueueNotFound() refuses to create the queue behind it.

recoveryScope: 'business' (the default every ConnectorMQClient sets) says the client belongs to a business service. It has never meant "may create business queues".

Architecture

mq-client-core (this library)
├── BaseClient - Core AMQP operations
├── RabbitMQClient - Transport implementation
└── Basic publish/consume functionality

conn-infra-mq (connector for business services)
├── Uses mq-client-core internally
├── ConnectorMQClient - Orchestrator with layers
├── WorkflowRouter - Business workflow routing
└── Additional business-specific features

Infrastructure services (gateway, registry, validator)
└── Use mq-client-core directly (not the connector)

Related Packages

• @onlineapps/conn-infra-mq - Full connector with business-specific features (uses this library internally)

License

MIT