Class RedisStreamsBroker

Redis Streams message broker implementation.

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

Since

0.1.0

Implements

MessageBrokerAdapter

Index

Constructors

constructor

Methods

connect disconnect isHealthy isReady publish publishWithHeaders reachability request requestWithHeaders respond subscribe subscribeWithHeaders

Constructors

constructor

new RedisStreamsBroker(
    runtime: IRuntimeServices,
    serializer: ISerializer,
    options?: RedisStreamsOptions,
): RedisStreamsBroker
Creates a new Redis Streams broker.
Parameters
- runtime: IRuntimeServices
  Runtime services for uuid, timestamps, and timers
- serializer: ISerializer
  Serializer for message payloads
- Optionaloptions: RedisStreamsOptions
  Redis connection and polling options
Returns RedisStreamsBroker

Throws
If a retry, reclaim, retention, sweep or publish bound is invalid
- Defined in packages/messaging-plugin/src/brokers/redis-streams-broker.ts:189

Methods

connect

connect(): Promise<void>
Connects the broker to Redis.

Returns Promise<void>
Resolves when connected

Since
0.1.0
Implementation of MessageBrokerAdapter.connect
- Defined in packages/messaging-plugin/src/brokers/redis-streams-broker.ts:265

disconnect

disconnect(): Promise<void>
Disconnects the broker and clears all subscriptions.

Returns Promise<void>
Resolves when disconnected

Since
0.1.0
Implementation of MessageBrokerAdapter.disconnect
- Defined in packages/messaging-plugin/src/brokers/redis-streams-broker.ts:282

isHealthy

isHealthy(): Promise<boolean>
Boolean port member (M70c): false only when positively unreachable.

Returns Promise<boolean>
true when reachable or unprobeable, false when ping fails

Since
0.1.0
Implementation of MessageBrokerAdapter.isHealthy
- Defined in packages/messaging-plugin/src/brokers/redis-streams-broker.ts:334

isReady

isReady(): boolean
Checks if the broker is connected.

Returns boolean
true if connected, false otherwise

Since
0.1.0
Implementation of MessageBrokerAdapter.isReady
- Defined in packages/messaging-plugin/src/brokers/redis-streams-broker.ts:303

publish

publish<T>(topic: string, message: T, options?: PublishOptions): Promise<void>
Publishes a message to a topic (Redis stream).
Type Parameters
- T
  The message payload type
Parameters
- topic: string
  The topic (stream name) to publish to
- message: T
  The message payload
- Optionaloptions: PublishOptions
  Caller headers, ordering key and de-duplication id
Returns Promise<void>
Resolves when published

Throws
When publishTimeoutMs expires; delivery remains possible

Since
0.1.0
Implementation of MessageBrokerAdapter.publish
- Defined in packages/messaging-plugin/src/brokers/redis-streams-broker.ts:350

publishWithHeaders

publishWithHeaders<T>(
    topic: string,
    message: T,
    headers: Readonly<Record<string, string>>,
    options?: PublishOptions,
): Promise<void>
Internal
Publishes with framework-owned headers under the configured acknowledgement bound.
Type Parameters
- T
Parameters
- topic: string
  Destination topic
- message: T
  Payload to serialize
- headers: Readonly<Record<string, string>>
  Framework-owned transport headers
- Optionaloptions: PublishOptions
  Caller headers, ordering key and de-duplication id
Returns Promise<void>
Resolves when the transport acknowledges the publish

Throws
When publishTimeoutMs expires; delivery remains possible
Implementation of MessageBrokerAdapter.publishWithHeaders
- Defined in packages/messaging-plugin/src/brokers/redis-streams-broker.ts:364

reachability

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.

Returns Promise<boolean | undefined>
true/false/undefined as described

Since
0.1.0
Implementation of MessageBrokerAdapter.reachability
- Defined in packages/messaging-plugin/src/brokers/redis-streams-broker.ts:320

request

request<TReq, TRes>(
    topic: string,
    message: TReq,
    options?: RequestOptions,
): Promise<TRes>
Sends a request and awaits a single correlated reply.
Type Parameters
- TReq
  The request payload type
- TRes
  The reply payload type
Parameters
- topic: string
  Destination topic a responder is listening on
- message: TReq
  The request payload
- Optionaloptions: RequestOptions
  Reply timeout behavior
Returns Promise<TRes>
The reply payload

Since
0.1.0
Implementation of MessageBrokerAdapter.request
- Defined in packages/messaging-plugin/src/brokers/redis-streams-broker.ts:719

requestWithHeaders

requestWithHeaders<TReq, TRes>(
    topic: string,
    message: TReq,
    headers: Readonly<Record<string, string>>,
    options?: RequestOptions,
): Promise<TRes>
Internal
Sends request-reply traffic with framework-owned headers.
Type Parameters
- TReq
- TRes
Parameters
- topic: string
- message: TReq
- headers: Readonly<Record<string, string>>
- Optionaloptions: RequestOptions
Returns Promise<TRes>
Implementation of MessageBrokerAdapter.requestWithHeaders
- Defined in packages/messaging-plugin/src/brokers/redis-streams-broker.ts:724

respond

respond<TReq, TRes>(
    topic: string,
    handler: RequestHandler<TReq, TRes>,
    options?: SubscribeOptions,
): Promise<ISubscription>
Registers a responder whose result is returned to the requesting caller.
Type Parameters
- TReq
  The request payload type
- TRes
  The reply payload type
Parameters
- topic: string
  The request topic to respond on
- handler: RequestHandler<TReq, TRes>
  Invoked per request; its result is returned to the caller
- Optionaloptions: SubscribeOptions
  Consumer group behavior
Returns Promise<ISubscription>
The active subscription

Since
0.1.0
Implementation of MessageBrokerAdapter.respond
- Defined in packages/messaging-plugin/src/brokers/redis-streams-broker.ts:744

subscribe<T>(
    topic: string,
    handler: MessageHandler<T>,
    options?: SubscribeOptions,
): Promise<ISubscription>
Subscribes to a topic using Redis Streams consumer groups.
Type Parameters
- T
  The message payload type
Parameters
- topic: string
  The topic (stream name) to subscribe to
- handler: MessageHandler<T>
  The handler to invoke for each message
- Optionaloptions: SubscribeOptions
  Optional subscription options (queue for consumer group)
Returns Promise<ISubscription>
The subscription handle

Since
0.1.0
Implementation of MessageBrokerAdapter.subscribe
- Defined in packages/messaging-plugin/src/brokers/redis-streams-broker.ts:405

subscribeWithHeaders

subscribeWithHeaders<T>(
    topic: string,
    handler: MessageHandler<T>,
    options?: SubscribeOptions,
): Promise<ISubscription>
Internal
Subscribes through the header-aware internal path.
Type Parameters
- T
Parameters
- topic: string
- handler: MessageHandler<T>
- Optionaloptions: SubscribeOptions
Returns Promise<ISubscription>
Implementation of MessageBrokerAdapter.subscribeWithHeaders
- Defined in packages/messaging-plugin/src/brokers/redis-streams-broker.ts:700

Class RedisStreamsBroker

Since

Implements

Index

Constructors

Methods

Constructors

constructor

Parameters

Returns RedisStreamsBroker

Throws

Methods

connect

Returns Promise<void>

Since

disconnect

Returns Promise<void>

Since

isHealthy

Returns Promise<boolean>

Since

isReady

Returns boolean

Since

publish

Type Parameters

Parameters

Returns Promise<void>

Throws

Since

publishWithHeaders

Type Parameters

Parameters

Returns Promise<void>

Throws

reachability

Returns Promise<boolean | undefined>

Since

request

Type Parameters

Parameters

Returns Promise<TRes>

Since

requestWithHeaders

Type Parameters

Parameters

Returns Promise<TRes>

respond

Type Parameters

Parameters

Returns Promise<ISubscription>

Since

subscribe

Type Parameters

Parameters

Returns Promise<ISubscription>

Since

subscribeWithHeaders

Type Parameters

Parameters

Returns Promise<ISubscription>

Settings

On This Page