This is an automated email from the ASF dual-hosted git repository.
RongtongJin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/rocketmq-clients.git
The following commit(s) were added to refs/heads/master by this push:
new fa490c2a Feat/nodejs message interceptor hooks (#1384)
fa490c2a is described below
commit fa490c2a7a7f81b06ecdd554110b3221d47f6541
Author: zhaohai <[email protected]>
AuthorDate: Fri Oct 9 11:20:45 2026 +0800
Feat/nodejs message interceptor hooks (#1384)
* feat(nodejs): support NOTIFY_UNSUBSCRIBE_LITE_COMMAND and message
interceptor hooks
- BaseClient: add onNotifyUnsubscribeLiteCommand hook (warn by default,
overridden by lite push consumers)
- TelemetrySession: dispatch NOTIFY_UNSUBSCRIBE_LITE_COMMAND (was falling
through to the unknown-command branch)
- LitePushConsumerImpl: override the command handler with the
endpoints-aware
signature and forward to the lite subscription manager
- BaseClient: drop the commented-out recover-orphaned-transaction reply
(warn only, no reply)
- New hook package: MessageHookPoints, MessageHookPointsStatus,
Attribute/AttributeKey, MessageInterceptor(+Context/Impl),
CompositedMessageInterceptor and InflightRequestCountInterceptor; clients
expose addMessageInterceptor and doBefore/doAfter, configurable via
BaseClientOptions#messageInterceptor
- Wire hook points: SEND (producer), RECEIVE/ACK/CHANGE_INVISIBLE_DURATION
(consumer), CONSUME (consume task); push consumers register the built-in
inflight request count interceptor
- Tests: cover telemetry dispatch and interceptor chain ordering, error
isolation, attribute propagation and inflight counting
* feat(nodejs): wire COMMIT_TRANSACTION / ROLLBACK_TRANSACTION /
FORWARD_TO_DLQ hook points
Complete the remaining 3 of the 8 MessageHookPoints:
- K-2 Producer#endTransaction fires COMMIT_TRANSACTION /
ROLLBACK_TRANSACTION
(selected by resolution === COMMIT), hook status OK/ERROR taken from the
response Code or the thrown error.
- K-3 Consumer#forwardMessageToDeadLetterQueueViaRpc fires FORWARD_TO_DLQ
around every DLQ RPC attempt. ProcessQueue now forwards the MessageView
so interceptors observe the real message.
- Add 4 regression tests (transaction hooks, DLQ hook, error paths);
offline suite 119/119, integration 16/16.
* fix(nodejs): align message interceptor hooks with the Java client
Reviewer feedback on the hook framework: RECEIVE had to be intercepted
before the RPC so in-flight request counters observe real requests, the
PushConsumer settlement paths had to fire the same ACK /
CHANGE_INVISIBLE_DURATION hooks as SimpleConsumer (including retries),
and the composite interceptor had to keep Java's cumulative context
semantics so chained interceptors see each other's attributes.
---
nodejs/src/consumer/Consumer.ts | 68 +++---
nodejs/src/consumer/ProcessQueue.ts | 2 +
nodejs/src/hook/CompositedMessageInterceptor.ts | 12 +-
nodejs/test/hook.test.ts | 285 ++++++++++++++++++++++--
4 files changed, 320 insertions(+), 47 deletions(-)
diff --git a/nodejs/src/consumer/Consumer.ts b/nodejs/src/consumer/Consumer.ts
index f312fe4f..233db92c 100644
--- a/nodejs/src/consumer/Consumer.ts
+++ b/nodejs/src/consumer/Consumer.ts
@@ -64,6 +64,12 @@ export abstract class Consumer extends BaseClient {
const timeout = this.requestTimeout + awaitDuration;
let status: Status.AsObject | undefined;
+ // RECEIVE hook point, mirroring Java ProcessQueueImpl: intercept before
the
+ // RPC so that interceptors which count in-flight requests observe the real
+ // request, and carry this same context into doAfter.
+ const context = new
MessageInterceptorContextImpl(MessageHookPoints.RECEIVE);
+ this.doBefore(context, []);
+
try {
this.logger.debug?.('Receiving messages from broker, topic=%s,
endpoints=%s, batchSize=%d, clientId=%s',
request.getMessageQueue()?.getTopic()?.getName(), endpoints,
request.getBatchSize(), (this as any).clientId);
@@ -97,17 +103,11 @@ export abstract class Consumer extends BaseClient {
}
const messages = messageList.map(message => new MessageView(message, mq,
transportDeliveryTimestamp));
- // RECEIVE hook point, mirroring Java ProcessQueueImpl.
- const context = new
MessageInterceptorContextImpl(MessageHookPoints.RECEIVE,
MessageHookPointsStatus.OK);
- this.doBefore(context, messages);
this.doAfter(MessageInterceptorContextImpl.withStatus(context,
MessageHookPointsStatus.OK), messages);
return messages;
} catch (err) {
this.logger.error('Failed to receive messages, topic=%s, endpoints=%s,
clientId=%s, error=%s',
request.getMessageQueue()?.getTopic()?.getName(), endpoints, (this as
any).clientId, err);
- // RECEIVE hook point with error status.
- const context = new
MessageInterceptorContextImpl(MessageHookPoints.RECEIVE);
- this.doBefore(context, []);
this.doAfter(MessageInterceptorContextImpl.withStatus(context,
MessageHookPointsStatus.ERROR), []);
throw err;
}
@@ -173,33 +173,37 @@ export abstract class Consumer extends BaseClient {
}
/**
- * Expose public methods for ProcessQueue to access RPC operations
+ * Expose public methods for ProcessQueue to access RPC operations. Each
attempt
+ * — including retries driven by ProcessQueue — triggers the matching hook
point,
+ * so PushConsumer and SimpleConsumer observability stays consistent.
*/
- async ackMessageViaRpc(endpoints: any, request: AckMessageRequest, timeout:
number) {
- const res = await this.rpcClientManager.ackMessage(endpoints, request,
timeout);
- return res;
+ async ackMessageViaRpc(endpoints: any, request: AckMessageRequest, timeout:
number, messageView: MessageView) {
+ return this.#invokeWithHook(
+ MessageHookPoints.ACK,
+ messageView,
+ () => this.rpcClientManager.ackMessage(endpoints, request, timeout),
+ );
}
- async changeInvisibleDurationViaRpc(endpoints: any, request:
ChangeInvisibleDurationRequest, timeout: number) {
- const res = await this.rpcClientManager.changeInvisibleDuration(endpoints,
request, timeout);
- return res;
+ async changeInvisibleDurationViaRpc(endpoints: any, request:
ChangeInvisibleDurationRequest, timeout: number,
+ messageView: MessageView) {
+ return this.#invokeWithHook(
+ MessageHookPoints.CHANGE_INVISIBLE_DURATION,
+ messageView,
+ () => this.rpcClientManager.changeInvisibleDuration(endpoints, request,
timeout),
+ );
}
- /**
- * Forward the message to the dead letter queue, mirroring Java
- * PushConsumerImpl#forwardMessageToDeadLetterQueue. The FORWARD_TO_DLQ hook
- * point is triggered around every RPC attempt.
- */
- async forwardMessageToDeadLetterQueueViaRpc(endpoints: any, request: any,
timeout: number,
- messageView: MessageView) {
- // FORWARD_TO_DLQ hook point, mirroring Java
PushConsumerImpl#forwardMessageToDeadLetterQueue.
- const context = new
MessageInterceptorContextImpl(MessageHookPoints.FORWARD_TO_DLQ);
+ async #invokeWithHook<T extends { getStatus(): Status | undefined
}>(hookPoint: MessageHookPoints,
+ messageView: MessageView,
+ rpc: () => Promise<T>): Promise<T> {
+ const context = new MessageInterceptorContextImpl(hookPoint);
const generalMessages: GeneralMessage[] = [ messageView ];
this.doBefore(context, generalMessages);
try {
- const response = await
this.rpcClientManager.forwardMessageToDeadLetterQueue(endpoints, request,
timeout);
- const hookStatus = response.getStatus()?.getCode() === Code.OK ?
- MessageHookPointsStatus.OK : MessageHookPointsStatus.ERROR;
+ const response = await rpc();
+ const hookStatus = response.getStatus()?.getCode() === Code.OK
+ ? MessageHookPointsStatus.OK : MessageHookPointsStatus.ERROR;
this.doAfter(MessageInterceptorContextImpl.withStatus(context,
hookStatus), generalMessages);
return response;
} catch (err) {
@@ -208,6 +212,20 @@ export abstract class Consumer extends BaseClient {
}
}
+ /**
+ * Forward the message to the dead letter queue, mirroring Java
+ * PushConsumerImpl#forwardMessageToDeadLetterQueue. The FORWARD_TO_DLQ hook
+ * point is triggered around every RPC attempt.
+ */
+ async forwardMessageToDeadLetterQueueViaRpc(endpoints: any, request: any,
timeout: number,
+ messageView: MessageView) {
+ return this.#invokeWithHook(
+ MessageHookPoints.FORWARD_TO_DLQ,
+ messageView,
+ () => this.rpcClientManager.forwardMessageToDeadLetterQueue(endpoints,
request, timeout),
+ );
+ }
+
/**
* Get the consumer group name.
*
diff --git a/nodejs/src/consumer/ProcessQueue.ts
b/nodejs/src/consumer/ProcessQueue.ts
index 8563b06b..c485da09 100644
--- a/nodejs/src/consumer/ProcessQueue.ts
+++ b/nodejs/src/consumer/ProcessQueue.ts
@@ -247,6 +247,7 @@ export class ProcessQueue {
endpoints,
this.#consumer.wrapAckMessageRequest(messageView),
this.#consumer.requestTimeoutValue,
+ messageView,
);
const status = response.getStatus()?.toObject();
if (status?.code === Code.INVALID_RECEIPT_HANDLE) {
@@ -284,6 +285,7 @@ export class ProcessQueue {
endpoints,
this.#consumer.wrapChangeInvisibleDurationRequest(messageView,
duration),
this.#consumer.requestTimeoutValue,
+ messageView,
);
const status = response.getStatus()?.toObject();
if (status?.code === Code.INVALID_RECEIPT_HANDLE) {
diff --git a/nodejs/src/hook/CompositedMessageInterceptor.ts
b/nodejs/src/hook/CompositedMessageInterceptor.ts
index a884d0bc..1276db60 100644
--- a/nodejs/src/hook/CompositedMessageInterceptor.ts
+++ b/nodejs/src/hook/CompositedMessageInterceptor.ts
@@ -48,9 +48,10 @@ export class CompositedMessageInterceptor implements
MessageInterceptor {
doBefore(context0: MessageInterceptorContext, messages: GeneralMessage[]) {
const attributeMap = new Map<number, Map<AttributeKey<any>,
Attribute<any>>>();
+ // A single cumulative context is shared by the whole chain, mirroring the
Java
+ // implementation, so that later interceptors observe what earlier ones
recorded.
+ const context = new
MessageInterceptorContextImpl(context0.getMessageHookPoints(),
context0.getStatus());
for (let index = 0; index < this.#interceptors.length; index++) {
- const context = new
MessageInterceptorContextImpl(context0.getMessageHookPoints(),
context0.getStatus());
- // Share the attributes accumulated by the caller and previous
interceptors.
for (const [ key, value ] of context0.getAttributes()) {
context.putAttribute(key, value);
}
@@ -66,13 +67,6 @@ export class CompositedMessageInterceptor implements
MessageInterceptor {
attributeMap.set(index, context.getAttributes());
}
context0.putAttribute(INTERCEPTOR_ATTRIBUTES_KEY,
Attribute.create(attributeMap));
- // Propagate attributes collected by the first interceptor to the caller
context.
- const firstAttributes = attributeMap.get(0);
- if (firstAttributes) {
- for (const [ key, value ] of firstAttributes) {
- context0.putAttribute(key, value);
- }
- }
}
doAfter(context0: MessageInterceptorContext, messages: GeneralMessage[]) {
diff --git a/nodejs/test/hook.test.ts b/nodejs/test/hook.test.ts
index c6e72039..f9e31b39 100644
--- a/nodejs/test/hook.test.ts
+++ b/nodejs/test/hook.test.ts
@@ -19,16 +19,24 @@
* Tests for the message interceptor / hook framework and the
* NOTIFY_UNSUBSCRIBE_LITE_COMMAND telemetry dispatch:
* - Telemetry command dispatch vs Java (J-1)
+ * - RECEIVE hook timing and context reuse (J-2)
* - Message interceptor chain (J-3)
* - Transaction hook points (K-2)
* - FORWARD_TO_DLQ hook point (K-3)
+ * - PushConsumer settlement hooks (K-4)
*/
import { describe, it } from 'node:test';
import * as assert from 'node:assert';
import { Endpoints } from '../src/route';
import { Producer } from '../src/producer';
-import { Consumer } from '../src/consumer';
+import {
+ ConsumeResult,
+ ConsumeResultSuspend,
+ Consumer,
+ FilterExpression,
+ ProcessQueue,
+} from '../src/consumer';
import { Message } from '../src/message';
import { Settings } from '../src/client';
import {
@@ -46,12 +54,67 @@ import { Status } from
'../proto/apache/rocketmq/v2/definition_pb';
import {
HeartbeatRequest,
NotifyClientTerminationRequest,
+ ReceiveMessageRequest,
+ ReceiveMessageResponse,
} from '../proto/apache/rocketmq/v2/service_pb';
function statusOf(code: Code): Status.AsObject {
return { code, message: 'mock message', requestId: '' } as unknown as
Status.AsObject;
}
+/**
+ * Minimal response shape accepted by both the hook wrapper (which reads
+ * `getStatus().getCode()`) and ProcessQueue (which reads
`getStatus().toObject()`).
+ */
+function responseOf(code: Code) {
+ return {
+ getStatus: () => ({
+ getCode: () => code,
+ toObject: () => statusOf(code),
+ }),
+ };
+}
+
+/**
+ * Concrete {@link Consumer} used to exercise the hook points without a broker.
+ */
+class HookTestConsumer extends Consumer {
+ protected getSettings(): Settings {
+ return {} as Settings;
+ }
+ protected wrapHeartbeatRequest(): HeartbeatRequest {
+ return new HeartbeatRequest();
+ }
+ protected wrapNotifyClientTerminationRequest():
NotifyClientTerminationRequest {
+ return new NotifyClientTerminationRequest();
+ }
+
+ /** Expose the protected receive entry point used by ProcessQueue. */
+ receiveFrom(mq: any, awaitDuration = 0) {
+ return this.receiveMessage(new ReceiveMessageRequest(), mq, awaitDuration);
+ }
+}
+
+/** Records the hook points as `before:POINT` / `after:POINT:STATUS` strings.
*/
+function eventRecorder(events: string[], key?: AttributeKey<string>, value?:
string): MessageInterceptor {
+ return {
+ doBefore: context => {
+
events.push(`before:${MessageHookPoints[context.getMessageHookPoints()]}`);
+ if (key && value !== undefined) {
+ context.putAttribute(key, Attribute.create(value));
+ }
+ },
+ doAfter: context => {
+ const status = context.getStatus();
+ events.push(`after:${MessageHookPoints[context.getMessageHookPoints()]}:`
+ + `${status === undefined ? 'UNSET' :
MessageHookPointsStatus[status]}`);
+ if (key) {
+ events.push(`attr:${context.getAttribute(key)?.get()}`);
+ }
+ },
+ };
+}
+
describe('Telemetry command dispatch vs Java (J-1)', () => {
it('should dispatch NOTIFY_UNSUBSCRIBE_LITE_COMMAND to the client', () => {
const { EventEmitter } = require('node:events');
@@ -84,6 +147,72 @@ describe('Telemetry command dispatch vs Java (J-1)', () => {
});
});
+describe('RECEIVE hook timing vs Java (J-2)', () => {
+ const mq = { broker: { endpoints: new Endpoints('127.0.0.1:8081') } } as any;
+
+ function createConsumer(messageInterceptor: MessageInterceptor |
MessageInterceptor[]) {
+ return new HookTestConsumer({
+ endpoints: '127.0.0.1:8081',
+ namespace: '',
+ consumerGroup: 'TestGroup',
+ messageInterceptor,
+ } as any);
+ }
+
+ function okReceiveResponse() {
+ return [ new ReceiveMessageResponse().setStatus(new
Status().setCode(Code.OK)) ];
+ }
+
+ it('should run RECEIVE doBefore before the RPC so in-flight requests are
observable', async () => {
+ const events: string[] = [];
+ const inflight = new InflightRequestCountInterceptor();
+ const consumer = createConsumer([ eventRecorder(events), inflight ]);
+ let inflightDuringRpc = -1;
+ (consumer as any).rpcClientManager = {
+ receiveMessage: async () => {
+ inflightDuringRpc = inflight.getInflightReceiveRequestCount();
+ events.push('rpc');
+ return okReceiveResponse();
+ },
+ };
+ assert.deepStrictEqual(await consumer.receiveFrom(mq), []);
+ // Before the fix doBefore ran after the RPC, so the counter was always 0
here.
+ assert.strictEqual(inflightDuringRpc, 1);
+ assert.strictEqual(inflight.getInflightReceiveRequestCount(), 0);
+ assert.deepStrictEqual(events, [ 'before:RECEIVE', 'rpc',
'after:RECEIVE:OK' ]);
+ });
+
+ it('should carry the doBefore context into doAfter when messages are
received', async () => {
+ const events: string[] = [];
+ const key = AttributeKey.create<string>('receive-attr');
+ const consumer = createConsumer(eventRecorder(events, key,
'set-in-before'));
+ (consumer as any).rpcClientManager = {
+ receiveMessage: async () => okReceiveResponse(),
+ };
+ await consumer.receiveFrom(mq);
+ assert.deepStrictEqual(events, [
+ 'before:RECEIVE', 'after:RECEIVE:OK', 'attr:set-in-before',
+ ]);
+ });
+
+ it('should reuse the same context and fire once when the RPC fails', async
() => {
+ const events: string[] = [];
+ const key = AttributeKey.create<string>('receive-attr');
+ const inflight = new InflightRequestCountInterceptor();
+ const consumer = createConsumer([ eventRecorder(events, key,
'set-in-before'), inflight ]);
+ (consumer as any).rpcClientManager = {
+ receiveMessage: async () => {
+ throw new Error('receive failed');
+ },
+ };
+ await assert.rejects(async () => consumer.receiveFrom(mq));
+ assert.strictEqual(inflight.getInflightReceiveRequestCount(), 0);
+ assert.deepStrictEqual(events, [
+ 'before:RECEIVE', 'after:RECEIVE:ERROR', 'attr:set-in-before',
+ ]);
+ });
+});
+
describe('Transaction hook points vs Java (K-2)', () => {
function createProducer(events: string[]) {
const producer = new Producer({
@@ -155,18 +284,6 @@ describe('Transaction hook points vs Java (K-2)', () => {
});
describe('FORWARD_TO_DLQ hook point vs Java (K-3)', () => {
- class HookTestConsumer extends Consumer {
- protected getSettings(): Settings {
- return {} as Settings;
- }
- protected wrapHeartbeatRequest(): HeartbeatRequest {
- return new HeartbeatRequest();
- }
- protected wrapNotifyClientTerminationRequest():
NotifyClientTerminationRequest {
- return new NotifyClientTerminationRequest();
- }
- }
-
function createConsumer(events: string[]) {
return new HookTestConsumer({
endpoints: '127.0.0.1:8081',
@@ -219,6 +336,118 @@ describe('FORWARD_TO_DLQ hook point vs Java (K-3)', () =>
{
});
});
+describe('PushConsumer settlement hooks vs Java (K-4)', () => {
+ const mq = {
+ topic: { name: 'TopicTest' },
+ broker: { name: 'broker-a', endpoints: new Endpoints('127.0.0.1:8081') },
+ queueId: 0,
+ } as any;
+
+ /** PushConsumer-shaped surface that {@link ProcessQueue} relies on for
settlement. */
+ class FakePushConsumer extends HookTestConsumer {
+ get requestTimeoutValue(): number {
+ return 3000;
+ }
+ wrapAckMessageRequest(): any {
+ return {};
+ }
+ wrapChangeInvisibleDurationRequest(): any {
+ return {};
+ }
+ getRetryPolicy() {
+ return { getMaxAttempts: () => 1, getNextAttemptDelay: () => 0 };
+ }
+ getPushConsumerSettings() {
+ // ProcessQueue reads the semaphore bound, long-polling timeout and
receive
+ // batch size from the subscription settings; the defaults mirror
+ // PushSubscriptionSettings.
+ return {
+ getConsumeConcurrentlyMax: () => 32,
+ getLongPollingTimeout: () => 30000,
+ getReceiveBatchSize: () => 32,
+ };
+ }
+ incrementConsumptionOkQuantity(): void { /* not collected in tests */ }
+ incrementConsumptionErrorQuantity(): void { /* not collected in tests */ }
+ protected isLiteConsumer(): boolean {
+ // Only lite consumers keep a ConsumeResultSuspend, which drives the
+ // CHANGE_INVISIBLE_DURATION settlement path.
+ return true;
+ }
+ }
+
+ function createConsumer(events: string[]) {
+ return new FakePushConsumer({
+ endpoints: '127.0.0.1:8081',
+ namespace: '',
+ consumerGroup: 'TestGroup',
+ messageInterceptor: eventRecorder(events),
+ } as any);
+ }
+
+ function messageView() {
+ return {
+ endpoints: new Endpoints('127.0.0.1:8081'),
+ topic: 'TopicTest',
+ messageId: '010000000000000001',
+ deliveryAttempt: 1,
+ body: Buffer.from('body'),
+ incrementAndGetDeliveryAttempt: () => 2,
+ } as any;
+ }
+
+ it('should fire the ACK hook from ProcessQueue once per attempt, including
retries', async () => {
+ const events: string[] = [];
+ const consumer = createConsumer(events);
+ let attempts = 0;
+ (consumer as any).rpcClientManager = {
+ ackMessage: async () => {
+ attempts++;
+ events.push('rpc');
+ // The first attempt asks for a retry, which sleeps for the ack
backoff.
+ return responseOf(attempts === 1 ? Code.INTERNAL_ERROR : Code.OK);
+ },
+ };
+ const processQueue = new ProcessQueue(consumer as any, mq, new
FilterExpression('*'));
+ await processQueue.eraseFifoMessage(messageView(), ConsumeResult.SUCCESS);
+ assert.strictEqual(attempts, 2);
+ assert.deepStrictEqual(events, [
+ 'before:ACK', 'rpc', 'after:ACK:ERROR',
+ 'before:ACK', 'rpc', 'after:ACK:OK',
+ ]);
+ });
+
+ it('should fire the CHANGE_INVISIBLE_DURATION hook when consumption is
suspended', async () => {
+ const events: string[] = [];
+ const consumer = createConsumer(events);
+ (consumer as any).rpcClientManager = {
+ changeInvisibleDuration: async () => {
+ events.push('rpc');
+ return responseOf(Code.OK);
+ },
+ };
+ const processQueue = new ProcessQueue(consumer as any, mq, new
FilterExpression('*'));
+ await processQueue.eraseFifoMessage(messageView(),
ConsumeResultSuspend.of(50));
+ assert.deepStrictEqual(events, [
+ 'before:CHANGE_INVISIBLE_DURATION', 'rpc',
'after:CHANGE_INVISIBLE_DURATION:OK',
+ ]);
+ });
+
+ it('should mark the settlement hook as ERROR when the RPC throws', async ()
=> {
+ const events: string[] = [];
+ const consumer = createConsumer(events);
+ (consumer as any).rpcClientManager = {
+ ackMessage: async () => {
+ throw new Error('ack failed');
+ },
+ };
+ await assert.rejects(async () => {
+ await consumer.ackMessageViaRpc(new Endpoints('127.0.0.1:8081'), {} as
any, 3000, messageView());
+ });
+ assert.deepStrictEqual(events, [ 'before:ACK', 'after:ACK:ERROR' ]);
+ });
+});
+
describe('Message interceptor chain (J-3)', () => {
it('should run doBefore in order and doAfter in reverse order', () => {
const calls: string[] = [];
@@ -263,6 +492,36 @@ describe('Message interceptor chain (J-3)', () => {
assert.strictEqual(seen, 'v');
});
+ it('should accumulate attributes across interceptors in doBefore', () => {
+ const first = AttributeKey.create<string>('first');
+ const second = AttributeKey.create<string>('second');
+ const seen: Array<Array<string | undefined>> = [];
+ const afterSeen: Array<Array<string | undefined>> = [];
+ const mk = (put?: AttributeKey<string>, value?: string):
MessageInterceptor => ({
+ doBefore: context => {
+ seen.push([ context.getAttribute(first)?.get(),
context.getAttribute(second)?.get() ]);
+ if (put && value !== undefined) {
+ context.putAttribute(put, Attribute.create(value));
+ }
+ },
+ doAfter: context => {
+ afterSeen.push([ context.getAttribute(first)?.get(),
context.getAttribute(second)?.get() ]);
+ },
+ });
+ const composited = new CompositedMessageInterceptor([
+ mk(first, 'v1'),
+ mk(second, 'v2'),
+ mk(),
+ ]);
+ const context = new MessageInterceptorContextImpl(MessageHookPoints.SEND);
+ composited.doBefore(context, []);
+ // Each interceptor must observe everything the previous ones recorded.
+ assert.deepStrictEqual(seen, [[ undefined, undefined ], [ 'v1', undefined
], [ 'v1', 'v2' ]]);
+ composited.doAfter(MessageInterceptorContextImpl.withStatus(context,
MessageHookPointsStatus.OK), []);
+ // doAfter runs in reverse order and keeps the accumulated attributes.
+ assert.deepStrictEqual(afterSeen, [[ 'v1', 'v2' ], [ 'v1', 'v2' ], [ 'v1',
'v2' ]]);
+ });
+
it('should count inflight receive requests via
InflightRequestCountInterceptor', () => {
const interceptor = new InflightRequestCountInterceptor();
const composited = new CompositedMessageInterceptor([ interceptor ]);
