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 cf0893a9 feat(nodejs): implement LiteSimpleConsumer (#1379)
cf0893a9 is described below
commit cf0893a9259c704ecb6b70a94691735aa977dd0b
Author: zhaohai <[email protected]>
AuthorDate: Thu Sep 17 14:46:21 2026 +0800
feat(nodejs): implement LiteSimpleConsumer (#1379)
* feat(nodejs): implement LiteSimpleConsumer
Port the lite simple consumer from the reference client implementation:
- Add `LiteSimpleConsumer` (interface + builder options) and
`LiteSimpleConsumerImpl` with `subscribeLite` / `unsubscribeLite`,
`receive`, `ack`, `changeInvisibleDuration` and startup/shutdown flow
- Rework `LiteSubscriptionManager` to be host-agnostic and sync lite
subscriptions (FULL / PARTIAL_ADD / PARTIAL_REMOVE) to every sync
endpoint, with quota handling and notify-unsubscribe support
- Handle `NOTIFY_UNSUBSCRIBE_LITE_COMMAND` in `BaseClient` /
`TelemetrySession` and align the callback signature across consumers
- Fix lite ack path: include `liteTopic` in ack and
change-invisible-duration requests for lite consumers, matching the
server-side receipt handle resolution
- Add offline unit tests (builder validation, subscription sync,
route pruning) and real-cluster integration tests (end-to-end
send/receive/ack, subscribe-then-send delivery, unsubscribe-stop,
blank topic validation)
* docs(nodejs): add LiteSimpleConsumer example
Add a runnable LiteSimpleConsumer example and document the lite topic
usage in the Node.js README:
- examples/LiteSimpleConsumerExample.ts: build a lite simple consumer
bound to a parent topic, subscribe/unsubscribe lite topics with
OffsetOption, receive-ack loop with changeInvisibleDuration fallback
on ack failure, and graceful shutdown on SIGINT
- README: new "Lite Topic" section covering the server prerequisites,
producer-side liteTopic send, and the LiteSimpleConsumer pull/ack flow
* refactor(nodejs): stop exporting LiteSimpleConsumerImpl from the consumer
barrel
The consumer barrel re-exported LiteSimpleConsumerImpl, so its host-only
helpers (getRpcClientManager, getLogger, getRequestTimeout,
getSyncEndpoints)
ended up in the published .d.ts and became public API - hard to change once
the
package is released.
Export the interface + builder only (`export * from
'./LiteSimpleConsumer'`).
Users still obtain instances through `build()`, which returns the
`LiteSimpleConsumer` interface. Tests import the impl by file path, so they
are
unaffected. LitePushConsumerImpl is intentionally left exported.
Verified:
- emitted declarations: the entry chain (index.d.ts -> consumer/index.d.ts)
no
longer references LiteSimpleConsumerImpl, while
`build(): Promise<LiteSimpleConsumer>` is unchanged;
- tsc -p tsconfig.prod.json and -p tsconfig.test.json: 0 errors;
- eslint: 0 errors;
- non-integration suite: 97/97 pass.
---
nodejs/README.md | 61 ++++
nodejs/examples/LiteSimpleConsumerExample.ts | 128 +++++++
nodejs/src/client/BaseClient.ts | 6 +
nodejs/src/client/TelemetrySession.ts | 6 +
nodejs/src/consumer/Consumer.ts | 12 +-
nodejs/src/consumer/LitePushConsumerImpl.ts | 12 +-
nodejs/src/consumer/LiteSimpleConsumer.ts | 251 ++++++++++++++
nodejs/src/consumer/LiteSimpleConsumerImpl.ts | 221 ++++++++++++
nodejs/src/consumer/LiteSubscriptionManager.ts | 99 ++++--
nodejs/src/consumer/index.ts | 1 +
.../LiteSimpleConsumer.integration.test.ts | 201 +++++++++++
nodejs/test/consumer/LiteSimpleConsumer.test.ts | 374 +++++++++++++++++++++
12 files changed, 1332 insertions(+), 40 deletions(-)
diff --git a/nodejs/README.md b/nodejs/README.md
index e36a0417..9a4d051c 100644
--- a/nodejs/README.md
+++ b/nodejs/README.md
@@ -169,6 +169,67 @@ try {
}
```
+### Lite Topic
+
+Lite topics are light-weight subscriptions on top of one parent topic. A lite
consumer binds to the
+parent topic and (un)subscribes lite topics at runtime, which makes it cheap
to serve a large number
+of short-lived topics.
+
+Server prerequisites: broker with `enableLmq=true` and
`enableMultiDispatch=true`, a parent topic
+created with `message.type=LITE`, and a consumer group created with the
attribute
+`+lite.bind.topic=<parentTopic>`.
+
+Send with lite topic
+
+```ts
+import { Producer } from 'rocketmq-client-nodejs';
+
+const producer = new Producer({
+ endpoints: '127.0.0.1:8081',
+});
+await producer.startup();
+
+await producer.send({
+ topic: 'yourParentTopic',
+ liteTopic: 'lite-topic-1', // The lite topic the message belongs to
+ body: Buffer.from('This is a lite message'),
+});
+
+await producer.shutdown();
+```
+
+LiteSimpleConsumer (pull and ack explicitly)
+
+```ts
+import { LiteSimpleConsumerBuilder, OffsetOption } from
'rocketmq-client-nodejs';
+
+// Bind to the parent topic, the consumer group must be bound to it as well
+const consumer = await new LiteSimpleConsumerBuilder()
+ .setClientConfiguration({ endpoints: '127.0.0.1:8081' })
+ .setConsumerGroup('yourConsumerGroup')
+ .bindTopic('yourParentTopic')
+ .setAwaitDuration(5000)
+ .build();
+
+// Subscribe lite topics, with or without a consume-from offset
+await consumer.subscribeLite('lite-topic-1', OffsetOption.MIN_OFFSET);
+await consumer.subscribeLite('lite-topic-2');
+
+// Pull a batch and ack every message
+const messages = await consumer.receive(16, 15000);
+for (const message of messages) {
+ console.log(message.liteTopic, message.body.toString());
+ await consumer.ack(message);
+}
+
+// Release the subscription when it's no longer needed
+await consumer.unsubscribeLite('lite-topic-2');
+await consumer.close();
+```
+
+A runnable version lives in `examples/LiteSimpleConsumerExample.ts` (see also
+`examples/LiteProducerExample.ts` and `examples/LitePushConsumerExample.ts`).
+
## Current Progress
### Message Type
diff --git a/nodejs/examples/LiteSimpleConsumerExample.ts
b/nodejs/examples/LiteSimpleConsumerExample.ts
new file mode 100644
index 00000000..b7e44880
--- /dev/null
+++ b/nodejs/examples/LiteSimpleConsumerExample.ts
@@ -0,0 +1,128 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+/**
+ * LiteSimpleConsumer Example
+ *
+ * This example demonstrates how to use LiteSimpleConsumer to pull messages
from
+ * lite topics with explicit receive / ack control.
+ *
+ * Prerequisites:
+ * - RocketMQ proxy plus a broker with lite support enabled
+ * (broker.conf: enableLmq=true, enableMultiDispatch=true)
+ * - A parent topic created with message.type=LITE
+ * - A consumer group created with the attribute +lite.bind.topic=<parentTopic>
+ * - Messages published to the parent topic with the liteTopic property set,
+ * see LiteProducerExample.ts
+ */
+
+import { LiteSimpleConsumerBuilder, OffsetOption, type MessageView } from
'../src';
+import { endpoints, namespace, consumerGroup, sessionCredentials,
liteTopicConfig } from './ProducerSingleton';
+
+const BIND_TOPIC = liteTopicConfig.parentTopic;
+const INVISIBLE_DURATION = 15000; // 15s, the visibility timeout of received
messages
+
+async function main() {
+ console.log('========== LiteSimpleConsumer Example ==========\n');
+
+ // Build and start the lite simple consumer, it binds to one parent topic
only.
+ const consumer = await new LiteSimpleConsumerBuilder()
+ .setClientConfiguration({
+ endpoints,
+ namespace,
+ sessionCredentials,
+ })
+ .setConsumerGroup(consumerGroup)
+ .bindTopic(BIND_TOPIC) // Replace with your actual parent topic
+ .setAwaitDuration(5000) // Long-polling timeout of receive()
+ .build();
+
+ console.log(`✓ Consumer started, bindTopic=${BIND_TOPIC},
group=${consumerGroup}\n`);
+
+ let running = true;
+ process.on('SIGINT', () => {
+ console.log('\nReceived SIGINT, stopping...');
+ running = false;
+ });
+
+ try {
+ // Subscribe to lite topics. All of them must belong to the bound parent
topic.
+ console.log('Subscribing to lite topics...\n');
+
+ await consumer.subscribeLite('lite-topic-1', OffsetOption.MIN_OFFSET);
+ console.log('✓ Subscribed to lite-topic-1 (from the minimum offset)');
+
+ await consumer.subscribeLite('lite-topic-2');
+ console.log('✓ Subscribed to lite-topic-2 (from the latest offset)');
+
+ await consumer.subscribeLite('lite-topic-3', OffsetOption.ofTailN(100));
+ console.log('✓ Subscribed to lite-topic-3 (the last 100 messages)\n');
+
+ console.log('Current lite topic set:', [ ...consumer.getLiteTopicSet() ],
'\n');
+ console.log('Receiving messages. Press Ctrl+C to exit...\n');
+
+ // Pull messages by batch, then ack them one by one.
+ while (running) {
+ const messages: MessageView[] = await consumer.receive(16,
INVISIBLE_DURATION);
+ if (messages.length === 0) {
+ continue;
+ }
+
+ for (const message of messages) {
+ console.log('Received message:', {
+ messageId: message.messageId,
+ topic: message.topic,
+ liteTopic: message.liteTopic,
+ tag: message.tag,
+ keys: message.keys,
+ body: message.body.toString('utf-8'),
+ });
+
+ try {
+ // Ack as soon as the business logic succeeds, otherwise the message
+ // becomes visible again after the invisible duration expires.
+ await consumer.ack(message);
+ console.log(`✓ Acked message ${message.messageId}\n`);
+ } catch (error) {
+ console.error(`✗ Failed to ack message ${message.messageId}:`,
error);
+
+ // Not ready to process it yet? Extend the invisible duration instead
+ // of dropping the message.
+ await consumer.changeInvisibleDuration(message, INVISIBLE_DURATION);
+ }
+ }
+ }
+
+ // Release the subscriptions that are no longer needed, this frees the lite
+ // topic quota of the consumer group.
+ console.log('\nUnsubscribing from lite-topic-3...');
+ await consumer.unsubscribeLite('lite-topic-3');
+ console.log('✓ Unsubscribed, remaining lite topic set:', [
...consumer.getLiteTopicSet() ], '\n');
+ } catch (error) {
+ console.error('Error during consumption:', error);
+ throw error;
+ } finally {
+ console.log('Shutting down consumer...');
+ await consumer.close();
+ console.log('✓ Consumer closed successfully');
+ }
+}
+
+main().catch(error => {
+ console.error('Fatal error:', error);
+ process.exit(1);
+});
diff --git a/nodejs/src/client/BaseClient.ts b/nodejs/src/client/BaseClient.ts
index 7bd22802..acf694d5 100644
--- a/nodejs/src/client/BaseClient.ts
+++ b/nodejs/src/client/BaseClient.ts
@@ -30,6 +30,7 @@ import {
VerifyMessageCommand,
PrintThreadStackTraceCommand,
ReconnectEndpointsCommand,
+ NotifyUnsubscribeLiteCommand,
TelemetryCommand,
ThreadStackTrace,
HeartbeatRequest,
@@ -518,6 +519,11 @@ export abstract class BaseClient {
this.telemetry(endpoints, telemetryCommand);
}
+ onNotifyUnsubscribeLiteCommand(endpoints: Endpoints, command:
NotifyUnsubscribeLiteCommand) {
+ this.logger.warn('Ignore notify unsubscribe lite command from remote,
which is not expected, clientId=%s, endpoints=%s, command=%j',
+ this.clientId, endpoints.facade, command.toObject());
+ }
+
onPrintThreadStackTraceCommand(endpoints: Endpoints, command:
PrintThreadStackTraceCommand) {
const obj = command.toObject();
this.logger.warn('Ignore orphaned transaction recovery command from
remote, which is not expected, clientId=%s, command=%j',
diff --git a/nodejs/src/client/TelemetrySession.ts
b/nodejs/src/client/TelemetrySession.ts
index d2abdef4..d0fcf231 100644
--- a/nodejs/src/client/TelemetrySession.ts
+++ b/nodejs/src/client/TelemetrySession.ts
@@ -134,6 +134,12 @@ export class TelemetrySession {
this.#baseClient.onReconnectEndpointsCommand(endpoints,
command.getReconnectEndpointsCommand()!);
break;
}
+ case TelemetryCommand.CommandCase.NOTIFY_UNSUBSCRIBE_LITE_COMMAND: {
+ this.#logger.info('Receive notify unsubscribe lite command from
remote, endpoints=%s, clientId=%s',
+ endpoints, clientId);
+ this.#baseClient.onNotifyUnsubscribeLiteCommand(endpoints,
command.getNotifyUnsubscribeLiteCommand()!);
+ break;
+ }
default: {
const commandObj = command.toObject();
this.#logger.warn('Receive unrecognized command from remote,
endpoints=%s, commandCase=%j, command=%j, clientId=%s',
diff --git a/nodejs/src/consumer/Consumer.ts b/nodejs/src/consumer/Consumer.ts
index ecbfbbb0..23a4330a 100644
--- a/nodejs/src/consumer/Consumer.ts
+++ b/nodejs/src/consumer/Consumer.ts
@@ -104,9 +104,12 @@ export abstract class Consumer extends BaseClient {
const request = new AckMessageRequest()
.setGroup(createResource(this.consumerGroup))
.setTopic(createResource(messageView.topic));
- request.addEntries()
+ const entry = request.addEntries()
.setMessageId(messageView.messageId)
.setReceiptHandle(messageView.receiptHandle);
+ if (this.isLiteConsumer() && messageView.liteTopic) {
+ entry.setLiteTopic(messageView.liteTopic);
+ }
const res = await this.rpcClientManager.ackMessage(endpoints, request,
this.requestTimeout);
// FIXME: handle fail ack
const response = res.toObject();
@@ -121,6 +124,9 @@ export abstract class Consumer extends BaseClient {
.setReceiptHandle(messageView.receiptHandle)
.setInvisibleDuration(createDuration(invisibleDuration))
.setMessageId(messageView.messageId);
+ if (this.isLiteConsumer() && messageView.liteTopic) {
+ request.setLiteTopic(messageView.liteTopic);
+ }
const res = await
this.rpcClientManager.changeInvisibleDuration(messageView.endpoints, request,
this.requestTimeout);
const response = res.toObject();
@@ -158,10 +164,10 @@ export abstract class Consumer extends BaseClient {
/**
* Check if this is a lite consumer.
*
- * @return true if this is a LITE_PUSH_CONSUMER
+ * @return true if this is a LITE_PUSH_CONSUMER or LITE_SIMPLE_CONSUMER
*/
protected isLiteConsumer(): boolean {
const clientType = (this as any).getClientType?.();
- return clientType === ClientType.LITE_PUSH_CONSUMER;
+ return clientType === ClientType.LITE_PUSH_CONSUMER || clientType ===
ClientType.LITE_SIMPLE_CONSUMER;
}
}
diff --git a/nodejs/src/consumer/LitePushConsumerImpl.ts
b/nodejs/src/consumer/LitePushConsumerImpl.ts
index 37440b51..158a883d 100644
--- a/nodejs/src/consumer/LitePushConsumerImpl.ts
+++ b/nodejs/src/consumer/LitePushConsumerImpl.ts
@@ -218,9 +218,10 @@ export class LitePushConsumerImpl extends PushConsumer
implements LitePushConsum
* <p>This method is called when the server sends a notification to
unsubscribe
* from a lite topic, typically due to quota violations or administrative
actions.</p>
*
+ * @param _endpoints - The server endpoints
* @param command - The unsubscribe command from the server
*/
- onNotifyUnsubscribeLiteCommand(command: NotifyUnsubscribeLiteCommand) {
+ onNotifyUnsubscribeLiteCommand(_endpoints: Endpoints, command:
NotifyUnsubscribeLiteCommand) {
this.liteSubscriptionManager.onNotifyUnsubscribeLiteCommand(command);
}
@@ -278,4 +279,13 @@ export class LitePushConsumerImpl extends PushConsumer
implements LitePushConsum
getRequestTimeout(): number {
return this.requestTimeout;
}
+
+ /**
+ * Endpoints the lite subscription manager should sync to.
+ *
+ * @internal
+ */
+ getSyncEndpoints(): Endpoints[] {
+ return [ this.endpoints ];
+ }
}
diff --git a/nodejs/src/consumer/LiteSimpleConsumer.ts
b/nodejs/src/consumer/LiteSimpleConsumer.ts
new file mode 100644
index 00000000..cd0f9ff5
--- /dev/null
+++ b/nodejs/src/consumer/LiteSimpleConsumer.ts
@@ -0,0 +1,251 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import { BaseClientOptions } from '../client';
+import { MessageView } from '../message';
+import { OffsetOption } from './OffsetOption';
+import { LiteSimpleConsumerImpl } from './LiteSimpleConsumerImpl';
+
+/**
+ * LiteSimpleConsumer interface for consuming messages from lite topics with
+ * explicit receive/ack control.
+ *
+ * <p>Similar to SimpleConsumer, but bound to a single parent topic and able to
+ * (un)subscribe lite topics dynamically through the lite subscription sync
+ * protocol.</p>
+ */
+export interface LiteSimpleConsumer {
+ /**
+ * Get the load balancing group for the lite simple consumer.
+ *
+ * @return Consumer load balancing group
+ */
+ getConsumerGroup(): string;
+
+ /**
+ * Subscribe to a lite topic.
+ *
+ * <p>The subscribeLite() method initiates network requests and performs
quota
+ * verification, so it may fail. It's important to check the result of this
+ * call to ensure that the subscription was successfully added. Possible
+ * failure scenarios include:</p>
+ * <ul>
+ * <li>Network request errors, which can be retried.</li>
+ * <li>Quota verification failures, indicated by
LiteSubscriptionQuotaExceededException.
+ * In this case, evaluate whether the quota is insufficient and
promptly
+ * unsubscribe from unused subscriptions using unsubscribeLite() to
free
+ * up resources.</li>
+ * </ul>
+ *
+ * @param liteTopic - The name of the lite topic to subscribe
+ * @throws ClientException if an error occurs during subscription
+ */
+ subscribeLite(liteTopic: string): Promise<void>;
+
+ /**
+ * Subscribe to a lite topic with an offset option to specify the consume
from
+ * offset.
+ *
+ * @param liteTopic - The name of the lite topic to subscribe
+ * @param offsetOption - The consume from offset option
+ * @throws ClientException if an error occurs during subscription
+ */
+ subscribeLite(liteTopic: string, offsetOption: OffsetOption): Promise<void>;
+
+ /**
+ * Unsubscribe from a lite topic.
+ *
+ * @param liteTopic - The name of the lite topic to unsubscribe from
+ * @throws ClientException if an error occurs during unsubscription
+ */
+ unsubscribeLite(liteTopic: string): Promise<void>;
+
+ /**
+ * Get the lite topic immutable set.
+ *
+ * @return Lite topic immutable set
+ */
+ getLiteTopicSet(): Set<string>;
+
+ /**
+ * Fetch messages from the server synchronously.
+ *
+ * <p>This method returns immediately if there are messages available.
+ * Otherwise, it will await the passed timeout. If the timeout expires, an
+ * empty list will be returned.</p>
+ *
+ * @param maxMessageNum - Max message num of server returned
+ * @param invisibleDuration - Set the invisibleDuration of messages to return
+ * @return List of message views
+ */
+ receive(maxMessageNum?: number, invisibleDuration?: number):
Promise<MessageView[]>;
+
+ /**
+ * Ack the consumption of the message which is returned by receive().
+ *
+ * @param messageView - Message view with the receipt handle to ack
+ * @throws ClientException if an error occurs during ack
+ */
+ ack(messageView: MessageView): Promise<void>;
+
+ /**
+ * Change the invisible duration of a received message.
+ *
+ * @param messageView - Message view to change invisible duration
+ * @param invisibleDuration - New invisible duration
+ * @throws ClientException if an error occurs
+ */
+ changeInvisibleDuration(messageView: MessageView, invisibleDuration:
number): Promise<void>;
+
+ /**
+ * Close the consumer and release all related resources.
+ *
+ * <p>Once the consumer is closed, <strong>it could not be started once
+ * again.</strong></p>
+ */
+ close(): Promise<void>;
+}
+
+export interface LiteSimpleConsumerOptions extends BaseClientOptions {
+ consumerGroup: string;
+ /**
+ * The parent topic the lite consumer binds to. All lite topics subscribed by
+ * this consumer must belong to this parent topic.
+ */
+ bindTopic: string;
+ /**
+ * set await duration for long-polling, default is 30000ms
+ */
+ awaitDuration?: number;
+ /**
+ * max retry attempts for temporary errors (e.g., internal server error),
default is 3
+ */
+ maxRetryAttempts?: number;
+}
+
+const CONSUMER_GROUP_PATTERN = /^[a-zA-Z0-9_-]+$/;
+
+/**
+ * LiteSimpleConsumer builder class.
+ *
+ * <p>This class provides a fluent API for configuring and creating lite simple
+ * consumers with explicit receive/ack control over lite topics.</p>
+ */
+export class LiteSimpleConsumerBuilder {
+ private options: Partial<LiteSimpleConsumerOptions> = {};
+
+ /**
+ * Set the bind topic for the lite simple consumer.
+ *
+ * @param bindTopic - The parent topic to bind
+ * @return This builder instance
+ * @throws Error if bindTopic is blank
+ */
+ bindTopic(bindTopic: string): LiteSimpleConsumerBuilder {
+ if (!bindTopic || bindTopic.trim().length === 0) {
+ throw new Error('bindTopic should not be blank');
+ }
+ this.options.bindTopic = bindTopic;
+ return this;
+ }
+
+ /**
+ * Set the client configuration.
+ *
+ * @param options - Client configuration options
+ * @return This builder instance
+ * @throws Error if options is null/undefined
+ */
+ setClientConfiguration(options: BaseClientOptions):
LiteSimpleConsumerBuilder {
+ if (!options) {
+ throw new Error('clientConfiguration should not be null');
+ }
+ Object.assign(this.options, options);
+ return this;
+ }
+
+ /**
+ * Set the consumer group.
+ *
+ * @param consumerGroup - Consumer group name
+ * @return This builder instance
+ * @throws Error if consumerGroup is null or doesn't match the pattern
+ */
+ setConsumerGroup(consumerGroup: string): LiteSimpleConsumerBuilder {
+ if (!consumerGroup) {
+ throw new Error('consumerGroup should not be null');
+ }
+ if (!CONSUMER_GROUP_PATTERN.test(consumerGroup)) {
+ throw new Error(`consumerGroup does not match the pattern
${CONSUMER_GROUP_PATTERN.source}`);
+ }
+ this.options.consumerGroup = consumerGroup;
+ return this;
+ }
+
+ /**
+ * Set the await duration for long-polling receive requests.
+ *
+ * @param awaitDuration - Maximum time to block when no message is available
+ * @return This builder instance
+ * @throws Error if awaitDuration is not positive
+ */
+ setAwaitDuration(awaitDuration: number): LiteSimpleConsumerBuilder {
+ if (awaitDuration <= 0) {
+ throw new Error('awaitDuration should be positive');
+ }
+ this.options.awaitDuration = awaitDuration;
+ return this;
+ }
+
+ /**
+ * Finalize the build of LiteSimpleConsumer and start.
+ *
+ * <p>This method will block until the lite simple consumer starts
+ * successfully.</p>
+ *
+ * @return Promise resolving to started LiteSimpleConsumer instance
+ * @throws Error if required parameters are not set
+ */
+ async build(): Promise<LiteSimpleConsumer> {
+ if (!this.options.endpoints) {
+ throw new Error('clientConfiguration has not been set yet');
+ }
+ if (!this.options.consumerGroup) {
+ throw new Error('consumerGroup has not been set yet');
+ }
+ if (!this.options.bindTopic) {
+ throw new Error('bindTopic has not been set yet');
+ }
+
+ const options: LiteSimpleConsumerOptions = {
+ endpoints: this.options.endpoints,
+ namespace: this.options.namespace ?? '',
+ consumerGroup: this.options.consumerGroup!,
+ bindTopic: this.options.bindTopic,
+ awaitDuration: this.options.awaitDuration,
+ maxRetryAttempts: this.options.maxRetryAttempts,
+ sslEnabled: this.options.sslEnabled,
+ sessionCredentials: this.options.sessionCredentials,
+ requestTimeout: this.options.requestTimeout,
+ logger: this.options.logger,
+ };
+
+ const liteSimpleConsumer = new LiteSimpleConsumerImpl(options);
+ await liteSimpleConsumer.startup();
+ return liteSimpleConsumer;
+ }
+}
diff --git a/nodejs/src/consumer/LiteSimpleConsumerImpl.ts
b/nodejs/src/consumer/LiteSimpleConsumerImpl.ts
new file mode 100644
index 00000000..531e7907
--- /dev/null
+++ b/nodejs/src/consumer/LiteSimpleConsumerImpl.ts
@@ -0,0 +1,221 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import { ClientType, Permission, Settings as SettingsPB } from
'../../proto/apache/rocketmq/v2/definition_pb';
+import { NotifyUnsubscribeLiteCommand } from
'../../proto/apache/rocketmq/v2/service_pb';
+import { Endpoints, TopicRouteData } from '../route';
+import { Resource } from '../route/Resource';
+import { MASTER_BROKER_ID } from '../util';
+import { ILogger } from '../client/Logger';
+import { RpcClientManager } from '../client/RpcClientManager';
+import { SimpleConsumer, SimpleConsumerOptions } from './SimpleConsumer';
+import { FilterExpression } from './FilterExpression';
+import { LiteSubscriptionManager, LiteSubscriptionHost } from
'./LiteSubscriptionManager';
+import { OffsetOption } from './OffsetOption';
+import { LiteSimpleConsumer, LiteSimpleConsumerOptions } from
'./LiteSimpleConsumer';
+
+/**
+ * Keep only the first readable master queue of the route. Lite consumers talk
+ * to brokers only through the bound parent topic, so a single queue is enough
+ * and it avoids useless route bookkeeping for the other queues.
+ */
+export function firstReadableMasterQueue(topicRouteData: TopicRouteData):
TopicRouteData {
+ const readable = topicRouteData.messageQueues.find(
+ mq => mq.broker.id === MASTER_BROKER_ID &&
+ (mq.permission === Permission.READ || mq.permission ===
Permission.READ_WRITE),
+ );
+ return new TopicRouteData(readable ? [ readable.toProtobuf() ] : []);
+}
+
+/**
+ * Implementation of LiteSimpleConsumer.
+ *
+ * <p>LiteSimpleConsumer extends SimpleConsumer to provide explicit receive/ack
+ * control over lite topics. It binds to a single parent topic and manages lite
+ * topic subscriptions dynamically via the lite subscription sync protocol.</p>
+ */
+export class LiteSimpleConsumerImpl extends SimpleConsumer
+ implements LiteSimpleConsumer, LiteSubscriptionHost {
+ readonly #bindTopic: Resource;
+ readonly #liteSubscriptionManager: LiteSubscriptionManager;
+
+ constructor(options: LiteSimpleConsumerOptions) {
+ if (!options.bindTopic || options.bindTopic.trim().length === 0) {
+ throw new TypeError('bindTopic should not be blank');
+ }
+ // Default subscription: (bindTopic, *) for code reuse.
+ const subscriptions = new Map<string, FilterExpression | string>()
+ .set(options.bindTopic, FilterExpression.SUB_ALL);
+
+ super({
+ ...options,
+ subscriptions,
+ } as SimpleConsumerOptions);
+
+ this.#bindTopic = new Resource(options.namespace, options.bindTopic);
+ const groupResource = new Resource(options.namespace,
options.consumerGroup);
+ this.#liteSubscriptionManager = new LiteSubscriptionManager(this,
this.#bindTopic, groupResource);
+ }
+
+ /**
+ * Get the client type.
+ *
+ * @return The client type identifier for lite simple consumer
+ */
+ protected getClientType(): ClientType {
+ return ClientType.LITE_SIMPLE_CONSUMER;
+ }
+
+ protected onTopicRouteDataUpdate(topic: string, topicRouteData:
TopicRouteData) {
+ super.onTopicRouteDataUpdate(topic,
firstReadableMasterQueue(topicRouteData));
+ }
+
+ /**
+ * Start up the consumer.
+ *
+ * <p>This method initializes the consumer and starts the lite subscription
+ * manager. It must be called before the consumer can receive messages.</p>
+ */
+ async startup() {
+ await super.startup();
+ this.#liteSubscriptionManager.startUp();
+ }
+
+ /**
+ * Shutdown the consumer.
+ *
+ * <p>This method gracefully shuts down the consumer, releasing all resources
+ * and stopping the lite subscription manager.</p>
+ */
+ async shutdown() {
+ this.#liteSubscriptionManager.shutdown();
+ await super.shutdown();
+ }
+
+ async subscribeLite(liteTopic: string): Promise<void>;
+
+ /**
+ * Subscribe to a lite topic with an offset option to specify the consume
+ * from offset.
+ *
+ * @param liteTopic - The name of the lite topic to subscribe
+ * @param offsetOption - The consume from offset option
+ */
+ async subscribeLite(liteTopic: string, offsetOption: OffsetOption):
Promise<void>;
+
+ async subscribeLite(liteTopic: string, offsetOption?: OffsetOption):
Promise<void> {
+ if (!liteTopic || liteTopic.trim().length === 0) {
+ throw new Error('liteTopic should not be blank');
+ }
+ await this.#liteSubscriptionManager.subscribeLite(liteTopic, offsetOption
?? null);
+ }
+
+ /**
+ * Unsubscribe from a lite topic.
+ *
+ * @param liteTopic - The name of the lite topic to unsubscribe from
+ */
+ async unsubscribeLite(liteTopic: string): Promise<void> {
+ if (!liteTopic || liteTopic.trim().length === 0) {
+ throw new Error('liteTopic should not be blank');
+ }
+ await this.#liteSubscriptionManager.unsubscribeLite(liteTopic);
+ }
+
+ /**
+ * Get the lite topic set.
+ *
+ * @return Set of currently subscribed lite topic names
+ */
+ getLiteTopicSet(): Set<string> {
+ return this.#liteSubscriptionManager.getLiteTopicSet();
+ }
+
+ /**
+ * Get the load balancing group for the consumer.
+ *
+ * @return Consumer group name
+ */
+ getConsumerGroup(): string {
+ return this.#liteSubscriptionManager.getConsumerGroupName();
+ }
+
+ /**
+ * Handle notify unsubscribe lite command from server.
+ *
+ * @param _endpoints - The server endpoints
+ * @param command - The unsubscribe command from the server
+ */
+ onNotifyUnsubscribeLiteCommand(_endpoints: Endpoints, command:
NotifyUnsubscribeLiteCommand) {
+ this.#liteSubscriptionManager.onNotifyUnsubscribeLiteCommand(command);
+ }
+
+ /**
+ * Handle settings command from server.
+ *
+ * @param endpoints - The server endpoints
+ * @param settings - The settings configuration
+ */
+ onSettingsCommand(endpoints: Endpoints, settings: SettingsPB) {
+ super.onSettingsCommand(endpoints, settings);
+ this.#liteSubscriptionManager.sync(settings);
+ }
+
+ /**
+ * Close the consumer.
+ */
+ async close(): Promise<void> {
+ await this.shutdown();
+ }
+
+ /**
+ * Get the logger.
+ *
+ * @internal
+ */
+ getLogger(): ILogger {
+ return this.logger;
+ }
+
+ /**
+ * Get the RPC client manager (protected access for internal use).
+ *
+ * @internal
+ */
+ getRpcClientManager(): RpcClientManager {
+ return this.rpcClientManager;
+ }
+
+ /**
+ * Get the request timeout (protected access for internal use).
+ *
+ * @internal
+ */
+ getRequestTimeout(): number {
+ return this.requestTimeout;
+ }
+
+ /**
+ * Endpoints the lite subscription manager should sync to: every endpoint
+ * present in the consumed routes.
+ *
+ * @internal
+ */
+ getSyncEndpoints(): Endpoints[] {
+ return this.getTotalRouteEndpoints();
+ }
+}
diff --git a/nodejs/src/consumer/LiteSubscriptionManager.ts
b/nodejs/src/consumer/LiteSubscriptionManager.ts
index 2d89a13b..179bad71 100644
--- a/nodejs/src/consumer/LiteSubscriptionManager.ts
+++ b/nodejs/src/consumer/LiteSubscriptionManager.ts
@@ -21,15 +21,37 @@ import {
SyncLiteSubscriptionRequest,
SyncLiteSubscriptionResponse,
} from '../../proto/apache/rocketmq/v2/service_pb';
+import { Endpoints } from '../route';
+import { ILogger } from '../client/Logger';
+import { RpcClientManager } from '../client/RpcClientManager';
import { ClientException } from '../exception';
-import { Resource } from '../route';
-import { LitePushConsumerImpl } from './LitePushConsumerImpl';
+import { Resource } from '../route/Resource';
import { OffsetOption } from './OffsetOption';
const SYNC_LITE_SUBSCRIPTION_INTERVAL = 30000; // 30 seconds
/**
- * Manages lite topic subscriptions for LitePushConsumer.
+ * Minimal host contract required by LiteSubscriptionManager.
+ *
+ * Implemented by LitePushConsumerImpl and LiteSimpleConsumerImpl so the same
+ * manager can keep lite subscriptions in sync for both consumer flavors.
+ */
+export interface LiteSubscriptionHost {
+ clientId: string;
+ isRunning(): boolean;
+ getLogger(): ILogger;
+ getRpcClientManager(): RpcClientManager;
+ getRequestTimeout(): number;
+ /**
+ * Endpoints the lite subscription should be synced to. Lite subscriptions
+ * must reach every route endpoint, not only the endpoint the client was
+ * configured with.
+ */
+ getSyncEndpoints(): Endpoints[];
+}
+
+/**
+ * Manages lite topic subscriptions for lite consumers.
*
* <p>LiteSubscriptionManager handles:
* - Maintaining the set of subscribed lite topics
@@ -38,7 +60,7 @@ const SYNC_LITE_SUBSCRIPTION_INTERVAL = 30000; // 30 seconds
* - Handling unsubscribe commands from server</p>
*/
export class LiteSubscriptionManager {
- private readonly consumerImpl: LitePushConsumerImpl;
+ private readonly host: LiteSubscriptionHost;
private readonly bindTopic: Resource;
private readonly group: Resource;
private readonly liteTopicSet = new Set<string>();
@@ -47,11 +69,11 @@ export class LiteSubscriptionManager {
private syncTimer?: NodeJS.Timeout;
constructor(
- consumerImpl: LitePushConsumerImpl,
+ host: LiteSubscriptionHost,
bindTopic: Resource,
group: Resource,
) {
- this.consumerImpl = consumerImpl;
+ this.host = host;
this.bindTopic = bindTopic;
this.group = group;
this.liteSubscriptionQuota = 100; // Default quota
@@ -129,7 +151,7 @@ export class LiteSubscriptionManager {
offsetOption?: OffsetOption | null,
): Promise<void> {
// Check if consumer is running
- if (!this.consumerImpl.isRunning()) {
+ if (!this.host.isRunning()) {
throw new ClientException(500, 'Consumer is not running');
}
@@ -152,15 +174,15 @@ export class LiteSubscriptionManager {
);
this.liteTopicSet.add(liteTopic);
- this.consumerImpl.getLogger().info(
+ this.host.getLogger().info(
'SubscribeLite %s, topic=%s, group=%s, clientId=%s',
liteTopic,
this.getBindTopicName(),
this.getConsumerGroupName(),
- this.consumerImpl.clientId,
+ this.host.clientId,
);
} catch (error) {
- this.consumerImpl.getLogger().error(
+ this.host.getLogger().error(
'Failed to subscribeLite %s, error=%s',
liteTopic,
error,
@@ -174,7 +196,7 @@ export class LiteSubscriptionManager {
*/
public async unsubscribeLite(liteTopic: string): Promise<void> {
// Check if consumer is running
- if (!this.consumerImpl.isRunning()) {
+ if (!this.host.isRunning()) {
throw new ClientException(500, 'Consumer is not running');
}
@@ -191,15 +213,15 @@ export class LiteSubscriptionManager {
);
this.liteTopicSet.delete(liteTopic);
- this.consumerImpl.getLogger().info(
+ this.host.getLogger().info(
'UnsubscribeLite %s, topic=%s, group=%s, clientId=%s',
liteTopic,
this.getBindTopicName(),
this.getConsumerGroupName(),
- this.consumerImpl.clientId,
+ this.host.clientId,
);
} catch (error) {
- this.consumerImpl.getLogger().error(
+ this.host.getLogger().error(
'Failed to unsubscribeLite %s, error=%s',
liteTopic,
error,
@@ -220,23 +242,23 @@ export class LiteSubscriptionManager {
null,
);
} catch (error) {
- this.consumerImpl.getLogger().error(
+ this.host.getLogger().error(
'Schedule syncAllLiteSubscription error, clientId=%s, error=%s',
- this.consumerImpl.clientId,
+ this.host.clientId,
error,
);
}
}
/**
- * Sync lite subscription with server.
+ * Sync lite subscription with every route endpoint.
*/
private async syncLiteSubscription(
action: LiteSubscriptionAction,
liteTopics: string[],
offsetOption: OffsetOption | null,
): Promise<void> {
- const logger = this.consumerImpl.getLogger();
+ const logger = this.host.getLogger();
if (logger.debug) {
logger.debug(
'SyncLiteSubscription: action=%s, liteTopics=[%s], offsetOption=%s',
@@ -258,29 +280,34 @@ export class LiteSubscriptionManager {
request.setOffsetOption(offsetOption.toProtobuf());
}
- try {
- // Call RPC client using public methods from LitePushConsumerImpl
- const response: SyncLiteSubscriptionResponse = await
this.consumerImpl.getRpcClientManager().syncLiteSubscription(
- this.consumerImpl.getEndpoints(),
- request,
- this.consumerImpl.getRequestTimeout(),
- );
+ const endpointsList = this.host.getSyncEndpoints();
+ if (endpointsList.length === 0) {
+ throw new ClientException(500, 'No endpoints available to sync lite
subscription');
+ }
- // Handle response status
- const status = response.getStatus();
- if (status && status.getCode() !== Code.OK) {
- throw new ClientException(
- status.getCode(),
- `Failed to sync lite subscription: ${status.getMessage()}`,
- );
- }
+ try {
+ // The lite subscription must reach every route endpoint; the sync fails
+ // if any endpoint rejects it.
+ await Promise.all(endpointsList.map(async endpoints => {
+ const response: SyncLiteSubscriptionResponse = await
this.host.getRpcClientManager()
+ .syncLiteSubscription(endpoints, request,
this.host.getRequestTimeout());
+
+ // Handle response status
+ const status = response.getStatus();
+ if (status && status.getCode() !== Code.OK) {
+ throw new ClientException(
+ status.getCode(),
+ `Failed to sync lite subscription: ${status.getMessage()}`,
+ );
+ }
+ }));
if (logger.info) {
logger.info(
'SyncLiteSubscription success: action=%s, liteTopics=[%s],
clientId=%s',
LiteSubscriptionAction[action],
liteTopics.join(', '),
- this.consumerImpl.clientId,
+ this.host.clientId,
);
}
} catch (error) {
@@ -289,7 +316,7 @@ export class LiteSubscriptionManager {
'SyncLiteSubscription failed: action=%s, liteTopics=[%s],
clientId=%s, error=%s',
LiteSubscriptionAction[action],
liteTopics.join(', '),
- this.consumerImpl.clientId,
+ this.host.clientId,
error,
);
}
@@ -302,7 +329,7 @@ export class LiteSubscriptionManager {
*/
public onNotifyUnsubscribeLiteCommand(command: NotifyUnsubscribeLiteCommand)
{
const liteTopic = command.getLiteTopic();
- this.consumerImpl.getLogger().info(
+ this.host.getLogger().info(
'Notify unsubscribe lite: liteTopic=%s, group=%s, bindTopic=%s',
liteTopic,
this.getConsumerGroupName(),
diff --git a/nodejs/src/consumer/index.ts b/nodejs/src/consumer/index.ts
index ff7a14ff..0d617401 100644
--- a/nodejs/src/consumer/index.ts
+++ b/nodejs/src/consumer/index.ts
@@ -34,6 +34,7 @@ export * from './LiteStandardConsumeService';
export * from './ProcessQueue';
export * from './PushConsumer';
export * from './LitePushConsumer';
+export * from './LiteSimpleConsumer';
export * from './OffsetOption';
export { LitePushConsumerImpl } from './LitePushConsumerImpl';
export { LiteSubscriptionManager } from './LiteSubscriptionManager';
diff --git a/nodejs/test/consumer/LiteSimpleConsumer.integration.test.ts
b/nodejs/test/consumer/LiteSimpleConsumer.integration.test.ts
new file mode 100644
index 00000000..5f6d0bb3
--- /dev/null
+++ b/nodejs/test/consumer/LiteSimpleConsumer.integration.test.ts
@@ -0,0 +1,201 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+/**
+ * Real-scenario integration tests for LiteSimpleConsumer.
+ *
+ * Requires a running proxy at the configured endpoints (default
localhost:8081)
+ * with the following prerequisites:
+ * - the bind topic exists and is created with the LITE message type;
+ * - the consumer group exists and carries the attribute
+ * lite.bind.topic=<bindTopic> (lite group).
+ *
+ * The local development broker must run with enableLmq=true and
+ * enableMultiDispatch=true so lite messages reach the LMQ queues.
+ */
+
+import { describe, it, afterEach } from 'node:test';
+import * as assert from 'node:assert';
+import { Producer } from '../../src';
+import { LiteSimpleConsumerImpl } from
'../../src/consumer/LiteSimpleConsumerImpl';
+import { OffsetOption } from '../../src/consumer/OffsetOption';
+import { endpoints, namespace } from '../helper';
+
+const BIND_TOPIC = process.env.ROCKETMQ_NODEJS_CLIENT_LITE_TOPIC ??
'lite-parent-topic';
+const CONSUMER_GROUP = process.env.ROCKETMQ_NODEJS_CLIENT_LITE_GROUP ??
'nodejs-lite-unittest-group';
+
+describe('test/consumer/LiteSimpleConsumer.integration.test.ts', () => {
+ let producer: Producer | null = null;
+ let consumer: LiteSimpleConsumerImpl | null = null;
+
+ afterEach(async () => {
+ if (consumer) {
+ await consumer.shutdown().catch(() => undefined);
+ consumer = null;
+ }
+ if (producer) {
+ await producer.shutdown().catch(() => undefined);
+ producer = null;
+ }
+ });
+
+ async function startClientPair() {
+ producer = new Producer({
+ endpoints,
+ namespace,
+ topic: BIND_TOPIC,
+ });
+ await producer.startup();
+
+ consumer = new LiteSimpleConsumerImpl({
+ endpoints,
+ namespace,
+ consumerGroup: CONSUMER_GROUP,
+ bindTopic: BIND_TOPIC,
+ awaitDuration: 5000,
+ });
+ await consumer.startup();
+ assert.strictEqual(consumer.getConsumerGroup(), CONSUMER_GROUP);
+ assert.strictEqual(consumer.getLiteTopicSet().size, 0);
+ }
+
+ /**
+ * Send a lite message with retry. Right after the LITE topic is created, the
+ * proxy may serve a cached topic route without the message.type attribute,
+ * which makes the producer reject the send client-side. Refreshing the route
+ * cache and retrying bridges that propagation window.
+ */
+ async function sendLiteWithRetry(options: Parameters<Producer['send']>[0]) {
+ let lastErr: unknown = null;
+ const deadline = Date.now() + 30000;
+ while (Date.now() < deadline) {
+ try {
+ return await producer!.send(options);
+ } catch (err) {
+ const message = err instanceof Error ? err.message : String(err);
+ if (!message.includes('message type not match')) {
+ throw err;
+ }
+ lastErr = err;
+ await (producer as unknown as { updateRoutes(): Promise<void>
}).updateRoutes();
+ await new Promise(resolve => setTimeout(resolve, 500));
+ }
+ }
+ throw lastErr ?? new Error('sendLiteWithRetry deadline exceeded');
+ }
+
+ it('should send and receive lite messages end-to-end', async () => {
+ await startClientPair();
+
+ const liteTopic = `lite-topic-it-${Date.now()}`;
+ await consumer!.subscribeLite(liteTopic, OffsetOption.MIN_OFFSET);
+ assert.ok(consumer!.getLiteTopicSet().has(liteTopic));
+
+ const sentIds: string[] = [];
+ for (let i = 0; i < 5; i++) {
+ const receipt = await sendLiteWithRetry({
+ topic: BIND_TOPIC,
+ liteTopic,
+ keys: [ `lite-it-key-${i}` ],
+ body: Buffer.from(`lite-it-body-${i}`),
+ });
+ assert.ok(receipt.messageId);
+ sentIds.push(receipt.messageId);
+ }
+ assert.strictEqual(sentIds.length, 5);
+
+ // Receive with a fixed invisible duration and ack every message.
+ const receivedIds: string[] = [];
+ const receivedBodies: string[] = [];
+ const deadline = Date.now() + 30000;
+ while (receivedIds.length < sentIds.length && Date.now() < deadline) {
+ const views = await consumer!.receive(10, 15000);
+ for (const view of views) {
+ receivedIds.push(view.messageId as string);
+ receivedBodies.push(Buffer.from(view.body as Uint8Array).toString());
+ await consumer!.ack(view);
+ }
+ }
+
+ assert.strictEqual(receivedIds.length, sentIds.length);
+ assert.deepStrictEqual(receivedIds.sort(), sentIds.sort());
+ for (let i = 0; i < 5; i++) {
+ assert.ok(receivedBodies.includes(`lite-it-body-${i}`));
+ }
+ });
+
+ it('should deliver messages sent after subscribeLite without offset option',
async () => {
+ await startClientPair();
+
+ const liteTopic = `lite-topic-live-${Date.now()}`;
+ await consumer!.subscribeLite(liteTopic);
+ assert.ok(consumer!.getLiteTopicSet().has(liteTopic));
+
+ const sentIds: string[] = [];
+ for (let i = 0; i < 3; i++) {
+ const receipt = await sendLiteWithRetry({
+ topic: BIND_TOPIC,
+ liteTopic,
+ keys: [ `lite-live-key-${i}` ],
+ body: Buffer.from(`lite-live-body-${i}`),
+ });
+ sentIds.push(receipt.messageId as string);
+ }
+
+ const receivedIds: string[] = [];
+ const deadline = Date.now() + 30000;
+ while (receivedIds.length < sentIds.length && Date.now() < deadline) {
+ const views = await consumer!.receive(10, 15000);
+ for (const view of views) {
+ receivedIds.push(view.messageId as string);
+ await consumer!.ack(view);
+ }
+ }
+ assert.strictEqual(receivedIds.length, sentIds.length);
+ assert.deepStrictEqual(receivedIds.sort(), sentIds.sort());
+ });
+
+ it('should stop delivering after unsubscribeLite', async () => {
+ await startClientPair();
+
+ const liteTopic = `lite-topic-unsub-${Date.now()}`;
+ await consumer!.subscribeLite(liteTopic);
+ await consumer!.unsubscribeLite(liteTopic);
+ assert.strictEqual(consumer!.getLiteTopicSet().size, 0);
+
+ // Messages for the unsubscribed lite topic must not be delivered.
+ await producer!.send({
+ topic: BIND_TOPIC,
+ liteTopic,
+ keys: [ 'lite-unsub-key' ],
+ body: Buffer.from('lite-unsub-body'),
+ });
+
+ const views = await consumer!.receive(10, 15000);
+ assert.strictEqual(views.length, 0);
+ });
+
+ it('should reject blank lite topic names', async () => {
+ await startClientPair();
+ await assert.rejects(async () => {
+ await consumer!.subscribeLite(' ');
+ }, /liteTopic should not be blank/);
+ await assert.rejects(async () => {
+ await consumer!.unsubscribeLite('');
+ }, /liteTopic should not be blank/);
+ });
+});
diff --git a/nodejs/test/consumer/LiteSimpleConsumer.test.ts
b/nodejs/test/consumer/LiteSimpleConsumer.test.ts
new file mode 100644
index 00000000..116bee05
--- /dev/null
+++ b/nodejs/test/consumer/LiteSimpleConsumer.test.ts
@@ -0,0 +1,374 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+/**
+ * Offline tests for LiteSimpleConsumer (builder, impl, route filter) and
+ * LiteSubscriptionManager. No broker required.
+ */
+
+import { describe, it } from 'node:test';
+import * as assert from 'node:assert';
+import {
+ Broker as BrokerPB,
+ ClientType,
+ LiteSubscriptionAction,
+ MessageQueue as MessageQueuePB,
+ Permission,
+ Settings as SettingsPB,
+ Status,
+ Code,
+ Subscription as SubscriptionPB,
+} from '../../proto/apache/rocketmq/v2/definition_pb';
+import {
+ NotifyUnsubscribeLiteCommand,
+ SyncLiteSubscriptionResponse,
+} from '../../proto/apache/rocketmq/v2/service_pb';
+import { createResource } from '../../src/util';
+import { Endpoints, Resource, TopicRouteData } from '../../src/route';
+import { LiteSimpleConsumerBuilder } from
'../../src/consumer/LiteSimpleConsumer';
+import {
+ LiteSimpleConsumerImpl,
+ firstReadableMasterQueue,
+} from '../../src/consumer/LiteSimpleConsumerImpl';
+import {
+ LiteSubscriptionManager,
+ LiteSubscriptionHost,
+} from '../../src/consumer/LiteSubscriptionManager';
+import { OffsetOption } from '../../src/consumer/OffsetOption';
+
+const ENDPOINTS = '127.0.0.1:8081';
+
+function buildProtoQueue(id: number, brokerId: number, permission:
Permission): MessageQueuePB {
+ const broker = new BrokerPB()
+ .setName(`broker-${brokerId}`)
+ .setId(brokerId)
+ .setEndpoints(new Endpoints('127.0.0.1:10911').toProtobuf());
+ return new MessageQueuePB()
+ .setId(id)
+ .setTopic(createResource('lite-parent-topic'))
+ .setBroker(broker)
+ .setPermission(permission);
+}
+
+interface SyncRecord {
+ endpoints: Endpoints;
+ action: LiteSubscriptionAction;
+ liteTopics: string[];
+}
+
+class FakeHost implements LiteSubscriptionHost {
+ clientId = 'fake-client-id';
+ running = true;
+ quota: number | null = null;
+ maxLiteTopicSize: number | null = null;
+ readonly endpointsList = [ new Endpoints('127.0.0.1:8081'), new
Endpoints('127.0.0.1:8082') ];
+ readonly syncRecords: SyncRecord[] = [];
+ readonly logger = {
+ info: () => undefined,
+ warn: () => undefined,
+ error: () => undefined,
+ debug: () => undefined,
+ };
+
+ isRunning(): boolean {
+ return this.running;
+ }
+
+ getLogger() {
+ return this.logger;
+ }
+
+ // eslint-disable-next-line @typescript-eslint/no-explicit-any
+ getRpcClientManager(): any {
+ return {
+ syncLiteSubscription: async (endpoints: Endpoints, request: any) => {
+ this.syncRecords.push({
+ endpoints,
+ action: request.getAction(),
+ liteTopics: request.getLiteTopicSetList(),
+ });
+ return new SyncLiteSubscriptionResponse().setStatus(new
Status().setCode(Code.OK));
+ },
+ };
+ }
+
+ getRequestTimeout(): number {
+ return 3000;
+ }
+
+ getSyncEndpoints(): Endpoints[] {
+ return this.endpointsList;
+ }
+
+ resetRecords() {
+ this.syncRecords.length = 0;
+ }
+}
+
+describe('test/consumer/LiteSimpleConsumer.test.ts', () => {
+
+ describe('LiteSimpleConsumerBuilder', () => {
+ it('should chain bindTopic and validate blank topic', () => {
+ const builder = new LiteSimpleConsumerBuilder();
+ assert.strictEqual(builder.bindTopic('test-bind-topic'), builder);
+ assert.throws(() => builder.bindTopic(''), /bindTopic should not be
blank/);
+ assert.throws(() => builder.bindTopic(' '), /bindTopic should not be
blank/);
+ });
+
+ it('should validate consumer group', () => {
+ const builder = new LiteSimpleConsumerBuilder();
+ assert.throws(() => builder.setConsumerGroup(null as any),
/consumerGroup should not be null/);
+ assert.throws(() => builder.setConsumerGroup('invalid group!'), /does
not match the pattern/);
+ assert.strictEqual(builder.setConsumerGroup('valid-group_1'), builder);
+ });
+
+ it('should validate await duration', () => {
+ const builder = new LiteSimpleConsumerBuilder();
+ assert.throws(() => builder.setAwaitDuration(0), /awaitDuration should
be positive/);
+ assert.throws(() => builder.setAwaitDuration(-1), /awaitDuration should
be positive/);
+ assert.strictEqual(builder.setAwaitDuration(5000), builder);
+ });
+
+ it('should fail to build without required options', async () => {
+ await assert.rejects(async () => {
+ await new LiteSimpleConsumerBuilder().build();
+ }, /clientConfiguration has not been set yet/);
+
+ await assert.rejects(async () => {
+ await new LiteSimpleConsumerBuilder().setClientConfiguration({
endpoints: ENDPOINTS, namespace: '' }).build();
+ }, /consumerGroup has not been set yet/);
+
+ await assert.rejects(async () => {
+ await new LiteSimpleConsumerBuilder()
+ .setClientConfiguration({ endpoints: ENDPOINTS, namespace: '' })
+ .setConsumerGroup('valid-group')
+ .build();
+ }, /bindTopic has not been set yet/);
+ });
+ });
+
+ describe('LiteSimpleConsumerImpl constructor', () => {
+ const baseOptions = {
+ endpoints: ENDPOINTS,
+ namespace: '',
+ consumerGroup: 'lite-unittest-group',
+ bindTopic: 'lite-parent-topic',
+ };
+
+ it('should reject blank bindTopic', () => {
+ assert.throws(() => {
+ new LiteSimpleConsumerImpl({ ...baseOptions, bindTopic: ' ' });
+ }, /bindTopic should not be blank/);
+ });
+
+ it('should use LITE_SIMPLE_CONSUMER client type', () => {
+ const impl = new LiteSimpleConsumerImpl(baseOptions);
+ assert.strictEqual((impl as any).getClientType(),
ClientType.LITE_SIMPLE_CONSUMER);
+ });
+
+ it('should subscribe the bind topic with SUB_ALL by default', () => {
+ const impl = new LiteSimpleConsumerImpl(baseOptions);
+ const settings = (impl as any).getSettings();
+ const protobuf = settings.toProtobuf();
+ const subscription = protobuf.getSubscription()!;
+ assert.strictEqual(subscription.getGroup()!.getName(),
'lite-unittest-group');
+ const entries = subscription.getSubscriptionsList();
+ assert.strictEqual(entries.length, 1);
+ assert.strictEqual(entries[0].getTopic()!.getName(),
'lite-parent-topic');
+ assert.strictEqual(entries[0].getExpression()!.getExpression(), '*');
+ assert.strictEqual(protobuf.getClientType(),
ClientType.LITE_SIMPLE_CONSUMER);
+ });
+
+ it('should expose group and empty lite topic set before startup', () => {
+ const impl = new LiteSimpleConsumerImpl(baseOptions);
+ assert.strictEqual(impl.getConsumerGroup(), 'lite-unittest-group');
+ assert.strictEqual(impl.getLiteTopicSet().size, 0);
+ });
+ });
+
+ describe('firstReadableMasterQueue', () => {
+ it('should keep only the first readable master queue', () => {
+ const routeData = new TopicRouteData([
+ buildProtoQueue(0, 1, Permission.READ_WRITE), // slave readable
+ buildProtoQueue(1, 0, Permission.WRITE), // master not readable
+ buildProtoQueue(2, 0, Permission.READ_WRITE), // first readable master
+ buildProtoQueue(3, 0, Permission.READ), // second readable master
+ ]);
+
+ const filtered = firstReadableMasterQueue(routeData);
+ assert.strictEqual(filtered.messageQueues.length, 1);
+ assert.strictEqual(filtered.messageQueues[0].broker.id, 0);
+ assert.strictEqual(filtered.messageQueues[0].queueId, 2);
+ });
+
+ it('should return empty route when no readable master exists', () => {
+ const routeData = new TopicRouteData([
+ buildProtoQueue(0, 0, Permission.WRITE),
+ buildProtoQueue(1, 1, Permission.READ),
+ ]);
+
+ const filtered = firstReadableMasterQueue(routeData);
+ assert.strictEqual(filtered.messageQueues.length, 0);
+ });
+ });
+
+ describe('LiteSubscriptionManager', () => {
+ const bindTopic = 'lite-parent-topic';
+
+ function createManager(host = new FakeHost()) {
+ const manager = new LiteSubscriptionManager(
+ host,
+ new Resource('', bindTopic),
+ new Resource('', 'lite-unittest-group'),
+ );
+ return { host, manager };
+ }
+
+ it('should sync PARTIAL_ADD to every sync endpoint on subscribeLite',
async () => {
+ const { host, manager } = createManager();
+ await manager.subscribeLite('lite-topic-1', OffsetOption.MIN_OFFSET);
+
+ assert.strictEqual(host.syncRecords.length, 2);
+ for (const record of host.syncRecords) {
+ assert.strictEqual(record.action, LiteSubscriptionAction.PARTIAL_ADD);
+ assert.deepStrictEqual(record.liteTopics, [ 'lite-topic-1' ]);
+ }
+ assert.ok(host.syncRecords.some(r => r.endpoints.facade ===
'127.0.0.1:8081'));
+ assert.ok(host.syncRecords.some(r => r.endpoints.facade ===
'127.0.0.1:8082'));
+ assert.ok(manager.getLiteTopicSet().has('lite-topic-1'));
+ assert.strictEqual(manager.getBindTopicName(), bindTopic);
+ assert.strictEqual(manager.getConsumerGroupName(),
'lite-unittest-group');
+ });
+
+ it('should reject subscribeLite when not running', async () => {
+ const { host, manager } = createManager();
+ host.running = false;
+ await assert.rejects(async () => {
+ await manager.subscribeLite('lite-topic-1');
+ }, /Consumer is not running/);
+ assert.strictEqual(host.syncRecords.length, 0);
+ });
+
+ it('should skip duplicate subscribeLite without extra rpc', async () => {
+ const { host, manager } = createManager();
+ await manager.subscribeLite('lite-topic-1');
+ host.resetRecords();
+ await manager.subscribeLite('lite-topic-1');
+ assert.strictEqual(host.syncRecords.length, 0);
+ assert.strictEqual(manager.getLiteTopicSet().size, 1);
+ });
+
+ it('should validate lite topic name and length', async () => {
+ const { manager } = createManager();
+ await assert.rejects(async () => {
+ await manager.subscribeLite(' ');
+ }, /liteTopic is blank/);
+ await assert.rejects(async () => {
+ await manager.subscribeLite('x'.repeat(65));
+ }, /liteTopic length exceeded max length 64/);
+ });
+
+ it('should throw when quota exceeded', async () => {
+ const { manager } = createManager();
+ const settings = new SettingsPB()
+ .setSubscription(new SubscriptionPB().setLiteSubscriptionQuota(1));
+ manager.sync(settings);
+ await manager.subscribeLite('lite-topic-1');
+ await assert.rejects(async () => {
+ await manager.subscribeLite('lite-topic-2');
+ }, /Lite subscription quota exceeded 1/);
+ assert.strictEqual(manager.getLiteTopicSet().size, 1);
+ });
+
+ it('should adopt maxLiteTopicSize from settings', async () => {
+ const { manager } = createManager();
+ const settings = new SettingsPB()
+ .setSubscription(new SubscriptionPB().setMaxLiteTopicSize(8));
+ manager.sync(settings);
+ await assert.rejects(async () => {
+ await manager.subscribeLite('x'.repeat(9));
+ }, /liteTopic length exceeded max length 8/);
+ });
+
+ it('should sync PARTIAL_REMOVE on unsubscribeLite', async () => {
+ const { host, manager } = createManager();
+ await manager.subscribeLite('lite-topic-1');
+ host.resetRecords();
+
+ await manager.unsubscribeLite('lite-topic-1');
+ assert.strictEqual(host.syncRecords.length, 2);
+ assert.ok(host.syncRecords.every(r => r.action ===
LiteSubscriptionAction.PARTIAL_REMOVE));
+ assert.strictEqual(manager.getLiteTopicSet().size, 0);
+ });
+
+ it('should skip unsubscribeLite for unknown topic', async () => {
+ const { host, manager } = createManager();
+ await manager.unsubscribeLite('never-subscribed');
+ assert.strictEqual(host.syncRecords.length, 0);
+ });
+
+ it('should fail subscription when any endpoint rejects the sync', async ()
=> {
+ const host = new FakeHost();
+ const manager = new LiteSubscriptionManager(
+ host,
+ new Resource('', bindTopic),
+ new Resource('', 'lite-unittest-group'),
+ );
+ // eslint-disable-next-line @typescript-eslint/no-explicit-any
+ (host as any).getRpcClientManager = () => ({
+ syncLiteSubscription: async (endpoints: Endpoints) => {
+ if (endpoints.facade === '127.0.0.1:8081') {
+ return new SyncLiteSubscriptionResponse().setStatus(
+ new Status().setCode(Code.INTERNAL_SERVER_ERROR),
+ );
+ }
+ return new SyncLiteSubscriptionResponse().setStatus(new
Status().setCode(Code.OK));
+ },
+ });
+
+ await assert.rejects(async () => {
+ await manager.subscribeLite('lite-topic-1');
+ }, /Failed to sync lite subscription/);
+ // Failed sync must not pollute the local set
+ assert.strictEqual(manager.getLiteTopicSet().size, 0);
+ });
+
+ it('should remove lite topic on notify unsubscribe command', async () => {
+ const { manager } = createManager();
+ await manager.subscribeLite('lite-topic-1');
+ const command = new
NotifyUnsubscribeLiteCommand().setLiteTopic('lite-topic-1');
+ manager.onNotifyUnsubscribeLiteCommand(command);
+ assert.strictEqual(manager.getLiteTopicSet().size, 0);
+ });
+
+ it('should ignore blank lite topic in notify unsubscribe command', async
() => {
+ const { manager } = createManager();
+ await manager.subscribeLite('lite-topic-1');
+ manager.onNotifyUnsubscribeLiteCommand(new
NotifyUnsubscribeLiteCommand().setLiteTopic(''));
+ assert.ok(manager.getLiteTopicSet().has('lite-topic-1'));
+ });
+
+ it('should shutdown cleanly and clear the topic set', async () => {
+ const { manager } = createManager();
+ await manager.subscribeLite('lite-topic-1');
+ manager.startUp();
+ manager.shutdown();
+ assert.strictEqual(manager.getLiteTopicSet().size, 0);
+ manager.shutdown(); // idempotent
+ });
+ });
+});
