### Install @moleculer/channels

Source: https://github.com/moleculerjs/moleculer-channels/blob/master/README.md

Install the @moleculer/channels package using npm.

```bash
npm i @moleculer/channels
```

--------------------------------

### Configure AMQP Adapter

Source: https://github.com/moleculerjs/moleculer-channels/blob/master/README.md

Configure the AMQP adapter for Moleculer Channels using a simple connection string. Ensure the 'amqplib' module is installed.

```javascript
// moleculer.config.js
const ChannelsMiddleware = require("@moleculer/channels").Middleware;

module.exports = {
    middlewares: [
        ChannelsMiddleware({
            adapter: "amqp://localhost:5672"
        })
    ]
};
```

--------------------------------

### Configure Moleculer Channels with Redis Adapter

Source: https://context7.com/moleculerjs/moleculer-channels/llms.txt

Set up Moleculer Channels middleware with the Redis adapter. This example shows standard Redis connection options and dead-lettering configuration. Ensure Redis is running and accessible.

```javascript
const { ServiceBroker } = require("moleculer");
const ChannelsMiddleware = require("@moleculer/channels").Middleware;

const broker = new ServiceBroker({
    middlewares: [
        ChannelsMiddleware({
            adapter: {
                type: "Redis",
                options: {
                    // Standard Redis connection
                    redis: {
                        host: "127.0.0.1",
                        port: 6379,
                        db: 0,
                        password: "secret",
                        // Consumer behavior options
                        consumerOptions: {
                            readTimeoutInterval: 0,      // Never timeout
                            minIdleTime: 3600000,        // 1 hour
                            claimInterval: 100,          // 100ms
                            startID: "$",                // New messages only
                            processingAttemptsInterval: 1000
                        }
                    },

                    // OR Redis Cluster configuration
                    // redis: {
                    //     cluster: {
                    //         nodes: [
                    //             { host: "127.0.0.1", port: 6380 },
                    //             { host: "127.0.0.1", port: 6381 },
                    //             { host: "127.0.0.1", port: 6382 }
                    //         ],
                    //         options: {
                    //             redisOptions: { password: "cluster-pass" }
                    //         }
                    //     }
                    // },

                    serializer: "JSON",  // or "MsgPack"
                    maxRetries: 3,
                    maxInFlight: 1,
                    deadLettering: {
                        enabled: true,
                        queueName: "FAILED_MESSAGES",
                        errorInfoTTL: 86400  // 24 hours
                    }
                }
            }
        })
    ]
});

broker.createService({
    name: "redis-consumer",

    channels: {
        "stream.events": {
            group: "event-processors",
            maxInFlight: 5,
            maxRetries: 3,

            // Redis-specific channel options
            redis: {
                startID: "0",              // Read from beginning
                minIdleTime: 5000,         // 5 second idle time
                claimInterval: 500,        // Claim every 500ms
                readTimeoutInterval: 1000  // 1 second read timeout
            },

            async handler(payload, raw) {
                // raw.id = Redis stream message ID (e.g., "1234567890123-0")
                // raw.key = message key if provided
                // raw.headers = message headers
                this.logger.info(`Message ${raw.id}:`, payload);
            }
        }
    },

    actions: {
        async publish(ctx) {
            await broker.sendToChannel("stream.events", ctx.params, {
                // Limit stream to ~1000 entries
                xaddMaxLen: "~1000",
                headers: { source: "api" }
            });
        }
    }
});

broker.start();

```

--------------------------------

### Configure Kafka Adapter

Source: https://github.com/moleculerjs/moleculer-channels/blob/master/README.md

Configure the Kafka adapter for Moleculer Channels using a simple broker connection string. Ensure the 'kafkajs' module is installed.

```javascript
// moleculer.config.js
const ChannelsMiddleware = require("@moleculer/channels").Middleware;

module.exports = {
    middlewares: [
        ChannelsMiddleware({
            adapter: "kafka://localhost:9092"
        })
    ]
};
```

--------------------------------

### Configure NATS JetStream Adapter

Source: https://github.com/moleculerjs/moleculer-channels/blob/master/README.md

Configure the Channels middleware to use the NATS JetStream adapter. Ensure the 'nats' module is installed.

```javascript
// moleculer.config.js
const ChannelsMiddleware = require("@moleculer/channels").Middleware;

module.exports = {
    middlewares: [
        ChannelsMiddleware({
            adapter: "nats://localhost:4222"
        })
    ]
};

```

--------------------------------

### Configure Redis Adapter in Moleculer

Source: https://github.com/moleculerjs/moleculer-channels/blob/master/README.md

Basic configuration for the Channels middleware using the Redis adapter. Ensure you have the 'ioredis' module installed.

```javascript
// moleculer.config.js
const ChannelsMiddleware = require("@moleculer/channels").Middleware;

module.exports = {
    middlewares: [
        ChannelsMiddleware({
            adapter: "redis://localhost:6379"
        })
    ]
};
```

--------------------------------

### Configuring ChannelsMiddleware with Redis Adapter

Source: https://github.com/moleculerjs/moleculer-channels/blob/master/README.md

Example of configuring the ChannelsMiddleware in `moleculer.config.js` to use a Redis adapter with a connection string. This sets up custom method names and property names for accessing the adapter and sending messages.

```javascript
// moleculer.config.js
const ChannelsMiddleware = require("@moleculer/channels").Middleware;

module.exports = {
    logger: true,

    middlewares: [
        ChannelsMiddleware({
            adapter: "redis://localhost:6379",
            sendMethodName: "sendToChannel",
            adapterPropertyName: "channelAdapter",
            schemaProperty: "channels"
        })
    ]
};
```

--------------------------------

### Define Channel Handler with Full Configuration

Source: https://context7.com/moleculerjs/moleculer-channels/llms.txt

Use this configuration for advanced channel handler setups, including consumer groups, retry policies, dead-lettering, and adapter-specific parameters for Redis, AMQP, Kafka, and NATS JetStream.

```javascript
broker.createService({
    name: "payments",

    channels: {
        "payment.process": {
            // Consumer group name (default: service full name)
            group: "payment-processors",

            // Max messages processed simultaneously (default: 1)
            maxInFlight: 10,

            // Max retry attempts before dead-lettering (default: 3)
            maxRetries: 5,

            // Dead-letter queue configuration
            deadLettering: {
                enabled: true,
                queueName: "FAILED_PAYMENTS",
                // AMQP-specific options
                exchangeName: "FAILED_PAYMENTS",
                exchangeOptions: { durable: true },
                queueOptions: { durable: true }
            },

            // Redis-specific options
            redis: {
                minIdleTime: 60000,     // 1 minute idle before claiming
                claimInterval: 1000,     // Check for claimable messages every 1s
                startID: "$",            // Only new messages
                readTimeoutInterval: 0   // No timeout
            },

            // AMQP-specific options
            amqp: {
                queueOptions: { durable: true },
                consumerOptions: { noAck: false }
            },

            // Kafka-specific options
            kafka: {
                fromBeginning: false,
                partitionsConsumedConcurrently: 3
            },

            // NATS JetStream-specific options
            nats: {
                consumerOptions: {
                    config: {
                        deliver_policy: "new",
                        ack_policy: "explicit"
                    }
                }
            },

            // Handler function
            async handler(payload, raw) {
                this.logger.info("Processing payment:", payload.transactionId);

                try {
                    await this.chargeCard(payload);
                    this.logger.info("Payment successful");
                } catch (error) {
                    this.logger.error("Payment failed:", error.message);
                    // Throwing an error will NACK the message and trigger retry
                    throw error;
                }
            }
        }
    }
});
```

--------------------------------

### Send Message to AMQP Channel

Source: https://context7.com/moleculerjs/moleculer-channels/llms.txt

Example of sending a message to the 'tasks.process' channel with custom message options including persistence, TTL, priority, correlation ID, and headers.

```javascript
await broker.sendToChannel("tasks.process", ctx.params, {
    persistent: true,
    ttl: 3600000,  // 1 hour
    priority: ctx.params.priority || 0,
    correlationId: ctx.requestID,
    headers: {
        "x-submitted-by": ctx.meta.user?.id
    }
});
```

--------------------------------

### Overwrite Channel Options in Service

Source: https://github.com/moleculerjs/moleculer-channels/blob/master/README.md

Example of overriding default adapter options like 'minIdleTime' and 'claimInterval' directly within a service's channel definition. Also demonstrates enabling dead-lettering.

```javascript
module.exports = {
    name: "payments",

    actions: {
        /*...*/
    },

    channels: {
        "order.created": {
            maxInFlight: 6,
            async handler(payload) {
                /*...*/
            }
        },
        "payment.processed": {
            redis: {
                minIdleTime: 10,
                claimInterval: 10
            }
            deadLettering: {
                enabled: true,
                queueName: "DEAD_LETTER"
            },
            async handler(payload) {
                /*...*/
            }
        }
    }
};
```

--------------------------------

### Publish Message to Kafka Channel

Source: https://context7.com/moleculerjs/moleculer-channels/llms.txt

Publish a message to the 'events.analytics' channel using `broker.sendToChannel`. This example demonstrates setting a partition key, explicit partition, acknowledgements, timeout, and custom headers.

```javascript
await broker.sendToChannel("events.analytics", ctx.params, {
    // Partition key for ordering
    key: ctx.params.userId,
    // Explicit partition
    partition: 0,
    // Required acknowledgements
    acks: -1,  // All replicas
    // Timeout
    timeout: 30000,
    // Headers
    headers: {
        "event-type": ctx.params.type,
        "source": "web-app"
    }
});
```

--------------------------------

### Define AMQP Consumer Service

Source: https://context7.com/moleculerjs/moleculer-channels/llms.txt

Create a Moleculer service that consumes messages from an AMQP channel. This example defines a 'tasks.process' channel with specific AMQP options for queues, exchanges, and consumers, and includes a handler for processing received tasks.

```javascript
broker.createService({
    name: "amqp-consumer",

    channels: {
        "tasks.process": {
            group: "task-workers",
            maxInFlight: 20,

            // AMQP-specific options
            amqp: {
                queueOptions: {
                    durable: true,
                    arguments: {
                        "x-message-ttl": 86400000  // 24 hours
                    }
                },
                exchangeOptions: {
                    type: "direct",
                    durable: true
                },
                consumerOptions: {
                    prefetch: 10
                }
            },

            async handler(payload, raw) {
                // raw.fields = AMQP delivery info
                // raw.properties = message properties
                this.logger.info("Task received:", payload);
            }
        }
    },

    actions: {
        async queueTask(ctx) {
            await broker.sendToChannel("tasks.process", ctx.params, {
                persistent: true,
                ttl: 3600000,  // 1 hour
                priority: ctx.params.priority || 0,
                correlationId: ctx.requestID,
                headers: {
                    "x-submitted-by": ctx.meta.user?.id
                }
            });
        },

        // Send directly to a queue (bypass exchange)
        async sendDirect(ctx) {
            await broker.sendToChannel("", ctx.params, {
                routingKey: "my-direct-queue"
            });
        }
    }
});
```

--------------------------------

### Configure Moleculer Broker with Channels and Tracing

Source: https://context7.com/moleculerjs/moleculer-channels/llms.txt

Set up the Moleculer broker with Channels middleware (enabling context globally) and Tracing middleware. Ensure Redis is available for the Channels adapter.

```javascript
const { ServiceBroker } = require("moleculer");
const ChannelsMiddleware = require("@moleculer/channels").Middleware;
const TracingMiddleware = require("@moleculer/channels").Tracing;

const broker = new ServiceBroker({
    tracing: {
        enabled: true,
        exporter: [{ type: "Console" }, { type: "Jaeger" }]
    },
    middlewares: [
        ChannelsMiddleware({
            adapter: "redis://localhost:6379",
            // Enable context globally for all handlers
            context: true
        }),
        // Add tracing middleware for channel spans
        TracingMiddleware()
    ]
});

// Publisher service - passes context when sending
broker.createService({
    name: "order-service",

    actions: {
        async submitOrder(ctx) {
            const order = ctx.params;

            await broker.sendToChannel("order.submitted", order, {
                // Pass current context to propagate meta and tracing
                ctx: ctx,
                headers: { "x-order-type": "standard" }
            });

            return { submitted: true };
        }
    }
});

// Consumer service - receives context
broker.createService({
    name: "fulfillment",

    channels: {
        "order.submitted": {
            // Enable context for this handler (or set globally)
            context: true,

            // Tracing configuration
            tracing: {
                spanName: ctx => `Process Order ${ctx.params.id}`,
                tags: {
                    params: true,  // Include params in span
                    meta: true     // Include meta in span
                }
            },

            async handler(ctx, raw) {
                // ctx is a full Moleculer Context
                this.logger.info("Order params:", ctx.params);
                this.logger.info("Request ID:", ctx.requestID);
                this.logger.info("Channel name:", ctx.channelName);
                this.logger.info("Parent channel:", ctx.parentChannelName);

                // Access meta from publisher
                if (ctx.meta.user) {
                    this.logger.info("Submitted by:", ctx.meta.user.name);
                }

                // Call other services with tracing propagated
                const inventory = await ctx.call("inventory.reserve", {
                    items: ctx.params.items
                });

                await ctx.call("shipping.schedule", {
                    orderId: ctx.params.id,
                    address: ctx.params.shippingAddress
                });

                this.logger.info("Order fulfilled");
            }
        }
    }
});

broker.start();

```

--------------------------------

### Configure Multiple Channel Adapters

Source: https://context7.com/moleculerjs/moleculer-channels/llms.txt

Register Redis, Kafka, and AMQP channel middlewares with specific configurations for simultaneous use. Each adapter is configured with its connection details and schema properties.

```javascript
const { ServiceBroker } = require("moleculer");
const ChannelsMiddleware = require("@moleculer/channels").Middleware;

const broker = new ServiceBroker({
    middlewares: [
        // Redis adapter for real-time events
        ChannelsMiddleware({
            adapter: "redis://localhost:6379",
            schemaProperty: "redisChannels",
            sendMethodName: "sendToRedis",
            adapterPropertyName: "redisAdapter"
        }),

        // Kafka adapter for high-throughput data pipelines
        ChannelsMiddleware({
            adapter: {
                type: "Kafka",
                options: {
                    kafka: {
                        brokers: ["kafka-1:9092", "kafka-2:9092"]
                    }
                }
            },
            schemaProperty: "kafkaChannels",
            sendMethodName: "sendToKafka",
            adapterPropertyName: "kafkaAdapter"
        }),

        // AMQP adapter for reliable task queues
        ChannelsMiddleware({
            adapter: "amqp://localhost:5672",
            schemaProperty: "amqpChannels",
            sendMethodName: "sendToAMQP",
            adapterPropertyName: "amqpAdapter"
        })
    ]
});

broker.createService({
    name: "multi-channel-service",

    // Redis channels for real-time notifications
    redisChannels: {
        "user.online": {
            async handler(payload) {
                this.logger.info("User online (Redis):", payload.userId);
            }
        }
    },

    // Kafka channels for analytics events
    kafkaChannels: {
        "analytics.pageview": {
            group: "analytics-processors",
            kafka: {
                partitionsConsumedConcurrently: 5
            },
            async handler(payload) {
                this.logger.info("Page view (Kafka):", payload);
            }
        }
    },

    // AMQP channels for task processing
    amqpChannels: {
        "tasks.email": {
            group: "email-workers",
            maxInFlight: 10,
            async handler(payload) {
                this.logger.info("Email task (AMQP):", payload.recipient);
                await this.sendEmail(payload);
            }
        }
    },

    actions: {
        async notifyUserOnline(ctx) {
            await broker.sendToRedis("user.online", { userId: ctx.params.userId });
        },

        async trackPageView(ctx) {
            await broker.sendToKafka("analytics.pageview", ctx.params);
        },

        async queueEmail(ctx) {
            await broker.sendToAMQP("tasks.email", ctx.params);
        }
    }
});

broker.start();

```

--------------------------------

### Configure Kafka Adapter with Options

Source: https://github.com/moleculerjs/moleculer-channels/blob/master/README.md

Configure the Kafka adapter with detailed options, including multiple brokers, producer/consumer options, retry settings, and dead-lettering. This provides advanced control over Kafka integration.

```javascript
// moleculer.config.js
const ChannelsMiddleware = require("@moleculer/channels").Middleware;

module.exports = {
    middlewares: [
        ChannelsMiddleware({
            adapter: {
                type: "Kafka",
                options: {
                    kafka: {
                        brokers: ["kafka-1:9092", "kafka-1:9092"],
                        // Options for `producer()`
                        producerOptions: {},
                        // Options for `consumer()`
                        consumerOptions: {}
                    },
                    maxRetries: 3,
                    deadLettering: {
                        enabled: false,
                        queueName: "DEAD_LETTER"
                    }
                }
            }
        })
    ]
};
```

--------------------------------

### Configure AMQP Adapter with Options

Source: https://github.com/moleculerjs/moleculer-channels/blob/master/README.md

Configure the AMQP adapter with detailed options, including connection URLs, socket options, queue/exchange assertions, message properties, and dead-lettering. This allows fine-grained control over RabbitMQ interactions.

```javascript
// moleculer.config.js
const ChannelsMiddleware = require("@moleculer/channels").Middleware;

module.exports = {
    middlewares: [
        ChannelsMiddleware({
            adapter: {
                type: "AMQP",
                options: {
                    amqp: {
                        url: "amqp://localhost:5672",
                        // Options for `Amqplib.connect`
                        socketOptions: {},
                        // Options for `assertQueue()`
                        queueOptions: {},
                        // Options for `assertExchange()`
                        exchangeOptions: {},
                        // Options for `channel.publish()`
                        messageOptions: {},
                        // Options for `channel.consume()`
                        consumerOptions: {},
                        // Note: options for `channel.assertExchange()` before first publishing in new exchange
                        publishAssertExchange: {
                            // Enable/disable calling once `channel.assertExchange()` before first publishing in new exchange by `sendToChannel`
                            enabled: false,
                            // Options for `channel.assertExchange()` before publishing
                            exchangeOptions: {}
                        }
                    },
                    maxInFlight: 10,
                    maxRetries: 3,
                    deadLettering: {
                        enabled: false
                        //queueName: "DEAD_LETTER",
                        //exchangeName: "DEAD_LETTER"
                    }
                }
            }
        })
    ]
};
```

--------------------------------

### Configure Kafka Adapter for Moleculer Channels

Source: https://context7.com/moleculerjs/moleculer-channels/llms.txt

Set up the Kafka adapter for Moleculer Channels, including Kafka broker addresses, custom logging, and producer/consumer options. Dead-lettering is also configured.

```javascript
const { ServiceBroker } = require("moleculer");
const ChannelsMiddleware = require("@moleculer/channels").Middleware;

const broker = new ServiceBroker({
    middlewares: [
        ChannelsMiddleware({
            adapter: {
                type: "Kafka",
                options: {
                    kafka: {
                        brokers: ["kafka-1:9092", "kafka-2:9092", "kafka-3:9092"],
                        // Custom logger
                        logCreator: () => ({ namespace, level, log }) => {
                            console.log(`[Kafka:${namespace}] ${level}: ${log.message}`);
                        },
                        // Producer options
                        producerOptions: {
                            allowAutoTopicCreation: true,
                            transactionTimeout: 30000
                        },
                        // Consumer options
                        consumerOptions: {
                            sessionTimeout: 30000,
                            heartbeatInterval: 3000
                        }
                    },
                    maxRetries: 3,
                    deadLettering: {
                        enabled: true,
                        queueName: "DEAD_LETTER_TOPIC"
                    }
                }
            }
        })
    ]
});
```

--------------------------------

### Configure Redis Adapter with Options

Source: https://github.com/moleculerjs/moleculer-channels/blob/master/README.md

Advanced configuration for the Redis adapter, specifying connection details and consumer options for the ioredis client. This allows fine-grained control over message consumption.

```javascript
// moleculer.config.js
const ChannelsMiddleware = require("@moleculer/channels").Middleware;

module.exports = {
    middlewares: [
        ChannelsMiddleware({
            adapter: {
                type: "Redis",
                options: {
                    redis: {
                        // ioredis constructor options: https://github.com/luin/ioredis#connect-to-redis
                        host: "127.0.0.1",
                        port: 6379,
                        db: 3,
                        password: "pass1234",
                        consumerOptions: {
                            // Timeout interval (in milliseconds) while waiting for new messages. By default never timeout
                            readTimeoutInterval: 0,
                            // Time (in milliseconds) after which pending messages are considered NACKed and should be claimed. Defaults to 1 hour.
                            minIdleTime: 5000,
                            // Interval (in milliseconds) between two claims
                            claimInterval: 100,
                            // "$" is a special ID. Consumers fetching data from the consumer group will only see new elements arriving in the stream.
                            // More info: https://redis.io/commands/XGROUP
                            startID: "$",
                            // Interval (in milliseconds) between message transfer into FAILED_MESSAGES channel
                            processingAttemptsInterval: 1000
                        }
                    }
                }
            }
        })
    ]
};
```

--------------------------------

### Create Custom Channel Middleware

Source: https://context7.com/moleculerjs/moleculer-channels/llms.txt

Implement custom middleware to wrap channel handlers and the sendToChannel method for logging, validation, or transformation. Requires importing ChannelsMiddleware.

```javascript
const { ServiceBroker } = require("moleculer");
const ChannelsMiddleware = require("@moleculer/channels").Middleware;

const AuditMiddleware = {
    name: "ChannelAudit",

    // Wrap channel handlers
    localChannel(next, chan) {
        return async (payload, raw) => {
            const startTime = Date.now();
            this.logger.info(`[AUDIT] Receiving message on '${chan.name}'`);

            try {
                const result = await next(payload, raw);
                const duration = Date.now() - startTime;
                this.logger.info(`[AUDIT] Processed '${chan.name}' in ${duration}ms`);
                return result;
            } catch (error) {
                this.logger.error(`[AUDIT] Error in '${chan.name}':`, error.message);
                throw error;
            }
        };
    },

    // Wrap sendToChannel method
    sendToChannel(next) {
        return async (channelName, payload, opts) => {
            this.logger.info(`[AUDIT] Publishing to '${channelName}'`);

            // Add audit metadata
            const enrichedOpts = {
                ...opts,
                headers: {
                    ...opts?.headers,
                    "x-audit-timestamp": Date.now().toString(),
                    "x-audit-node": this.broker.nodeID
                }
            };

            await next(channelName, payload, enrichedOpts);
            this.logger.info(`[AUDIT] Published to '${channelName}'`);
        };
    }
};

const broker = new ServiceBroker({
    middlewares: [
        AuditMiddleware,
        ChannelsMiddleware({
            adapter: "redis://localhost:6379"
        })
    ]
});

broker.start();
```

--------------------------------

### Configure Moleculer Broker with AMQP Channels Middleware

Source: https://context7.com/moleculerjs/moleculer-channels/llms.txt

Set up the Moleculer broker with the Channels middleware, specifying the AMQP adapter type and its options. This includes connection details, queue, exchange, message, and consumer configurations, as well as dead-lettering settings.

```javascript
const { ServiceBroker } = require("moleculer");
const ChannelsMiddleware = require("@moleculer/channels").Middleware;

const broker = new ServiceBroker({
    middlewares: [
        ChannelsMiddleware({
            adapter: {
                type: "AMQP",
                options: {
                    amqp: {
                        url: "amqp://user:pass@localhost:5672/vhost",
                        socketOptions: {
                            heartbeat: 30
                        },
                        // Default queue options
                        queueOptions: {
                            durable: true,
                            autoDelete: false
                        },
                        // Default exchange options
                        exchangeOptions: {
                            type: "fanout",
                            durable: true
                        },
                        // Default message options
                        messageOptions: {
                            persistent: true
                        },
                        // Default consumer options
                        consumerOptions: {
                            noAck: false
                        },
                        // Assert exchange before publishing
                        publishAssertExchange: {
                            enabled: true,
                            exchangeOptions: { durable: true }
                        }
                    },
                    maxRetries: 3,
                    maxInFlight: 10,
                    deadLettering: {
                        enabled: true,
                        queueName: "DEAD_LETTER",
                        exchangeName: "DEAD_LETTER",
                        exchangeOptions: { durable: true },
                        queueOptions: { durable: true }
                    }
                }
            }
        })
    ]
});
```

--------------------------------

### AMQP Configuration

Source: https://github.com/moleculerjs/moleculer-channels/blob/master/README.md

Configuration options for the AMQP adapter.

```APIDOC
## AMQP Configuration

### Description
Configuration options for the AMQP adapter.

### Parameters
#### Request Body
- **amqp.url** (String) - Optional - Connection URI.
- **amqp.socketOptions** (Object) - Optional - AMQP lib socket configuration. More info [here](http://www.squaremobius.net/amqp.node/channel_api.html#connect).
- **amqp.queueOptions** (Object) - Optional - AMQP lib queue configuration. More info [here](http://www.squaremobius.net/amqp.node/channel_api.html#channel_assertQueue).
- **amqp.exchangeOptions** (Object) - Optional - AMQP lib exchange configuration. More info [here](http://www.squaremobius.net/amqp.node/channel_api.html#channel_assertExchange).
- **amqp.messageOptions** (Object) - Optional - AMQP lib message configuration. More info [here](http://www.squaremobius.net/amqp.node/channel_api.html#channel_publish).
- **amqp.consumerOptions** (Object) - Optional - AMQP lib consume configuration. More info [here](http://www.squaremobius.net/amqp.node/channel_api.html#channel_consume).
- **amqp.publishAssertExchange.enabled** (Boolean) - Optional - Enable/disable calling once `channel.assertExchange()` before first publishing in new exchange by `sendToChannel()`. More info [here](http://www.squaremobius.net/amqp.node/channel_api.html#channel_assertExchange). Default: `false`.
- **amqp.publishAssertExchange.exchangeOptions** (Object) - Optional - AMQP lib exchange configuration. More info [here](http://www.squaremobius.net/amqp.node/channel_api.html#channel_assertExchange).
```

--------------------------------

### Configure Channels Middleware with Dead-Lettering

Source: https://context7.com/moleculerjs/moleculer-channels/llms.txt

Set up the ChannelsMiddleware with Redis adapter to enable dead-lettering. Configure the queue name, maximum retries, and optionally a custom error transformation function to store error details in message headers.

```javascript
const { ServiceBroker, Errors } = require("moleculer");
const ChannelsMiddleware = require("@moleculer/channels").Middleware;

const broker = new ServiceBroker({
    middlewares: [
        ChannelsMiddleware({
            adapter: {
                type: "Redis",
                options: {
                    redis: "localhost:6379",
                    maxRetries: 3,
                    deadLettering: {
                        enabled: true,
                        queueName: "FAILED_MESSAGES",
                        // TTL for error info in Redis (seconds)
                        errorInfoTTL: 86400,  // 24 hours
                        // Custom error transformation
                        transformErrorToHeaders: (err) => ({
                            "x-error-message": err.message,
                            "x-error-code": err.code?.toString() || "UNKNOWN",
                            "x-error-timestamp": Date.now().toString()
                        })
                    }
                }
            }
        })
    ]
});

// Service with failing handler
broker.createService({
    name: "processor",

    channels: {
        "task.process": {
            maxRetries: 3,
            deadLettering: {
                enabled: true,
                queueName: "DEAD_LETTER"
            },

            async handler(payload) {
                // Simulate processing failure
                if (!payload.valid) {
                    throw new Errors.MoleculerError(
                        "Invalid task data",
                        400,
                        "VALIDATION_ERROR",
                        { field: "data", reason: "missing required fields" }
                    );
                }
                await this.processTask(payload);
            }
        }
    }
});

// Dead-letter queue consumer
broker.createService({
    name: "error-handler",

    channels: {
        DEAD_LETTER: {
            context: true,
            group: "error-processors",

            async handler(ctx, raw) {
                // Original message payload
                this.logger.error("Failed message:", ctx.params);

                // Error info from headers
                this.logger.error("Error headers:", ctx.headers);
                // ctx.headers["x-error-message"] = "Invalid task data"
                // ctx.headers["x-error-code"] = "400"
                // ctx.headers["x-error-stack"] = "..." (base64 encoded)
                // ctx.headers["x-error-data"] = "..." (base64 encoded)
                // ctx.headers["x-original-channel"] = "task.process"
                // ctx.headers["x-original-group"] = "processor"

                // Alert operations team
                await ctx.call("notifications.alert", {
                    type: "dead-letter",
                    channel: ctx.headers["x-original-channel"],
                    error: ctx.headers["x-error-message"],
                    payload: ctx.params
                });

                // Store for later analysis
                await ctx.call("audit.logFailure", {
                    timestamp: new Date(),
                    error: ctx.headers,
                    message: ctx.params
                });
            }
        }
    }
});

broker.start();

```

--------------------------------

### Configure NATS JetStream Adapter with Options

Source: https://github.com/moleculerjs/moleculer-channels/blob/master/README.md

Configure the Channels middleware with detailed NATS JetStream adapter options, including connection, stream, and consumer configurations. This allows for fine-grained control over message handling and retries.

```javascript
// moleculer.config.js
const ChannelsMiddleware = require("@moleculer/channels").Middleware;

module.exports = {
    middlewares: [
        ChannelsMiddleware({
            adapter: {
                type: "NATS",
                options: {
                    nats: {
                        url: "nats://localhost:4222",
                        /** @type {ConnectionOptions} */
                        connectionOptions: {},
                        /** @type {StreamConfig} More info: https://docs.nats.io/jetstream/concepts/streams */
                        streamConfig: {},
                        /** @type {ConsumerOpts} More info: https://docs.nats.io/jetstream/concepts/consumers */
                        consumerOptions: {
                            config: {
                                // More info: https://docs.nats.io/jetstream/concepts/consumers#deliverpolicy-optstartseq-optstarttime
                                deliver_policy: "new",
                                // More info: https://docs.nats.io/jetstream/concepts/consumers#ackpolicy
                                ack_policy: "explicit",
                                // More info: https://docs.nats.io/jetstream/concepts/consumers#maxackpending
                                max_ack_pending: 1
                            }
                        }
                    },
                    maxInFlight: 10,
                    maxRetries: 3,
                    deadLettering: {
                        enabled: false,
                        queueName: "DEAD_LETTER"
                    }
                }
            }
        })
    ]
};

```

--------------------------------

### Configure Fake Adapter for Testing

Source: https://github.com/moleculerjs/moleculer-channels/blob/master/README.md

Configure the Channels middleware to use the 'Fake' adapter for unit and integration tests. This adapter uses the Moleculer event bus and does not support retries or dead-lettering.

```javascript
// moleculer.config.js
const ChannelsMiddleware = require("@moleculer/channels").Middleware;

module.exports = {
    middlewares: [
        ChannelsMiddleware({
            adapter: "Fake"
        })
    ]
};

```

--------------------------------

### Enable Context in All Channel Handlers

Source: https://github.com/moleculerjs/moleculer-channels/blob/master/README.md

Configure the Channels middleware to enable Moleculer Context for all channel handlers globally. This ensures `ctx.meta` and tracing information are available. Requires importing `ChannelsMiddleware`.

```javascript
// moleculer.config.js
const ChannelsMiddleware = require("@moleculer/channels").Middleware;

module.exports = {
    logger: true,

    middlewares: [
        ChannelsMiddleware({
            adapter: "redis://localhost:6379",
            // Enable context in all channel handlers
            context: true
        })
    ]
};

```

--------------------------------

### Configure Fake Adapter for Testing Moleculer Channels

Source: https://context7.com/moleculerjs/moleculer-channels/llms.txt

Utilize the Fake adapter for testing channel handlers without external dependencies. This adapter uses Moleculer's built-in event bus. It allows for direct invocation of channel handlers for unit testing.

```javascript
const { ServiceBroker } = require("moleculer");
const ChannelsMiddleware = require("@moleculer/channels").Middleware;

const broker = new ServiceBroker({
    middlewares: [
        ChannelsMiddleware({
            adapter: "Fake"
            // OR
            // adapter: { type: "Fake" }
        })
    ]
});

broker.createService({
    name: "test-service",

    channels: {
        "test.topic": {
            async handler(payload) {
                this.logger.info("Received:", payload);
                return { processed: true };
            }
        }
    }
});

// In tests, use emitLocalChannelHandler for direct handler testing
broker.start().then(async () => {
    // Test sending through channel
    await broker.sendToChannel("test.topic", { data: "test" });

    // Direct handler invocation for unit tests
    const service = broker.getLocalService("test-service");
    await service.emitLocalChannelHandler("test.topic", { data: "unit-test" });
});
```

--------------------------------

### Register Channel Tracing Middleware

Source: https://github.com/moleculerjs/moleculer-channels/blob/master/README.md

Integrate the `Tracing` middleware for context-based channel handlers to enable monitoring and tracing. This middleware requires `context: true` to be enabled for the channel handlers. Requires importing `Tracing` from `@moleculer/channels`.

```javascript
//moleculer.config.js
const TracingMiddleware = require("@moleculer/channels").Tracing;

module.exports = {
    logger: true,

    middlewares: [
        ChannelsMiddleware({
            adapter: "redis://localhost:6379",
            // Enable context in all channel handlers
            context: true
        }),
        TracingMiddleware()
    ]
};

```

--------------------------------

### Configure Redis Cluster Adapter

Source: https://github.com/moleculerjs/moleculer-channels/blob/master/README.md

Configuration for using the Redis adapter with a Redis Cluster. This includes specifying cluster nodes and optional ioredis cluster options.

```javascript
module.exports = {
    middlewares: [
        ChannelsMiddleware({
            adapter: {
                type: "Redis",
                options: {
                    redis: {
                        cluster: {
                            nodes: [
                                { port: 6380, host: "127.0.0.1" },
                                { port: 6381, host: "127.0.0.1" },
                                { port: 6382, host: "127.0.0.1" }
                            ],
                            options: {
                                /* More information: https://github.com/luin/ioredis#cluster */
                                redisOptions: {
                                    password: "fallback-password"
                                }
                            }
                        },
                        consumerOptions: {
                            // Timeout interval (in milliseconds) while waiting for new messages. By default never timeout
                            readTimeoutInterval: 0,
                            // Time (in milliseconds) after which pending messages are considered NACKed and should be claimed. Defaults to 1 hour.
                            minIdleTime: 5000,
                            // Interval (in milliseconds) between two claims
                            claimInterval: 100,
                            // "$" is a special ID. Consumers fetching data from the consumer group will only see new elements arriving in the stream.
                            // More info: https://redis.io/commands/XGROUP
                            startID: "$",
                            // Interval (in milliseconds) between message transfer into FAILED_MESSAGES channel
                            processingAttemptsInterval: 1000
                        }
                    },
                    deadLettering: {
                        transformErrorToHeaders: err => {

```

--------------------------------

### Produce AMQP Messages with Options

Source: https://github.com/moleculerjs/moleculer-channels/blob/master/README.md

Send messages to an AMQP channel with specific options for publishing, such as overriding exchange assertion behavior for a single message. This is useful for dynamic configurations.

```javascript
broker.sendToChannel(
    "order.created",
    {
        id: 1234,
        items: [
            /*...*/
        ]
    },
    {
        // Using specific `assertExchange()` options only for the current sending case
        publishAssertExchange: {
            // Enable/disable calling once `channel.assertExchange()` before first publishing in new exchange by `sendToChannel`
            enabled: true,
            // Options for `channel.assertExchange()` before publishing
            exchangeOptions: {
                /*...*/
            }
        }
    }
);
```

--------------------------------

### Configure Capped Streams with MAXLEN in Moleculer

Source: https://github.com/moleculerjs/moleculer-channels/blob/master/README.md

Demonstrates how to configure a Redis stream to be a capped stream by setting the `xaddMaxLen` option in `sendToChannel`. This limits the stream size to a maximum length.

```javascript
broker.sendToChannel(
    "order.created",
    {
        id: 1234,
        items: [
            /*...*/
        ]
    },
    {
        xaddMaxLen: "~1000"
    }
);
```

--------------------------------

### Using Multiple Adapters in a Service Schema

Source: https://github.com/moleculerjs/moleculer-channels/blob/master/README.md

Define channels for different adapters within a service schema. The `channels` property is used for the default adapter, while custom properties like `redisChannels` and `amqpChannels` (defined by `schemaProperty` in the middleware) are used for specific adapters.

```javascript
module.exports = {
    name: "payments",

    actions: {
        /*...*/
    },

    channels: {
        "default.options.topic": {
            group: "mygroup",
            async handler(payload) {
                /*...*/
            }
        }
    },
    redisChannels: {
        "redis.topic": {
            group: "mygroup",
            async handler(payload) {
                /*...*/
            }
        }
    },
    amqpChannels: {
        "amqp.topic": {
            group: "mygroup",
            async handler(payload) {
                /*...*/
            }
        }
    }
};
```

--------------------------------

### Redis Consumer Options

Source: https://github.com/moleculerjs/moleculer-channels/blob/master/README.md

Configuration options for the Redis stream consumer.

```APIDOC
## Redis Consumer Options

### Description
Configuration options for the Redis stream consumer.

### Parameters
#### Request Body
- **redis.consumerOptions.startID** (String) - Optional - Starting point when consumers fetch data from the consumer group. By default equals to `$`, i.e., consumers will only see new elements arriving in the stream. More info [here](https://redis.io/commands/XGROUP).
- **redis.consumerOptions.processingAttemptsInterval** (Number) - Optional - Interval (in milliseconds) between message transfer into `FAILED_MESSAGES` channel. Default: `0`.
```

--------------------------------

### Registering Multiple Adapters in Moleculer Config

Source: https://github.com/moleculerjs/moleculer-channels/blob/master/README.md

Configure Moleculer to use multiple channel adapters by registering instances of ChannelsMiddleware with different adapter configurations in the `moleculer.config.js` file. Each middleware can specify a different adapter type and custom properties.

```javascript
const ChannelsMiddleware = require("@moleculer/channels").Middleware;

// moleculer.config.js
module.exports = {
    logger: true,
    logLevel: "error",
    middlewares: [
        // Default options
        ChannelsMiddleware({
            adapter: {
                type: "Kafka",
                options: {}
            }
        }),
        ChannelsMiddleware({
            adapter: "Redis",
            schemaProperty: "redisChannels",
            sendMethodName: "sendToRedisChannel",
            adapterPropertyName: "redisAdapter"
        }),
        ChannelsMiddleware({
            adapter: "AMQP",
            schemaProperty: "amqpChannels",
            sendMethodName: "sendToAMQPChannel",
            adapterPropertyName: "amqpAdapter"
        })
    ]
};
```

--------------------------------

### Channel Options Configuration

Source: https://github.com/moleculerjs/moleculer-channels/blob/master/README.md

Configuration options for Moleculer channels.

```APIDOC
## Channel Options

This section details the available configuration options for Moleculer channels.

### Options Table

| Name | Type | Supported adapters | Description |
|---|---|---|---|
| `group` | `String` | * | Group name. It's used as a consumer group in adapter. By default, it's the full name of service (with version) |
| `maxInFlight` | `Number` | Redis | Max number of messages under processing at the same time. |
| `maxRetries` | `Number` | * | Maximum number of retries before sending the message to dead-letter-queue or drop. |
| `deadLettering.enabled` | `Boolean` | * | Enable "Dead-lettering" feature. |
| `deadLettering.queueName` | `String` | * | Name of dead-letter queue. |
| `deadLettering.errorInfoTTL` | `Number` | * | [Redis adapter only] TTL (in seconds) of error info messages stored in a separate hash. Default is `86400` (1 day). |
| `deadLettering.transformErrorToHeaders` | `Function` | * | Function to parse error instance to a plain object which will be stored in message headers. |
| `deadLettering.transformHeadersToErrorData` | `Function` | * | Function to parse plain error info object back to original data types headers. |
| `context` | `boolean` | * | Using Moleculer context in channel handlers. |
| `tracing` | `Object` | * | Tracing options same as [action tracing options](https://moleculer.services/docs/tracing.html#Customizing). It works only with `context: true`. |
| `handler` | `Function(payload: any, rawMessage: any)` | * | Channel handler function. It receives the payload at first parameter. The second parameter is a raw message which depends on the adapter. |
| `redis.startID` | `String` | Redis | Starting point when consumers fetch data from the consumer group. By default equals to `$`, i.e., consumers will only see new elements arriving in the stream. More info [here](https://redis.io/commands/XGROUP) |
| `redis.minIdleTime` | `Number` | Redis | Time (in milliseconds) after which pending messages are considered NACKed and should be claimed. Defaults to 1 hour. |
```

--------------------------------

### Custom Middleware for Channel Hooks

Source: https://github.com/moleculerjs/moleculer-channels/blob/master/README.md

Define custom middleware to wrap channel handlers and the `broker.sendToChannel` method. This allows for pre-processing and post-processing of messages and channel events. Requires importing `ChannelsMiddleware`.

```javascript
// moleculer.config.js
const ChannelsMiddleware = require("@moleculer/channels").Middleware;

const MyMiddleware = {
    name: "MyMiddleware",

    // Wrap the channel handlers
    localChannel(next, chan) {
        return async (msg, raw) => {
            this.logger.info(kleur.magenta(`  Before localChannel for '${chan.name}'`), msg);
            await next(msg, raw);
            this.logger.info(kleur.magenta(`  After localChannel for '${chan.name}'`), msg);
        };
    },

    // Wrap the `broker.sendToChannel` method
    sendToChannel(next) {
        return async (channelName, payload, opts) => {
            this.logger.info(kleur.yellow(`Before sendToChannel for '${channelName}'`), payload);
            await next(channelName, payload, opts);
            this.logger.info(kleur.yellow(`After sendToChannel for '${channelName}'`), payload);
        };
    }
};

module.exports = {
    logger: true,

    middlewares: [
        MyMiddleware,
        ChannelsMiddleware({
            adapter: "redis://localhost:6379"
        })
    ]
};

```

--------------------------------

### Publish Messages with sendToChannel

Source: https://context7.com/moleculerjs/moleculer-channels/llms.txt

Use `broker.sendToChannel()` to publish messages to a channel. This method supports various adapter-specific options for message delivery and routing.

```javascript
const { ServiceBroker } = require("moleculer");
const ChannelsMiddleware = require("@moleculer/channels").Middleware;

const broker = new ServiceBroker({
    middlewares: [
        ChannelsMiddleware({
            adapter: "redis://localhost:6379"
        })
    ]
});

broker.createService({
    name: "api",

    actions: {
        async createOrder(ctx) {
            const order = {
                id: ctx.params.orderId,
                items: ctx.params.items,
                customer: ctx.params.customer,
                total: ctx.params.total
            };

            // Save order to database
            await this.db.orders.insert(order);

            // Publish to channel (fire-and-forget from publisher perspective)
            await broker.sendToChannel("order.created", order, {
                // Custom headers passed to consumers
                headers: {
                    "x-priority": "high",
                    "x-source": "api-gateway"
                },

                // Redis: Limit stream length
                xaddMaxLen: "~1000",

                // AMQP: Message options
                persistent: true,
                ttl: 86400000,  // 24 hours
                priority: 5,

                // Kafka: Partitioning
                key: order.id.toString(),
                partition: 0,

                // AMQP: Send to specific queue instead of exchange
                // routingKey: "direct-queue-name"
            });

            return { status: "created", orderId: order.id };
        }
    }
});

broker.start();
```

--------------------------------

### Redis Cluster Configuration

Source: https://github.com/moleculerjs/moleculer-channels/blob/master/README.md

Options for configuring the Redis cluster connection.

```APIDOC
## Redis Cluster Configuration

### Description
Options for configuring the Redis cluster connection.

### Parameters
#### Request Body
- **redis.cluster** (Object) - Optional - Redis cluster connection options. More info [here](https://github.com/luin/ioredis#cluster).
- **redis.cluster.nodes** (Array) - Optional - Redis Cluster nodes list.
- **redis.cluster.clusterOptions** (Object) - Optional - Redis Cluster options.
```

--------------------------------

### Send Message Directly to AMQP Queue

Source: https://context7.com/moleculerjs/moleculer-channels/llms.txt

Demonstrates sending a message directly to a specific AMQP queue named 'my-direct-queue', bypassing any exchanges.

```javascript
await broker.sendToChannel("", ctx.params, {
    routingKey: "my-direct-queue"
});
```