class RedisStreamsBroker
implements MessageBrokerAdapter
Since 0.1.0

Redis Streams message broker implementation.

Uses Redis Streams for persistent message delivery with consumer groups for load-balanced processing.

Constructors

RedisStreamsBroker(
runtime: IRuntimeServices,
serializer: ISerializer,
)

Creates a new Redis Streams broker.

Parameters
runtime: IRuntimeServices

Runtime services for uuid, timestamps, and timers

serializer: ISerializer

Serializer for message payloads

optional
options: RedisStreamsOptions

Redis connection and polling options

Methods

Since 0.1.0
connect(): Promise<void>

Connects the broker to Redis.

Since 0.1.0
disconnect(): Promise<void>

Disconnects the broker and clears all subscriptions.

Since 0.1.0
isHealthy(): Promise<boolean>

Boolean port member (M70c): false only when positively unreachable.

Since 0.1.0
isReady(): boolean

Checks if the broker is connected.

Since 0.1.0
publish<T>(
topic: string,
message: T
): Promise<void>

Publishes a message to a topic (Redis stream).

publishWithHeaders<T>(
topic: string,
message: T,
headers: Readonly<Record<string, string>>
): Promise<void>

Publishes a message with framework-owned transport headers. @internal

Since 0.1.0
reachability(): Promise<boolean | undefined>

Tri-state backend reachability (M70c).

ioredis auto-reconnects via its default retry strategy, so the broker does not run a reconnect loop of its own; the truth is what ping() reports right now. true when ping() resolves, false when it rejects (the server is down or the socket is mid-reconnect), undefined when the injected client exposes no ping (a minimal fake) — in which case the indicator reports reachable: 'unknown' rather than lying.

Since 0.1.0
request<TReq, TRes>(
topic: string,
message: TReq,
options?: RequestOptions
): Promise<TRes>

Sends a request and awaits a single correlated reply.

requestWithHeaders<TReq, TRes>(
topic: string,
message: TReq,
headers: Readonly<Record<string, string>>,
options?: RequestOptions
): Promise<TRes>

Sends request-reply traffic with framework-owned headers. @internal

Since 0.1.0
respond<TReq, TRes>(
topic: string,
handler: RequestHandler<TReq, TRes>,
): Promise<ISubscription>

Registers a responder whose result is returned to the requesting caller.

Since 0.1.0
subscribe<T>(
topic: string,
handler: MessageHandler<T>,
): Promise<ISubscription>

Subscribes to a topic using Redis Streams consumer groups.

subscribeWithHeaders<T>(
topic: string,
handler: MessageHandler<T>,
): Promise<ISubscription>

Subscribes through the header-aware internal path. @internal

Usage

import { RedisStreamsBroker } from "messaging-plugin/src/index.ts";