# How OpenCTI Connector Architecture Works with RabbitMQ Message Queues

> Discover how OpenCTI connector architecture leverages RabbitMQ message queues for efficient platform-to-connector communication and work distribution. Learn more now.

- Repository: [OpenCTI Platform/opencti](https://github.com/opencti-platform/opencti)
- Tags: internals
- Published: 2026-02-19

---

**OpenCTI uses RabbitMQ as a message bus where each connector maintains two dedicated queues—a "listen" queue for connector-to-platform communication and a "push" queue for platform-to-connector work distribution, managed through the [`rabbitmq.js`](https://github.com/OpenCTI-Platform/opencti/blob/main/rabbitmq.js) module.**

The OpenCTI connector architecture relies on RabbitMQ to decouple data ingestion, enrichment, and processing workflows. Each connector operates as an independent process that communicates with the platform exclusively through message queues, ensuring reliable asynchronous processing. This architecture is implemented in the `OpenCTI-Platform/opencti` repository, specifically within the GraphQL backend's RabbitMQ abstraction layer.

## Core Architecture: Dual-Queue Design

Every connector in OpenCTI maintains **two dedicated queues** that isolate traffic by direction and purpose. This design prevents message collisions and enables independent scaling of producers and consumers.

The **listen queue** handles traffic from the connector to the platform:

- **Direction**: Connector → Platform
- **Exchange**: `CONNECTOR_EXCHANGE` (configured via `amqp.connector.exchange`)
- **Routing key**: `listen_routing_<connector-id>`
- **Purpose**: Publishes events such as new STIX objects, status updates, and processing confirmations

The **push queue** handles traffic from the platform to the connector:

- **Direction**: Platform → Connector
- **Exchange**: `WORKER_EXCHANGE` (configured via `amqp.worker.exchange`)
- **Routing key**: `push_routing_<connector-id>`
- **Purpose**: Distributes work items such as STIX bundles to ingest, enrichment requests, or remediation tasks

## Queue Lifecycle and Registration

Queues are provisioned automatically when the platform initializes, ensuring connectors have immediate infrastructure available.

### Automatic Queue Provisioning

The `registerConnectorQueues` function in [`opencti-platform/opencti-graphql/src/database/rabbitmq.js`](https://github.com/OpenCTI-Platform/opencti/blob/main/opencti-platform/opencti-graphql/src/database/rabbitmq.js) (lines 529-560) handles the complete setup:

1. Asserts the `CONNECTOR_EXCHANGE` and `WORKER_EXCHANGE` exchanges
2. Declares the listen and push queues for the connector
3. Binds each queue to its respective exchange using the connector-specific routing keys
4. Configures dead-letter routing for oversized bundles

### Configuration Structure

The `connectorConfig` function (lines 16-28 in the same file) bundles all RabbitMQ connection parameters into a structured object. This includes queue names, routing keys, exchange identifiers, and dead-letter routing configuration used when messages exceed size limits.

## Message Flow: Publishing and Consumption

Communication follows a strict request-response pattern mediated by the two exchanges.

### Platform-to-Connector Delivery

When the platform needs to assign work to a connector, it calls `pushToWorkerForConnector` (lines 71-74 in [`rabbitmq.js`](https://github.com/OpenCTI-Platform/opencti/blob/main/rabbitmq.js)). This function publishes to the `WORKER_EXCHANGE` using the connector's unique push routing key. The message typically contains a STIX bundle or an enrichment request.

### Connector-to-Platform Responses

Connectors report results or stream data using `pushToConnector` (lines 76-78). This publishes to the `CONNECTOR_EXCHANGE` with the listen routing key, allowing the platform to consume status updates, newly created entities, or processing confirmations.

### Message Consumption

Connectors initiate consumption by calling `consumeQueue` (lines 86-124). This function opens a channel on the listen queue, registers an asynchronous callback, and continuously processes platform-generated messages. The platform itself consumes push queue messages through internal worker managers such as `taskManager` and `syncManager`.

## Internal Queue Types

Beyond external connectors, OpenCTI utilizes the same RabbitMQ infrastructure for internal background processing.

The platform generates dedicated queues for:

- **Background tasks** – Created via `getInternalBackgroundTaskQueues` and limited by `app:task_scheduler:max_queues_breakdown` configuration. These handle asynchronous data processing jobs.
- **Playbooks** – Generated by `getInternalPlaybookQueues` with one queue per stored playbook definition, created at runtime to orchestrate automation workflows.
- **Synchronization streams** – Managed by `getInternalSyncQueues` to handle OpenCTI Streams for live data replication between instances.

All internal queues are registered using the same `registerConnectorQueues` logic, ensuring consistent durability, routing, and dead-letter handling across external and internal workers.

## Monitoring and Health Checks

OpenCTI exposes several utilities to ensure the message bus remains healthy and to optimize workload distribution.

The `rabbitMQIsAlive` function (lines 60-66 in [`rabbitmq.js`](https://github.com/OpenCTI-Platform/opencti/blob/main/rabbitmq.js)) performs periodic connectivity checks to verify the broker is reachable. For operational visibility, `getConnectorQueueSize` retrieves the current message count for a specific connector's push queue, enabling back-pressure detection.

The platform uses `getBestBackgroundConnectorId` to implement intelligent load balancing. By comparing queue depths across available background task connectors, the system routes new jobs to the worker with the lowest current backlog, preventing individual queue saturation.

## Code Examples

The following snippets demonstrate the essential interactions for connector developers and platform integrators.

### Building Connector Configuration

```javascript
import { connectorConfig } from './database/rabbitmq';

const config = connectorConfig('my-connector-id');
console.log(config.listenQueue); // listen_routing_my-connector-id

```

### Registering Queues Programmatically

```javascript
await registerConnectorQueues(
  'my-connector-id',
  'My Connector',
  'external',               // type: external, internal, etc.
  ['observable', 'artifact'] // scopes this connector handles
);

```

### Pushing Work to a Connector

```javascript
import { pushToWorkerForConnector } from './database/rabbitmq';

await pushToWorkerForConnector('my-connector-id', { 
  action: 'ingest', 
  data: stixBundle 
});

```

### Consuming Messages in a Connector

```javascript
import { consumeQueue } from './database/rabbitmq';

await consumeQueue(context, 'my-connector-id', setConnection, async (ctx, msg) => {
  const payload = JSON.parse(msg);
  // Process platform-assigned work
  await processStixBundle(payload);
});

```

### Responding to the Platform

```javascript
import { pushToConnector } from './database/rabbitmq';

await pushToConnector('my-connector-id', { 
  status: 'processed', 
  entityId: createdEntity.id 
});

```

## Key Source Files

Understanding the connector architecture requires familiarity with these specific files in the `OpenCTI-Platform/opencti` repository:

- **[`opencti-platform/opencti-graphql/src/database/rabbitmq.js`](https://github.com/OpenCTI-Platform/opencti/blob/main/opencti-platform/opencti-graphql/src/database/rabbitmq.js)** – Core RabbitMQ abstraction containing `registerConnectorQueues`, `connectorConfig`, `pushToWorkerForConnector`, `pushToConnector`, `consumeQueue`, and health check utilities.

- **[`opencti-platform/opencti-graphql/src/connector/connector-domain.ts`](https://github.com/OpenCTI-Platform/opencti/blob/main/opencti-platform/opencti-graphql/src/connector/connector-domain.ts)** – High-level domain logic exposing connector registration and queue discovery to other platform modules.

- **[`opencti-platform/opencti-graphql/src/resolvers/connector.js`](https://github.com/OpenCTI-Platform/opencti/blob/main/opencti-platform/opencti-graphql/src/resolvers/connector.js)** – GraphQL resolver layer providing connector-related queries and mutations to the frontend API.

- **[`opencti-platform/opencti-graphql/src/modules/playbook/playbook-domain.ts`](https://github.com/OpenCTI-Platform/opencti/blob/main/opencti-platform/opencti-graphql/src/modules/playbook/playbook-domain.ts)** – Demonstrates internal queue usage for automation playbooks, showing how `registerConnectorQueues` applies to internal components.

- **[`opencti-platform/opencti-graphql/src/modules/pir/pir-domain.ts`](https://github.com/OpenCTI-Platform/opencti/blob/main/opencti-platform/opencti-graphql/src/modules/pir/pir-domain.ts)** – Concrete implementation showing the Platform pushing jobs to a dedicated connector queue for PIR (Potential Incident Response) processing.

## Summary

- OpenCTI's connector architecture relies on **RabbitMQ** as the central message bus, ensuring reliable asynchronous communication between the platform and external connectors.
- Each connector operates two dedicated queues: a **listen queue** (connector → platform) using `CONNECTOR_EXCHANGE` and a **push queue** (platform → connector) using `WORKER_EXCHANGE`.
- Queue lifecycle management is handled by `registerConnectorQueues` in [`src/database/rabbitmq.js`](https://github.com/OpenCTI-Platform/opencti/blob/main/src/database/rabbitmq.js), which provisions exchanges, declares queues, and binds routing keys automatically.
- Message flow uses `pushToWorkerForConnector` for platform-initiated work and `pushToConnector` for connector responses, with `consumeQueue` handling continuous message consumption.
- Internal components like background tasks, playbooks, and sync streams reuse the same queue infrastructure, ensuring architectural consistency across the platform.

## Frequently Asked Questions

### How does OpenCTI ensure message delivery reliability between connectors and the platform?

OpenCTI implements **dedicated exchanges and routing keys** for each connector, with the `registerConnectorQueues` function configuring dead-letter routing for oversized bundles. The platform uses persistent message delivery and confirms queue declaration before accepting connector registrations, ensuring that messages are not lost during transmission.

### What is the difference between the CONNECTOR_EXCHANGE and WORKER_EXCHANGE in OpenCTI?

The **CONNECTOR_EXCHANGE** (configured via `amqp.connector.exchange`) handles messages flowing from connectors to the platform via listen queues, while the **WORKER_EXCHANGE** (configured via `amqp.worker.exchange`) distributes work items from the platform to connectors via push queues. This separation prevents routing conflicts and enables independent scaling of ingestion and processing workloads.

### Can internal OpenCTI components use the same queue infrastructure as external connectors?

Yes, internal components such as **background tasks**, **playbooks**, and **synchronization streams** utilize the same `registerConnectorQueues` logic found in [`src/database/rabbitmq.js`](https://github.com/OpenCTI-Platform/opencti/blob/main/src/database/rabbitmq.js). These internal queues are generated by functions like `getInternalBackgroundTaskQueues` and `getInternalPlaybookQueues`, ensuring consistent durability, routing, and monitoring across both external connectors and platform-native workers.

### How does OpenCTI handle load balancing across multiple connector instances?

The platform uses the `getBestBackgroundConnectorId` function to monitor queue depths via `getConnectorQueueSize` and routes new jobs to the connector with the lowest current backlog. This **intelligent load balancing** prevents individual queue saturation and ensures optimal distribution of processing workloads across the connector pool.