You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by ji...@apache.org on 2026/04/23 08:55:27 UTC
(rocketmq-clients) branch master updated: Fix priority issue.Supports lightweight messaging. (#1215)
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 d71c6c53 Fix priority issue.Supports lightweight messaging. (#1215)
d71c6c53 is described below
commit d71c6c53acdd773f695575f75284d9c1cce3fdd9
Author: zhaohai <33...@users.noreply.github.com>
AuthorDate: Thu Apr 23 16:55:22 2026 +0800
Fix priority issue.Supports lightweight messaging. (#1215)
* fix: Add LITE and PRIORITY message types to match protocol definition
- Add LITE = 5 and PRIORITY = 6 to MessageType enum in definition.proto
- Align Node.js client with the latest protocol specification
- Fix inconsistency between proto definition and generated TypeScript/JS files
* fix: Update proto submodule and nodejs proto definition
- Update protos submodule to include LITE and PRIORITY message types
- Sync nodejs proto definition with protocol changes
- Ensure npm run init preserves LITE and PRIORITY types
* fix: Sync nodejs proto definition with updated protocol
- Update definition.proto to include LITE and PRIORITY message types
- Generated from updated protos submodule using npm run init
- Ensures Node.js client stays in sync with protocol changes
* feat: Add priority message support for Node.js client
- Add PRIORITY message type detection in PublishingMessage
- Set priority field in toProtobuf method
- Validate mutual exclusivity: priority cannot be used with messageGroup, deliveryTimestamp
- Ensure transactional messages cannot have priority set
- Add ProducerPriorityMessageExample demonstrating priority usage
- Align with Java implementation semantics
Key changes:
1. PublishingMessage: Detect PRIORITY message type and set it accordingly
2. PublishingMessage.toProtobuf(): Include priority in SystemProperties
3. Message: Validate priority >= 0 and mutual exclusivity constraints
4. Example: Comprehensive demo showing priority levels and validation
* fix: Use optional chaining for setPriority to avoid TS error
- Use (systemProperties as any).setPriority?.() as workaround
- Avoids TypeScript error when proto definition hasn't been updated yet
- Will work correctly once official proto includes priority field
- Maintains backward compatibility
* update proto
* update proto
* add lite message
* Change protos submodule URL to HTTPS for CI compatibility
* rollback
* Complete mutual exclusion verification
* Support Lite Push
* update protos version
* add lite simple consumer
* fix getClientType
* fix getClientType
* update startup
* update name
* fix producer getClientType
* fix producer getClientType
* update proto
* All 12 review comments from reviewer zk-drizzle have been completed.
* remove log
* optimize simpleConsumer
* remove delay
* update example
* update example
* Securely handle concurrent refresh requests and exceptional scenarios
* update producer delay
* fix lite push
* fix equals hashcode
* Optimize hashCodeOfString
* Optimize log
* Optimization of timer cleaning sequence
* TelemeterySession releases logic optimization
---
nodejs/examples/LiteProducerExample.ts | 85 +++++
nodejs/examples/LitePushConsumerExample.ts | 102 ++++++
nodejs/examples/ProducerDelayMessageExample.ts | 42 +--
nodejs/examples/ProducerFifoMessageExample.ts | 34 +-
nodejs/examples/ProducerNormalMessageExample.ts | 2 +-
nodejs/examples/ProducerPriorityMessageExample.ts | 182 +++++++----
nodejs/examples/ProducerRecallMessageExample.ts | 5 +-
nodejs/examples/ProducerSingleton.ts | 7 +-
.../examples/ProducerTransactionMessageExample.ts | 2 +-
nodejs/proto/apache/rocketmq/v2/definition.proto | 50 +++
nodejs/proto/apache/rocketmq/v2/service.proto | 33 ++
nodejs/scripts/build-grpc.sh | 3 +-
nodejs/src/client/BaseClient.ts | 210 ++++++++++--
nodejs/src/client/ClientId.ts | 2 +-
nodejs/src/client/RpcClient.ts | 13 +
nodejs/src/client/RpcClientManager.ts | 23 +-
nodejs/src/client/TelemetrySession.ts | 38 ++-
nodejs/src/consumer/Assignment.ts | 9 +-
nodejs/src/consumer/Assignments.ts | 8 +
nodejs/src/consumer/Consumer.ts | 79 +++--
nodejs/src/consumer/FilterExpression.ts | 15 +
nodejs/src/consumer/LitePushConsumer.ts | 256 +++++++++++++++
nodejs/src/consumer/LitePushConsumerImpl.ts | 245 ++++++++++++++
nodejs/src/consumer/LiteSimpleConsumer.ts | 234 ++++++++++++++
nodejs/src/consumer/LiteSimpleConsumerImpl.ts | 263 +++++++++++++++
nodejs/src/consumer/LiteSubscriptionManager.ts | 352 +++++++++++++++++++++
nodejs/src/consumer/OffsetOption.ts | 202 ++++++++++++
nodejs/src/consumer/ProcessQueue.ts | 15 +-
nodejs/src/consumer/PushConsumer.ts | 110 ++++++-
nodejs/src/consumer/PushSubscriptionSettings.ts | 3 +-
nodejs/src/consumer/SimpleConsumer.ts | 52 ++-
nodejs/src/consumer/SimpleSubscriptionSettings.ts | 4 +-
nodejs/src/consumer/StandardConsumeService.ts | 15 +-
nodejs/src/consumer/index.ts | 6 +
nodejs/src/exception/InternalErrorException.ts | 10 +
.../LiteSubscriptionQuotaExceededException.ts} | 15 +-
.../LiteTopicQuotaExceededException.ts} | 15 +-
nodejs/src/exception/StatusChecker.ts | 3 +
nodejs/src/exception/index.ts | 2 +
nodejs/src/index.ts | 1 +
nodejs/src/message/Message.ts | 41 ++-
nodejs/src/message/MessageId.ts | 18 ++
nodejs/src/message/MessageView.ts | 127 ++++++++
nodejs/src/message/PublishingMessage.ts | 24 +-
nodejs/src/producer/Producer.ts | 25 +-
nodejs/src/producer/PublishingLoadBalancer.ts | 20 ++
nodejs/src/route/Broker.ts | 17 +
nodejs/src/route/Endpoints.ts | 28 ++
nodejs/src/route/MessageQueue.ts | 20 +-
nodejs/src/route/Resource.ts | 102 ++++++
nodejs/src/route/TopicRouteData.ts | 36 +++
nodejs/src/route/index.ts | 1 +
nodejs/src/util/index.ts | 19 ++
.../test/consumer/LitePushConsumerBuilder.test.ts | 222 +++++++++++++
nodejs/test/consumer/OffsetOption.test.ts | 114 +++++++
.../LiteSubscriptionQuotaExceededException.test.ts | 81 +++++
nodejs/test/message/LiteMessage.test.ts | 105 ++++++
nodejs/test/message/PublishingLiteMessage.test.ts | 59 ++++
protos | 2 +-
59 files changed, 3563 insertions(+), 245 deletions(-)
diff --git a/nodejs/examples/LiteProducerExample.ts b/nodejs/examples/LiteProducerExample.ts
new file mode 100644
index 00000000..446acc20
--- /dev/null
+++ b/nodejs/examples/LiteProducerExample.ts
@@ -0,0 +1,85 @@
+/**
+ * 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 { Producer, LiteTopicQuotaExceededException } from '../src';
+import { endpoints, namespace, sessionCredentials, liteTopicConfig } from './ProducerSingleton';
+
+async function main() {
+ console.log('=== Lite Producer Example ===\n');
+
+ // Create producer
+ const producer = new Producer({
+ endpoints,
+ namespace,
+ sessionCredentials,
+ maxAttempts: 3,
+ });
+
+ try {
+ // Start producer
+ await producer.startup();
+ console.log('Producer started successfully\n');
+
+ // Define message body
+ const body = Buffer.from('This is a lite message for Apache RocketMQ');
+
+ // Send Lite Topic messages
+ for (let i = 1; i <= 5; i++) {
+ const liteTopicName = `lite-topic-${i}`;
+ const message = {
+ topic: liteTopicConfig.parentTopic,
+ keys: [ `key-${i}` ],
+ body,
+ liteTopic: liteTopicName, // Set lite topic
+ };
+
+ try {
+ const sendReceipt = await producer.send(message);
+ console.log(`✓ Message ${i} sent successfully`);
+ console.log(` - Topic: ${liteTopicConfig.parentTopic}`);
+ console.log(` - Lite Topic: ${liteTopicName}`);
+ console.log(` - Message ID: ${sendReceipt.messageId}\n`);
+ } catch (error) {
+ if (error instanceof LiteTopicQuotaExceededException) {
+ // Lite topic quota exceeded.
+ // Evaluate and increase the lite topic resource limit.
+ console.error(`✗ Lite topic quota exceeded for ${liteTopicName}:`, error.message);
+ } else {
+ console.error(`✗ Failed to send message ${i}:`, error.message);
+ }
+ }
+
+ // Wait 1 second between sends
+ await new Promise(resolve => setTimeout(resolve, 1000));
+ }
+
+ console.log('All messages sent!\n');
+ } catch (error) {
+ console.error('Error:', error);
+ } finally {
+ // Shutdown producer
+ if (producer) {
+ await producer.shutdown();
+ console.log('Producer shutdown successfully');
+ }
+ }
+}
+
+main().catch(err => {
+ console.error('Program error:', err);
+ process.exit(1);
+});
diff --git a/nodejs/examples/LitePushConsumerExample.ts b/nodejs/examples/LitePushConsumerExample.ts
new file mode 100644
index 00000000..e0eff4b9
--- /dev/null
+++ b/nodejs/examples/LitePushConsumerExample.ts
@@ -0,0 +1,102 @@
+/**
+ * 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.
+ */
+
+/**
+ * LitePushConsumer Example
+ *
+ * This example demonstrates how to use LitePushConsumer to consume messages
+ * from lite topics with reduced overhead.
+ *
+ * Prerequisites:
+ * - RocketMQ server with lite topic support enabled
+ * - A parent topic created on the server
+ * - Consumer group registered on the server
+ */
+
+import {
+ LitePushConsumerBuilder,
+ ConsumeResult,
+ OffsetOption,
+ type MessageView,
+} from '../src';
+import { endpoints, namespace, consumerGroup, sessionCredentials } from './ProducerSingleton';
+
+async function main() {
+ console.log('========== LitePushConsumer Example ==========\n');
+
+ // Create and start LitePushConsumer using builder pattern
+ const consumer = await new LitePushConsumerBuilder()
+ .setClientConfiguration({
+ endpoints,
+ namespace,
+ sessionCredentials,
+ })
+ .setConsumerGroup(consumerGroup)
+ .bindTopic('your-parent-topic') // Replace with your actual parent topic
+ .setMessageListener({
+ async consume(messageView: MessageView) {
+ console.log('Received message:', {
+ messageId: messageView.messageId,
+ topic: messageView.topic,
+ liteTopic: messageView.liteTopic,
+ tag: messageView.tag,
+ keys: messageView.keys,
+ body: messageView.body.toString('utf-8'),
+ });
+ return ConsumeResult.SUCCESS;
+ },
+ })
+ .setMaxCacheMessageCount(1024)
+ .setMaxCacheMessageSizeInBytes(64 * 1024 * 1024) // 64MB
+ .setConsumptionThreadCount(20)
+ .build();
+
+ console.log('✓ Consumer started successfully\n');
+
+ try {
+ // Subscribe to lite topics with different offset strategies
+ console.log('Subscribing to lite topics...\n');
+
+ await consumer.subscribeLite('lite-topic-1');
+ console.log('✓ Subscribed to lite-topic-1 (from last offset)');
+
+ await consumer.subscribeLite('lite-topic-2', OffsetOption.MIN_OFFSET);
+ console.log('✓ Subscribed to lite-topic-2 (from minimum offset)');
+
+ await consumer.subscribeLite('lite-topic-3', OffsetOption.ofTailN(100));
+ console.log('✓ Subscribed to lite-topic-3 (last 100 messages)\n');
+
+ // Keep running to receive messages
+ console.log('Consumer is running. Press Ctrl+C to exit...\n');
+ // eslint-disable-next-line @typescript-eslint/no-empty-function
+ await new Promise(() => {});
+ } catch (error) {
+ console.error('Error during consumption:', error);
+ throw error;
+ } finally {
+ // Close consumer to release resources
+ console.log('\nShutting down consumer...');
+ await consumer.close();
+ console.log('✓ Consumer closed successfully');
+ }
+}
+
+// Run example
+main().catch(error => {
+ console.error('Fatal error:', error);
+ process.exit(1);
+});
diff --git a/nodejs/examples/ProducerDelayMessageExample.ts b/nodejs/examples/ProducerDelayMessageExample.ts
index 6ce066a8..eb5c9969 100644
--- a/nodejs/examples/ProducerDelayMessageExample.ts
+++ b/nodejs/examples/ProducerDelayMessageExample.ts
@@ -23,16 +23,17 @@ import { topics, endpoints, sessionCredentials, namespace } from './ProducerSing
endpoints,
namespace,
sessionCredentials,
- maxAttempts: 2,
+ maxAttempts: 3,
});
await producer.startup();
try {
- // 发送延迟消息
+ // Send delay message
+ const deliveryTimestamp = new Date(Date.now() + 5000); // Deliver after 5 seconds
const receipt = await producer.send({
topic: topics.delay,
tag: 'rocketmq-delay',
- delay: 5000,
+ deliveryTimestamp,
body: Buffer.from(JSON.stringify({
hello: 'rocketmq-client-nodejs world ',
now: Date(),
@@ -41,41 +42,8 @@ import { topics, endpoints, sessionCredentials, namespace } from './ProducerSing
console.log('✅ Message sent successfully');
console.log(' - Message ID:', receipt.messageId);
- console.log(' - Recall Handle:', receipt.recallHandle);
+ console.log(' - Delivery Timestamp:', deliveryTimestamp.toISOString());
console.log(' - Offset:', receipt.offset);
-
- // 检查 recallHandle 是否存在
- if (!receipt.recallHandle || receipt.recallHandle.trim() === '') {
- console.warn('\n⚠️ Warning: Recall handle is empty');
- console.log('This might be because:');
- console.log('1. The topic is not configured as DELAY type');
- console.log('2. Broker does not support mixed message type');
- console.log('3. The message was not recognized as a delay message');
-
- // 尝试检查 Topic 配置
- console.log('\n💡 Suggestion: Check if the topic "time-topic" has message.type=DELAY attribute');
- } else {
- console.log('\n🔄 Attempting to recall message...');
-
- try {
- const recallReceipt = await producer.recallMessage(topics.delay, receipt.recallHandle);
- console.log('✅ Message recalled successfully!');
- console.log(' - Recalled Message ID:', recallReceipt.messageId);
- } catch (recallError) {
- console.error('❌ Failed to recall message:');
- console.error(' Error:', (recallError as Error).message);
- console.error(' Status Code:', (recallError as any).code);
-
- // 提供更多调试信息
- if ((recallError as Error).message.includes('recall handle is invalid')) {
- console.log('\n📝 Possible reasons:');
- console.log(' 1. The recall handle format is incorrect');
- console.log(' 2. The message has already been delivered/consumed');
- console.log(' 3. The recall time window has expired');
- console.log(' 4. Broker configuration issue (enableMixedMessageType=false)');
- }
- }
- }
} catch (error) {
console.error('❌ Error sending message:', error);
} finally {
diff --git a/nodejs/examples/ProducerFifoMessageExample.ts b/nodejs/examples/ProducerFifoMessageExample.ts
index 1412191a..febd74cc 100644
--- a/nodejs/examples/ProducerFifoMessageExample.ts
+++ b/nodejs/examples/ProducerFifoMessageExample.ts
@@ -17,10 +17,10 @@
/**
* Producer with FIFO Message Example
- *
+ *
* This example demonstrates how to send FIFO (First-In-First-Out) messages
* that will be consumed in order by PushConsumer with FIFO support.
- *
+ *
* Key points for FIFO messages:
* - Messages with the same messageGroup are stored and delivered in order
* - Different messageGroups can be consumed concurrently
@@ -46,7 +46,7 @@ async function main() {
// Send FIFO messages with different message groups
// Messages within the same group will be consumed in strict order
-
+
// Group 1: Order processing sequence
console.log('Sending FIFO messages for Group 1 (Order-001)...');
await sendFifoMessages(producer, 'Order-001', [
@@ -72,22 +72,6 @@ async function main() {
{ type: 'TRANSFER', accountId: 'Alice', amount: 100, balance: 700 },
]);
- // Send some messages without group (will be consumed concurrently)
- console.log('\nSending non-FIFO messages (no messageGroup)...');
- for (let i = 0; i < 3; i++) {
- const receipt = await producer.send({
- topic: topics.fifo,
- tag: 'non-fifo',
- body: Buffer.from(JSON.stringify({
- type: 'NON_FIFO',
- index: i,
- timestamp: Date.now(),
- note: 'This message has no messageGroup, will be consumed concurrently',
- })),
- });
- console.log(`Non-FIFO message ${i} sent:`, receipt.messageId);
- }
-
console.log('\n✓ All FIFO messages sent successfully!');
console.log('\nNote: Start PushConsumerFifoMessageExample to consume these messages.');
console.log('Messages with the same messageGroup will be consumed in strict order.\n');
@@ -104,13 +88,13 @@ async function main() {
async function sendFifoMessages(
producer: Producer,
messageGroup: string,
- messages: Array<Record<string, any>>
+ messages: Array<Record<string, any>>,
) {
console.log(`\n Sending ${messages.length} messages to group "${messageGroup}"...`);
-
+
for (let i = 0; i < messages.length; i++) {
const messageData = messages[i];
-
+
const receipt = await producer.send({
topic: topics.fifo,
tag: 'fifo-message',
@@ -122,7 +106,7 @@ async function sendFifoMessages(
})),
messageGroup, // Required for FIFO ordering
});
-
+
console.log(` [${i + 1}/${messages.length}] Sent:`, {
messageId: receipt.messageId,
sequence: i + 1,
@@ -132,9 +116,9 @@ async function sendFifoMessages(
// Small delay between messages to ensure clear sequencing
await new Promise(resolve => setTimeout(resolve, 10));
}
-
+
console.log(` ✓ Completed sending ${messages.length} messages for group "${messageGroup}"`);
}
// Run example
-main().catch(console.error);
\ No newline at end of file
+main().catch(console.error);
diff --git a/nodejs/examples/ProducerNormalMessageExample.ts b/nodejs/examples/ProducerNormalMessageExample.ts
index e00a28cc..2bbc7be1 100644
--- a/nodejs/examples/ProducerNormalMessageExample.ts
+++ b/nodejs/examples/ProducerNormalMessageExample.ts
@@ -24,7 +24,7 @@ import { topics, endpoints, sessionCredentials, namespace } from './ProducerSing
endpoints,
namespace,
sessionCredentials,
- maxAttempts: 2,
+ maxAttempts: 3,
});
await producer.startup();
diff --git a/nodejs/examples/ProducerPriorityMessageExample.ts b/nodejs/examples/ProducerPriorityMessageExample.ts
index 40bbf696..0a6e512b 100644
--- a/nodejs/examples/ProducerPriorityMessageExample.ts
+++ b/nodejs/examples/ProducerPriorityMessageExample.ts
@@ -16,103 +16,153 @@
*/
/**
- * This example demonstrates how to send priority messages using Apache RocketMQ.
+ * Producer with Priority Message Example
*
- * Priority messages allow you to assign different priority levels to messages,
- * ensuring that higher priority messages are processed before lower priority ones.
+ * This example demonstrates how to send priority messages that will be
+ * consumed based on their priority level. Higher priority messages are
+ * delivered before lower priority ones.
*
- * Key points:
- * - Priority is a non-negative number (0, 1, 2, ...)
+ * Key points for priority messages:
+ * - Priority must be >= 0
* - Higher priority messages are delivered first
- * - Priority cannot be set together with deliveryTimestamp or messageGroup
- * - Messages without priority are treated as normal priority
+ * - Cannot be used with messageGroup, deliveryTimestamp, or liteTopic
+ * - Transactional messages cannot have priority set
*/
-
-import { Producer } from '../src/producer/Producer';
-import { Message } from '../src/message/Message';
-
-// Replace with your actual configuration
-const ENDPOINT = process.env.ROCKETMQ_ENDPOINT || 'yourEndpoint';
-const ACCESS_KEY = process.env.ROCKETMQ_ACCESS_KEY || 'yourAccessKey';
-const SECRET_KEY = process.env.ROCKETMQ_SECRET_KEY || 'yourSecretKey';
+import { Producer } from '../src';
+import { endpoints, sessionCredentials, namespace } from './ProducerSingleton';
async function main() {
- // Create a producer instance
+ console.log('========== Producer Priority Message Example ==========');
+
+ // Create producer instance
const producer = new Producer({
- namespace: process.env.ROCKETMQ_NAMESPACE || 'yourNamespace',
- endpoints: ENDPOINT,
- sessionCredentials: {
- accessKey: ACCESS_KEY,
- accessSecret: SECRET_KEY,
- },
- topics: [ 'yourPriorityTopic' ],
- requestTimeout: 3000,
+ namespace,
+ endpoints,
+ sessionCredentials,
+ maxAttempts: 3,
});
try {
- // Start the producer
await producer.startup();
- console.log('Producer started successfully');
+ console.log('Producer started successfully!\n');
- // Define message body
- const body = Buffer.from('This is a priority message for Apache RocketMQ', 'utf-8');
- const tag = 'yourMessageTagA';
+ // Send messages with different priority levels
+ console.log('Sending priority messages...');
- // Send high priority message (priority = 1, highest)
- const highPriorityMessage = new Message({
+ // Low priority message (priority = 1)
+ console.log('\n1. Sending LOW priority message (priority=1)...');
+ const lowPriorityReceipt = await producer.send({
topic: 'yourPriorityTopic',
- body,
- tag,
- keys: [ 'high-priority-key' ],
- priority: 1, // High priority
+ tag: 'low-priority',
+ body: Buffer.from(JSON.stringify({
+ type: 'LOW_PRIORITY',
+ content: 'This is a low priority notification',
+ timestamp: Date.now(),
+ })),
+ priority: 1, // Lower number = lower priority
});
+ console.log(' ✓ Low priority message sent:', lowPriorityReceipt.messageId);
- const highPriorityReceipt = await producer.send(highPriorityMessage);
- console.log('[High Priority] Message sent successfully, messageId=%s', highPriorityReceipt.messageId);
-
- // Send medium priority message (priority = 5)
- const mediumPriorityMessage = new Message({
+ // Medium priority message (priority = 5)
+ console.log('\n2. Sending MEDIUM priority message (priority=5)...');
+ const mediumPriorityReceipt = await producer.send({
topic: 'yourPriorityTopic',
- body: Buffer.from('Medium priority message', 'utf-8'),
- tag,
- keys: [ 'medium-priority-key' ],
+ tag: 'medium-priority',
+ body: Buffer.from(JSON.stringify({
+ type: 'MEDIUM_PRIORITY',
+ content: 'This is a medium priority alert',
+ timestamp: Date.now(),
+ })),
priority: 5, // Medium priority
});
+ console.log(' ✓ Medium priority message sent:', mediumPriorityReceipt.messageId);
- const mediumPriorityReceipt = await producer.send(mediumPriorityMessage);
- console.log('[Medium Priority] Message sent successfully, messageId=%s', mediumPriorityReceipt.messageId);
-
- // Send low priority message (priority = 10)
- const lowPriorityMessage = new Message({
+ // High priority message (priority = 10)
+ console.log('\n3. Sending HIGH priority message (priority=10)...');
+ const highPriorityReceipt = await producer.send({
topic: 'yourPriorityTopic',
- body: Buffer.from('Low priority message', 'utf-8'),
- tag,
- keys: [ 'low-priority-key' ],
- priority: 10, // Low priority
+ tag: 'high-priority',
+ body: Buffer.from(JSON.stringify({
+ type: 'HIGH_PRIORITY',
+ content: 'This is a high priority urgent alert',
+ timestamp: Date.now(),
+ })),
+ priority: 10, // Higher number = higher priority
});
+ console.log(' ✓ High priority message sent:', highPriorityReceipt.messageId);
- const lowPriorityReceipt = await producer.send(lowPriorityMessage);
- console.log('[Low Priority] Message sent successfully, messageId=%s', lowPriorityReceipt.messageId);
-
- // Send normal message (no priority specified)
- const normalMessage = new Message({
+ // Critical priority message (priority = 100)
+ console.log('\n4. Sending CRITICAL priority message (priority=100)...');
+ const criticalReceipt = await producer.send({
topic: 'yourPriorityTopic',
- body: Buffer.from('Normal priority message', 'utf-8'),
- tag,
- keys: [ 'normal-key' ],
+ tag: 'critical-priority',
+ body: Buffer.from(JSON.stringify({
+ type: 'CRITICAL_PRIORITY',
+ content: 'This is a critical emergency notification',
+ timestamp: Date.now(),
+ })),
+ priority: 100, // Very high priority
});
-
- const normalReceipt = await producer.send(normalMessage);
- console.log('[Normal Priority] Message sent successfully, messageId=%s', normalReceipt.messageId);
+ console.log(' ✓ Critical priority message sent:', criticalReceipt.messageId);
+
+ console.log('\n✓ All priority messages sent successfully!');
+ console.log('\nNote: Messages will be delivered to consumers based on priority.');
+ console.log('Higher priority messages (e.g., priority=100) will be delivered');
+ console.log('before lower priority ones (e.g., priority=1).\n');
+
+ // Demonstrate mutual exclusivity constraints
+ console.log('Demonstrating mutual exclusivity constraints...\n');
+
+ try {
+ // This will fail: priority + messageGroup
+ await producer.send({
+ topic: 'yourPriorityTopic',
+ body: Buffer.from('test'),
+ priority: 5,
+ messageGroup: 'group-1',
+ });
+ console.log('✗ ERROR: Should have failed when setting both priority and messageGroup');
+ } catch (error: any) {
+ console.log('✓ Correctly rejected: priority + messageGroup');
+ console.log(` Error: ${error.message}\n`);
+ }
+
+ try {
+ // This will fail: priority + deliveryTimestamp
+ await producer.send({
+ topic: 'yourPriorityTopic',
+ body: Buffer.from('test'),
+ priority: 5,
+ deliveryTimestamp: new Date(Date.now() + 60000),
+ });
+ console.log('✗ ERROR: Should have failed when setting both priority and deliveryTimestamp');
+ } catch (error: any) {
+ console.log('✓ Correctly rejected: priority + deliveryTimestamp');
+ console.log(` Error: ${error.message}\n`);
+ }
+
+ try {
+ // This will fail: negative priority
+ await producer.send({
+ topic: 'yourPriorityTopic',
+ body: Buffer.from('test'),
+ priority: -1,
+ });
+ console.log('✗ ERROR: Should have failed with negative priority');
+ } catch (error: any) {
+ console.log('✓ Correctly rejected: negative priority');
+ console.log(` Error: ${error.message}\n`);
+ }
+
+ console.log('All validation checks passed!\n');
} catch (error) {
console.error('Failed to send priority messages:', error);
+ throw error;
} finally {
- // Close the producer when done
await producer.shutdown();
- console.log('Producer closed');
}
}
-// Run the example
+// Run example
main().catch(console.error);
diff --git a/nodejs/examples/ProducerRecallMessageExample.ts b/nodejs/examples/ProducerRecallMessageExample.ts
index 95bed3bf..2907f01f 100644
--- a/nodejs/examples/ProducerRecallMessageExample.ts
+++ b/nodejs/examples/ProducerRecallMessageExample.ts
@@ -33,7 +33,7 @@ import { topics, endpoints, sessionCredentials, namespace } from './ProducerSing
endpoints,
namespace,
sessionCredentials,
- maxAttempts: 2,
+ maxAttempts: 3,
});
try {
await producer.startup();
@@ -42,10 +42,11 @@ import { topics, endpoints, sessionCredentials, namespace } from './ProducerSing
// Send delay messages
const sendReceipts: any[] = [];
for (let i = 0; i < 5; i++) {
+ const deliveryTimestamp = new Date(Date.now() + 10000); // Deliver after 10 seconds, enough time to recall
const receipt = await producer.send({
topic: topics.delay,
tag: 'recall-test',
- delay: 10000, // 10 seconds delay, enough time to recall
+ deliveryTimestamp,
body: Buffer.from(JSON.stringify({
id: i,
message: `Delay message ${i} - will be recalled`,
diff --git a/nodejs/examples/ProducerSingleton.ts b/nodejs/examples/ProducerSingleton.ts
index 7afd8d82..226e958f 100644
--- a/nodejs/examples/ProducerSingleton.ts
+++ b/nodejs/examples/ProducerSingleton.ts
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-import { SessionCredentials } from '../src/client';
+import { SessionCredentials } from '../src';
export const endpoints = process.env.ROCKETMQ_NODEJS_CLIENT_ENDPOINTS ?? '127.0.0.1:8080';
export const namespace = process.env.ROCKETMQ_NODEJS_CLIENT_NAMESPACE ?? '';
@@ -29,6 +29,11 @@ export const topics = {
export const consumerGroup = process.env.ROCKETMQ_NODEJS_CLIENT_GROUP ?? 'nodejs-unittest-group';
export const tag = process.env.TAG ?? '*';
+// Lite topic configuration
+export const liteTopicConfig = {
+ parentTopic: process.env.ROCKETMQ_NODEJS_LITE_PARENT_TOPIC ?? 'yourParentTopic',
+};
+
export let sessionCredentials: SessionCredentials | undefined;
if (process.env.ROCKETMQ_NODEJS_CLIENT_KEY && process.env.ROCKETMQ_NODEJS_CLIENT_SECRET) {
sessionCredentials = {
diff --git a/nodejs/examples/ProducerTransactionMessageExample.ts b/nodejs/examples/ProducerTransactionMessageExample.ts
index a8290f8f..bed35708 100644
--- a/nodejs/examples/ProducerTransactionMessageExample.ts
+++ b/nodejs/examples/ProducerTransactionMessageExample.ts
@@ -26,7 +26,7 @@ const { TransactionResolution } = pkg;
endpoints,
namespace,
sessionCredentials,
- maxAttempts: 2,
+ maxAttempts: 3,
checker: {
async check(messageView) {
console.log(messageView);
diff --git a/nodejs/proto/apache/rocketmq/v2/definition.proto b/nodejs/proto/apache/rocketmq/v2/definition.proto
index 468c4105..516474df 100644
--- a/nodejs/proto/apache/rocketmq/v2/definition.proto
+++ b/nodejs/proto/apache/rocketmq/v2/definition.proto
@@ -146,6 +146,12 @@ enum MessageType {
// Messages that are transactional. Only committed messages are delivered to
// subscribers.
TRANSACTION = 4;
+
+ // lite topic
+ LITE = 5;
+
+ // Messages that lower prioritised ones may need to wait for higher priority messages to be processed first
+ PRIORITY = 6;
}
enum DigestType {
@@ -186,6 +192,8 @@ enum ClientType {
PUSH_CONSUMER = 2;
SIMPLE_CONSUMER = 3;
PULL_CONSUMER = 4;
+ LITE_PUSH_CONSUMER = 5;
+ LITE_SIMPLE_CONSUMER = 6;
}
enum Encoding {
@@ -270,6 +278,12 @@ message SystemProperties {
// Information to identify whether this message is from dead letter queue.
optional DeadLetterQueue dead_letter_queue = 20;
+
+ // lite topic
+ optional string lite_topic = 21;
+
+ // Priority of message, which is optional
+ optional int32 priority = 22;
}
message DeadLetterQueue {
@@ -348,6 +362,8 @@ enum Code {
ILLEGAL_POLLING_TIME = 40018;
// Offset is illegal.
ILLEGAL_OFFSET = 40019;
+ // Format of lite topic is illegal.
+ ILLEGAL_LITE_TOPIC = 40020;
// Generic code indicates that the client request lacks valid authentication
// credentials for the requested resource.
@@ -389,6 +405,10 @@ enum Code {
// Requests are throttled.
TOO_MANY_REQUESTS = 42900;
+ // LiteTopic related quota exceeded
+ LITE_TOPIC_QUOTA_EXCEEDED = 42901;
+ LITE_SUBSCRIPTION_QUOTA_EXCEEDED = 42902;
+
// Generic code for the case that the server is unwilling to process the request because its header fields are too large.
// The request may be resubmitted after reducing the size of the request header fields.
REQUEST_HEADER_FIELDS_TOO_LARGE = 43100;
@@ -548,6 +568,21 @@ message Subscription {
// Long-polling timeout for `ReceiveMessageRequest`, which is essential for
// push consumer.
optional google.protobuf.Duration long_polling_timeout = 5;
+
+ // Only lite push consumer
+ // client-side lite subscription quota limit
+ optional int32 lite_subscription_quota = 6;
+
+ // Only lite push consumer
+ // Maximum length limit for lite topic
+ optional int32 max_lite_topic_size = 7;
+}
+
+enum LiteSubscriptionAction {
+ PARTIAL_ADD = 0;
+ PARTIAL_REMOVE = 1;
+ COMPLETE_ADD = 2;
+ COMPLETE_REMOVE = 3;
}
message Metric {
@@ -567,4 +602,19 @@ enum QueryOffsetPolicy {
// Use this option if time-based seek is targeted.
TIMESTAMP = 2;
+}
+
+message OffsetOption {
+ oneof offset_type {
+ Policy policy = 1;
+ int64 offset = 2;
+ int64 tail_n = 3;
+ int64 timestamp = 4;
+ }
+
+ enum Policy {
+ LAST = 0;
+ MIN = 1;
+ MAX = 2;
+ }
}
\ No newline at end of file
diff --git a/nodejs/proto/apache/rocketmq/v2/service.proto b/nodejs/proto/apache/rocketmq/v2/service.proto
index 18db185a..b58ac41c 100644
--- a/nodejs/proto/apache/rocketmq/v2/service.proto
+++ b/nodejs/proto/apache/rocketmq/v2/service.proto
@@ -114,6 +114,7 @@ message ReceiveMessageResponse {
message AckMessageEntry {
string message_id = 1;
string receipt_handle = 2;
+ optional string lite_topic = 3;
}
message AckMessageRequest {
@@ -148,6 +149,7 @@ message ForwardMessageToDeadLetterQueueRequest {
string message_id = 4;
int32 delivery_attempt = 5;
int32 max_delivery_attempts = 6;
+ optional string lite_topic = 7;
}
message ForwardMessageToDeadLetterQueueResponse { Status status = 1; }
@@ -193,6 +195,10 @@ message RecoverOrphanedTransactionCommand {
string transaction_id = 2;
}
+message NotifyUnsubscribeLiteCommand {
+ string lite_topic = 1;
+}
+
message TelemetryCommand {
optional Status status = 1;
@@ -221,6 +227,9 @@ message TelemetryCommand {
// Request client to reconnect server use the latest endpoints.
ReconnectEndpointsCommand reconnect_endpoints_command = 8;
+
+ // Request client to unsubscribe lite topic.
+ NotifyUnsubscribeLiteCommand notify_unsubscribe_lite_command = 9;
}
}
@@ -243,6 +252,10 @@ message ChangeInvisibleDurationRequest {
// For message tracing
string message_id = 5;
+
+ optional string lite_topic = 6;
+ // If true, server will not increment the retry times for this message
+ optional bool suspend = 7;
}
message ChangeInvisibleDurationResponse {
@@ -311,6 +324,22 @@ message RecallMessageResponse {
string message_id = 2;
}
+message SyncLiteSubscriptionRequest {
+ LiteSubscriptionAction action = 1;
+ // bindTopic for lite push consumer
+ Resource topic = 2;
+ // consumer group
+ Resource group = 3;
+ // lite subscription set of lite topics
+ repeated string lite_topic_set = 4;
+ optional int64 version = 5;
+ optional OffsetOption offset_option = 6;
+}
+
+message SyncLiteSubscriptionResponse {
+ Status status = 1;
+}
+
// For all the RPCs in MessagingService, the following error handling policies
// apply:
//
@@ -440,4 +469,8 @@ service MessagingService {
// for normal message, not supported for now.
rpc RecallMessage(RecallMessageRequest) returns (RecallMessageResponse) {
}
+
+ // Sync lite subscription info, lite push consumer only
+ rpc SyncLiteSubscription(SyncLiteSubscriptionRequest) returns (SyncLiteSubscriptionResponse) {}
+
}
\ No newline at end of file
diff --git a/nodejs/scripts/build-grpc.sh b/nodejs/scripts/build-grpc.sh
index 72fd8090..d8004c26 100755
--- a/nodejs/scripts/build-grpc.sh
+++ b/nodejs/scripts/build-grpc.sh
@@ -20,7 +20,7 @@ ARCH=$(uname -m)
# Proto buf generation
PWD=$(pwd)
PATH_ROOT=$(dirname "$PWD")
-PATH_PROTO_ROOT="${PATH_ROOT}/protos"
+PATH_PROTO_ROOT="${PWD}/proto"
PATH_PROTO_OUTPUT="${PWD}/proto"
PATH_PROTO_OUTPUT_ROCKETMQ_V2="${PWD}/proto/apache/rocketmq/v2"
@@ -51,7 +51,6 @@ generateGrpc() {
--ts_out="grpc_js:$PATH_PROTO_OUTPUT" \
--grpc_out="grpc_js:$PATH_PROTO_OUTPUT" \
"$PATH_PROTO/$PATH_FILE"
- cp "$PATH_PROTO/$PATH_FILE" "${PATH_PROTO_OUTPUT_ROCKETMQ_V2}/"
}
echo ""
diff --git a/nodejs/src/client/BaseClient.ts b/nodejs/src/client/BaseClient.ts
index d1651663..7bd22802 100644
--- a/nodejs/src/client/BaseClient.ts
+++ b/nodejs/src/client/BaseClient.ts
@@ -29,6 +29,7 @@ import {
RecoverOrphanedTransactionCommand,
VerifyMessageCommand,
PrintThreadStackTraceCommand,
+ ReconnectEndpointsCommand,
TelemetryCommand,
ThreadStackTrace,
HeartbeatRequest,
@@ -74,7 +75,6 @@ export interface BaseClientOptions {
*/
export abstract class BaseClient {
readonly clientId = ClientId.create();
- readonly clientType = ClientType.CLIENT_TYPE_UNSPECIFIED;
readonly sslEnabled: boolean;
readonly #sessionCredentials?: SessionCredentials;
readonly namespace: string;
@@ -83,6 +83,8 @@ export abstract class BaseClient {
protected readonly requestTimeout: number;
protected readonly topics = new Set<string>();
protected readonly topicRouteCache = new Map<string, TopicRouteData>();
+ // In-flight route query futures to avoid duplicate queries for the same topic
+ protected readonly inflightRouteFutures = new Map<string, Promise<TopicRouteData>>();
protected readonly logger: ILogger;
protected readonly rpcClientManager: RpcClientManager;
readonly #telemetrySessions = new Map<string, TelemetrySession>();
@@ -91,11 +93,36 @@ export abstract class BaseClient {
#timers: NodeJS.Timeout[] = [];
#running = false;
+ /**
+ * Get the client type.
+ * Subclasses should override this method to return their specific client type.
+ *
+ * @return The client type identifier
+ */
+ protected getClientType(): ClientType {
+ return ClientType.CLIENT_TYPE_UNSPECIFIED;
+ }
+
constructor(options: BaseClientOptions) {
+ // Validate required parameters
+ if (!options.endpoints) {
+ throw new TypeError('endpoints is required');
+ }
+ if (typeof options.endpoints !== 'string' || options.endpoints.trim().length === 0) {
+ throw new TypeError('endpoints must be a non-empty string');
+ }
+
+ // Validate optional parameters
+ if (options.requestTimeout !== undefined) {
+ if (typeof options.requestTimeout !== 'number' || options.requestTimeout < 1000) {
+ throw new RangeError('requestTimeout must be a number >= 1000ms');
+ }
+ }
+
this.logger = options.logger ?? getDefaultLogger();
this.sslEnabled = options.sslEnabled === true;
this.endpoints = new Endpoints(options.endpoints);
- this.namespace = options.namespace;
+ this.namespace = options.namespace ?? '';
this.#sessionCredentials = options.sessionCredentials;
// https://rocketmq.apache.org/docs/introduction/03limits/
// Default request timeout is 3000ms
@@ -103,7 +130,9 @@ export abstract class BaseClient {
this.rpcClientManager = new RpcClientManager(this, this.logger);
if (options.topics) {
for (const topic of options.topics) {
- this.topics.add(topic);
+ if (topic && topic.trim().length > 0) {
+ this.topics.add(topic);
+ }
}
}
}
@@ -128,17 +157,53 @@ export abstract class BaseClient {
async #startup() {
this.logger.info('Begin to execute startup flow, clientId=%s, topics=%d',
this.clientId, this.topics.size);
- // fetch topic route
- await this.updateRoutes();
+
+ // Fetch topic route with retry mechanism (align with Java client)
+ const maxAttempts = 3;
+ let lastError: Error | undefined;
+ for (let attempt = 1; attempt <= maxAttempts; attempt++) {
+ try {
+ await this.updateRoutes();
+ this.logger.info('Fetch topic route successfully during startup, clientId=%s, attempt=%d',
+ this.clientId, attempt);
+ break;
+ } catch (e) {
+ lastError = e as Error;
+ if (attempt < maxAttempts) {
+ const backoffMs = 1000 * attempt; // Simple linear backoff: 1s, 2s, 3s
+ this.logger.warn('Fetch topic route failed during startup, will retry, clientId=%s, attempt=%d/%d, error=%s, backoff=%dms',
+ this.clientId, attempt, maxAttempts, e instanceof Error ? e.message : String(e), backoffMs);
+ await this.sleep(backoffMs);
+ } else {
+ this.logger.error('Fetch topic route failed after %d attempts, clientId=%s, error=%s',
+ maxAttempts, this.clientId, e instanceof Error ? e.message : String(e));
+ }
+ }
+ }
+
+ if (lastError && !this.topicRouteCache.size) {
+ throw new Error(`Failed to fetch topic routes after ${maxAttempts} attempts`, { cause: lastError });
+ }
+
this.logger.info('Topic routes updated, clientId=%s', this.clientId);
// update topic route every 30s
this.#timers.push(setInterval(async () => {
- this.updateRoutes();
+ try {
+ await this.updateRoutes();
+ } catch (e) {
+ this.logger.error('Failed to update routes periodically, clientId=%s, error=%s',
+ this.clientId, e instanceof Error ? e.message : String(e));
+ }
}, 30000));
// sync settings every 5m
this.#timers.push(setInterval(async () => {
- this.#syncSettings();
+ try {
+ this.#syncSettings();
+ } catch (e) {
+ this.logger.error('Failed to sync settings, clientId=%s, error=%s',
+ this.clientId, e instanceof Error ? e.message : String(e));
+ }
}, 5 * 60000));
// heartbeat every 10s
@@ -168,29 +233,62 @@ export abstract class BaseClient {
return this.#running;
}
+ protected sleep(ms: number): Promise<void> {
+ return new Promise(resolve => setTimeout(resolve, ms));
+ }
+
async shutdown() {
this.logger.info('Begin to shutdown the rocketmq client, clientId=%s', this.clientId);
this.#running = false;
+
+ // 1. Clear all timers
while (this.#timers.length > 0) {
const timer = this.#timers.pop();
- clearInterval(timer);
+ if (timer) {
+ clearInterval(timer);
+ }
}
+ // 2. Notify server termination
await this.#notifyClientTermination();
+ // 3. Release all telemetry sessions
this.logger.info('Begin to release all telemetry sessions, clientId=%s', this.clientId);
this.#releaseTelemetrySessions();
this.logger.info('Release all telemetry sessions successfully, clientId=%s', this.clientId);
+ // 4. Close RPC connections
this.rpcClientManager.close();
+
+ // 5. Clear caches
+ this.topicRouteCache.clear();
+ this.inflightRouteFutures.clear();
+ this.isolated.clear();
+
this.logger.info('Shutdown the rocketmq client successfully, clientId=%s', this.clientId);
this.logger.close && this.logger.close();
}
async #doHeartbeat() {
- const request = this.wrapHeartbeatRequest();
- for (const endpoints of this.getTotalRouteEndpoints()) {
- await this.rpcClientManager.heartbeat(endpoints, request, this.requestTimeout);
+ try {
+ const request = this.wrapHeartbeatRequest();
+ const endpointsList = this.getTotalRouteEndpoints();
+ if (endpointsList.length === 0) {
+ debug('No endpoints available for heartbeat, clientId=%s', this.clientId);
+ return;
+ }
+ for (const endpoints of endpointsList) {
+ try {
+ await this.rpcClientManager.heartbeat(endpoints, request, this.requestTimeout);
+ } catch (e) {
+ // Log but don't throw - heartbeat is best-effort
+ this.logger.warn('Heartbeat failed for endpoints=%s, clientId=%s, error=%s',
+ endpoints.facade, this.clientId, e instanceof Error ? e.message : String(e));
+ }
+ }
+ } catch (e) {
+ this.logger.error('Unexpected error in heartbeat, clientId=%s, error=%s',
+ this.clientId, e instanceof Error ? e.message : String(e));
}
}
@@ -226,31 +324,60 @@ export abstract class BaseClient {
}
}
- protected async getRouteData(topic: string) {
- let topicRouteData = this.topicRouteCache.get(topic);
- if (!topicRouteData) {
- this.topics.add(topic);
- topicRouteData = await this.#fetchTopicRoute(topic);
+ protected async getRouteData(topic: string): Promise<TopicRouteData> {
+ // Check cache first
+ const cached = this.topicRouteCache.get(topic);
+ if (cached) {
+ return cached;
+ }
+
+ // Check if there's an in-flight request for this topic
+ const inFlight = this.inflightRouteFutures.get(topic);
+ if (inFlight) {
+ debug('Reusing in-flight route query for topic=%s', topic);
+ return inFlight;
+ }
+
+ // Add topic to topics set
+ this.topics.add(topic);
+
+ // Create new route query and cache the promise
+ const future = this.#fetchTopicRoute(topic);
+ this.inflightRouteFutures.set(topic, future);
+
+ try {
+ const result = await future;
+ return result;
+ } finally {
+ // Clean up in-flight cache after completion (success or failure)
+ this.inflightRouteFutures.delete(topic);
}
- return topicRouteData;
}
async #fetchTopicRoute(topic: string) {
const req = new QueryRouteRequest();
req.setTopic(createResource(topic));
req.setEndpoints(this.endpoints.toProtobuf());
- const response = await this.rpcClientManager.queryRoute(this.endpoints, req, this.requestTimeout);
- StatusChecker.check(response.getStatus()?.toObject());
- const topicRouteData = new TopicRouteData(response.getMessageQueuesList());
- const newEndpoints = this.findNewRouteEndpoints(topicRouteData.getTotalEndpoints());
- for (const endpoints of newEndpoints) {
- // sync current settings to new endpoints
- this.getTelemetrySession(endpoints).syncSettings();
+
+ try {
+ const response = await this.rpcClientManager.queryRoute(this.endpoints, req, this.requestTimeout);
+ StatusChecker.check(response.getStatus()?.toObject());
+ const topicRouteData = new TopicRouteData(response.getMessageQueuesList());
+ const newEndpoints = this.findNewRouteEndpoints(topicRouteData.getTotalEndpoints());
+ for (const endpoints of newEndpoints) {
+ // sync current settings to new endpoints
+ this.getTelemetrySession(endpoints).syncSettings();
+ }
+ this.topicRouteCache.set(topic, topicRouteData);
+ this.onTopicRouteDataUpdate(topic, topicRouteData);
+ debug('fetchTopicRoute topic=%o topicRouteData=%j', topic, topicRouteData);
+ return topicRouteData;
+ } catch (e) {
+ const error = e instanceof Error ? e : new Error(String(e));
+ this.logger.error('Failed to fetch topic route, clientId=%s, topic=%s, endpoints=%s, error=%s',
+ this.clientId, topic, this.endpoints.facade, error.message);
+ throw error;
}
- this.topicRouteCache.set(topic, topicRouteData);
- this.onTopicRouteDataUpdate(topic, topicRouteData);
- debug('fetchTopicRoute topic=%o topicRouteData=%j', topic, topicRouteData);
- return topicRouteData;
}
#syncSettings() {
@@ -401,4 +528,31 @@ export abstract class BaseClient {
telemetryCommand.setStatus(new Status().setCode(Code.OK));
this.telemetry(endpoints, telemetryCommand);
}
+
+ // eslint-disable-next-line @typescript-eslint/no-unused-vars
+ onReconnectEndpointsCommand(endpoints: Endpoints, _command: ReconnectEndpointsCommand) {
+ this.logger.info('Received reconnect endpoints command from remote, will refresh telemetry session, endpoints=%s, clientId=%s',
+ endpoints, this.clientId);
+ // Refresh the telemetry session to use the latest endpoints
+ const session = this.getTelemetrySession(endpoints);
+ session.refresh();
+ }
+
+ /**
+ * Get the endpoints of this client.
+ *
+ * @return The endpoints
+ */
+ getEndpoints(): Endpoints {
+ return this.endpoints;
+ }
+
+ /**
+ * Get the RPC client manager.
+ *
+ * @return The RPC client manager
+ */
+ getRpcClientManager(): RpcClientManager {
+ return this.rpcClientManager;
+ }
}
diff --git a/nodejs/src/client/ClientId.ts b/nodejs/src/client/ClientId.ts
index 76cbfd19..52cf8228 100644
--- a/nodejs/src/client/ClientId.ts
+++ b/nodejs/src/client/ClientId.ts
@@ -23,7 +23,7 @@ import { hostname } from 'node:os';
*/
export class ClientId {
static #hostname = hostname();
- static #index = 0n;
+ static #index = BigInt(0);
static create() {
return `${this.#hostname}@${process.pid}@${this.#index++}@${Date.now().toString(36)}`;
diff --git a/nodejs/src/client/RpcClient.ts b/nodejs/src/client/RpcClient.ts
index aceb0d56..b485cc8d 100644
--- a/nodejs/src/client/RpcClient.ts
+++ b/nodejs/src/client/RpcClient.ts
@@ -46,6 +46,8 @@ import {
ReceiveMessageResponse,
SendMessageRequest,
SendMessageResponse,
+ SyncLiteSubscriptionRequest,
+ SyncLiteSubscriptionResponse,
UpdateOffsetRequest,
UpdateOffsetResponse,
} from '../../proto/apache/rocketmq/v2/service_pb';
@@ -255,4 +257,15 @@ export class RpcClient {
});
});
}
+
+ async syncLiteSubscription(request: SyncLiteSubscriptionRequest, metadata: Metadata, duration: number) {
+ const client = this.#getAndActivityRpcClient();
+ const deadline = this.#getDeadline(duration);
+ return new Promise<SyncLiteSubscriptionResponse>((resolve, reject) => {
+ client.syncLiteSubscription(request, metadata, { deadline }, (e, res) => {
+ if (e) return reject(e);
+ resolve(res);
+ });
+ });
+ }
}
diff --git a/nodejs/src/client/RpcClientManager.ts b/nodejs/src/client/RpcClientManager.ts
index 77ed3cf7..c363eb71 100644
--- a/nodejs/src/client/RpcClientManager.ts
+++ b/nodejs/src/client/RpcClientManager.ts
@@ -26,6 +26,7 @@ import {
QueryAssignmentRequest, QueryOffsetRequest, QueryRouteRequest,
ReceiveMessageRequest, SendMessageRequest, UpdateOffsetRequest,
RecallMessageRequest,
+ SyncLiteSubscriptionRequest,
} from '../../proto/apache/rocketmq/v2/service_pb';
import { Endpoints } from '../route';
import { ILogger } from './Logger';
@@ -75,11 +76,21 @@ export class RpcClientManager {
}
close() {
+ // Clear idle check timer first
+ if (this.#clearIdleRpcClientsTimer) {
+ clearInterval(this.#clearIdleRpcClientsTimer);
+ }
+
+ // Close all RPC clients and clear the map
for (const [ endpoints, rpcClient ] of this.#rpcClients.entries()) {
- rpcClient.close();
- this.#rpcClients.delete(endpoints);
+ try {
+ rpcClient.close();
+ } catch (e) {
+ this.#logger.warn('Failed to close RPC client for endpoints=%s, clientId=%s, error=%s',
+ endpoints.facade, this.#baseClient.clientId, e instanceof Error ? e.message : String(e));
+ }
}
- clearInterval(this.#clearIdleRpcClientsTimer);
+ this.#rpcClients.clear();
}
async queryRoute(endpoints: Endpoints, request: QueryRouteRequest, duration: number) {
@@ -177,4 +188,10 @@ export class RpcClientManager {
const metadata = this.#baseClient.getRequestMetadata();
return await rpcClient.recallMessage(request, metadata, duration);
}
+
+ async syncLiteSubscription(endpoints: Endpoints, request: SyncLiteSubscriptionRequest, duration: number) {
+ const rpcClient = this.#getRpcClient(endpoints);
+ const metadata = this.#baseClient.getRequestMetadata();
+ return await rpcClient.syncLiteSubscription(request, metadata, duration);
+ }
}
diff --git a/nodejs/src/client/TelemetrySession.ts b/nodejs/src/client/TelemetrySession.ts
index e36fd19e..d2abdef4 100644
--- a/nodejs/src/client/TelemetrySession.ts
+++ b/nodejs/src/client/TelemetrySession.ts
@@ -26,6 +26,7 @@ export class TelemetrySession {
#baseClient: BaseClient;
#logger: ILogger;
#stream: ClientDuplexStream<TelemetryCommand, TelemetryCommand>;
+ #isRefreshing = false;
constructor(baseClient: BaseClient, endpoints: Endpoints, logger: ILogger) {
this.#endpoints = endpoints;
@@ -39,8 +40,13 @@ export class TelemetrySession {
release() {
this.#logger.info('Begin to release telemetry session, endpoints=%s, clientId=%s',
this.#endpoints, this.#baseClient.clientId);
- this.#stream.end();
- this.#stream.removeAllListeners();
+ try {
+ this.#stream.end();
+ this.#stream.removeAllListeners();
+ } catch (e) {
+ this.#logger.warn('Failed to release telemetry session gracefully, endpoints=%s, clientId=%s, error=%s',
+ this.#endpoints, this.#baseClient.clientId, e instanceof Error ? e.message : String(e));
+ }
}
write(command: TelemetryCommand) {
@@ -52,6 +58,28 @@ export class TelemetrySession {
this.write(command);
}
+ refresh() {
+ if (this.#isRefreshing) {
+ this.#logger.warn('Telemetry session is already refreshing, skip this request, endpoints=%s, clientId=%s',
+ this.#endpoints, this.#baseClient.clientId);
+ return;
+ }
+
+ this.#isRefreshing = true;
+ this.#logger.info('Refreshing telemetry session, endpoints=%s, clientId=%s',
+ this.#endpoints, this.#baseClient.clientId);
+
+ try {
+ this.release();
+ this.#renewStream(false);
+ } catch (err) {
+ this.#logger.error('Failed to refresh telemetry session, endpoints=%s, clientId=%s, error=%s',
+ this.#endpoints, this.#baseClient.clientId, err);
+ } finally {
+ this.#isRefreshing = false;
+ }
+ }
+
#renewStream(inited: boolean) {
try {
this.#logger.debug?.('Creating telemetry stream, endpoints=%s, clientId=%s, inited=%s',
@@ -100,6 +128,12 @@ export class TelemetrySession {
this.#baseClient.onPrintThreadStackTraceCommand(endpoints, command.getPrintThreadStackTraceCommand()!);
break;
}
+ case TelemetryCommand.CommandCase.RECONNECT_ENDPOINTS_COMMAND: {
+ this.#logger.info('Receive reconnect endpoints command from remote, endpoints=%s, clientId=%s',
+ endpoints, clientId);
+ this.#baseClient.onReconnectEndpointsCommand(endpoints, command.getReconnectEndpointsCommand()!);
+ 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/Assignment.ts b/nodejs/src/consumer/Assignment.ts
index e281b3f6..5e8a2c97 100644
--- a/nodejs/src/consumer/Assignment.ts
+++ b/nodejs/src/consumer/Assignment.ts
@@ -27,10 +27,11 @@ export class Assignment {
equals(other: Assignment): boolean {
if (this === other) return true;
if (!other) return false;
- return this.messageQueue === other.messageQueue ||
- (this.messageQueue.queueId === other.messageQueue.queueId &&
- this.messageQueue.topic.name === other.messageQueue.topic.name &&
- this.messageQueue.broker.name === other.messageQueue.broker.name);
+ return this.messageQueue.equals(other.messageQueue);
+ }
+
+ hashCode(): number {
+ return this.messageQueue.hashCode();
}
toString(): string {
diff --git a/nodejs/src/consumer/Assignments.ts b/nodejs/src/consumer/Assignments.ts
index a6655114..cdb8a6e8 100644
--- a/nodejs/src/consumer/Assignments.ts
+++ b/nodejs/src/consumer/Assignments.ts
@@ -40,6 +40,14 @@ export class Assignments {
return true;
}
+ hashCode(): number {
+ let hash = 17;
+ for (const assignment of this.#assignmentList) {
+ hash = hash * 31 + assignment.hashCode();
+ }
+ return hash;
+ }
+
toString(): string {
return `Assignments{assignmentList=[${this.#assignmentList.map(a => a.toString()).join(', ')}]}`;
}
diff --git a/nodejs/src/consumer/Consumer.ts b/nodejs/src/consumer/Consumer.ts
index 19c7f66c..6fa2093e 100644
--- a/nodejs/src/consumer/Consumer.ts
+++ b/nodejs/src/consumer/Consumer.ts
@@ -21,6 +21,7 @@ import {
ChangeInvisibleDurationRequest,
ReceiveMessageRequest, ReceiveMessageResponse,
} from '../../proto/apache/rocketmq/v2/service_pb';
+import { ClientType } from '../../proto/apache/rocketmq/v2/definition_pb';
import { MessageView } from '../message';
import { MessageQueue } from '../route';
import { StatusChecker } from '../exception';
@@ -56,28 +57,46 @@ export abstract class Consumer extends BaseClient {
const endpoints = mq.broker.endpoints;
const timeout = this.requestTimeout + awaitDuration;
let status: Status.AsObject | undefined;
- const responses = await this.rpcClientManager.receiveMessage(endpoints, request, timeout);
- const messageList: Message[] = [];
- let transportDeliveryTimestamp: Date | undefined;
- for (const response of responses) {
- switch (response.getContentCase()) {
- case ReceiveMessageResponse.ContentCase.STATUS:
- status = response.getStatus()?.toObject();
- break;
- case ReceiveMessageResponse.ContentCase.MESSAGE:
- messageList.push(response.getMessage()!);
- break;
- case ReceiveMessageResponse.ContentCase.DELIVERY_TIMESTAMP:
- transportDeliveryTimestamp = response.getDeliveryTimestamp()?.toDate();
- break;
- default:
- // this.logger.warn("[Bug] Not recognized content for receive message response, mq={}, " +
- // "clientId={}, response={}", mq, clientId, response);
+
+ 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);
+
+ const responses = await this.rpcClientManager.receiveMessage(endpoints, request, timeout);
+ const messageList: Message[] = [];
+ let transportDeliveryTimestamp: Date | undefined;
+
+ for (const response of responses) {
+ switch (response.getContentCase()) {
+ case ReceiveMessageResponse.ContentCase.STATUS:
+ status = response.getStatus()?.toObject();
+ break;
+ case ReceiveMessageResponse.ContentCase.MESSAGE:
+ messageList.push(response.getMessage()!);
+ break;
+ case ReceiveMessageResponse.ContentCase.DELIVERY_TIMESTAMP:
+ transportDeliveryTimestamp = response.getDeliveryTimestamp()?.toDate();
+ break;
+ default:
+ // this.logger.warn("[Bug] Not recognized content for receive message response, mq={}, " +
+ // "clientId={}, response={}", mq, clientId, response);
+ }
+ }
+
+ StatusChecker.check(status);
+
+ if (messageList.length > 0) {
+ this.logger.debug?.('Received %d messages successfully, topic=%s, endpoints=%s, clientId=%s',
+ messageList.length, request.getMessageQueue()?.getTopic()?.getName(), endpoints, (this as any).clientId);
}
+
+ const messages = messageList.map(message => new MessageView(message, mq, transportDeliveryTimestamp));
+ 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);
+ throw err;
}
- StatusChecker.check(status);
- const messages = messageList.map(message => new MessageView(message, mq, transportDeliveryTimestamp));
- return messages;
}
protected async ackMessage(messageView: MessageView) {
@@ -126,4 +145,24 @@ export abstract class Consumer extends BaseClient {
const res = await this.rpcClientManager.forwardMessageToDeadLetterQueue(endpoints, request, timeout);
return res;
}
+
+ /**
+ * Get the consumer group name.
+ *
+ * @return Consumer group name
+ */
+ getConsumerGroup(): string {
+ return this.consumerGroup;
+ }
+
+ /**
+ * Check if this is a lite 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
+ || clientType === ClientType.LITE_SIMPLE_CONSUMER;
+ }
}
diff --git a/nodejs/src/consumer/FilterExpression.ts b/nodejs/src/consumer/FilterExpression.ts
index e050e8cc..aa0859c6 100644
--- a/nodejs/src/consumer/FilterExpression.ts
+++ b/nodejs/src/consumer/FilterExpression.ts
@@ -16,6 +16,7 @@
*/
import { FilterType, FilterExpression as FilterExpressionPB } from '../../proto/apache/rocketmq/v2/definition_pb';
+import { hashCodeOfString } from '../util';
const TAG_EXPRESSION_SUB_ALL = '*';
@@ -36,6 +37,20 @@ export class FilterExpression {
.setExpression(this.expression);
}
+ equals(other: FilterExpression): boolean {
+ if (this === other) return true;
+ if (!other) return false;
+ return this.expression === other.expression &&
+ this.filterType === other.filterType;
+ }
+
+ hashCode(): number {
+ let hash = 17;
+ hash = hash * 31 + hashCodeOfString(this.expression);
+ hash = hash * 31 + this.filterType;
+ return hash;
+ }
+
toString() {
return `FilterExpression(${this.filterType},${this.expression})`;
}
diff --git a/nodejs/src/consumer/LitePushConsumer.ts b/nodejs/src/consumer/LitePushConsumer.ts
new file mode 100644
index 00000000..3517150e
--- /dev/null
+++ b/nodejs/src/consumer/LitePushConsumer.ts
@@ -0,0 +1,256 @@
+/**
+ * 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 { MessageListener } from './MessageListener';
+import { OffsetOption } from './OffsetOption';
+import { LitePushConsumerImpl } from './LitePushConsumerImpl';
+
+/**
+ * LitePushConsumer interface for consuming messages from lite topics.
+ *
+ * <p>LitePushConsumer is a specialized consumer designed for lightweight scenarios
+ * with reduced metadata and storage overhead. It supports dynamic subscription
+ * management for lite topics.</p>
+ */
+export interface LitePushConsumer {
+ /**
+ * 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 consumeFromOption 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>;
+
+ /**
+ * Get the load balancing group for the consumer.
+ *
+ * @return Consumer load balancing group
+ */
+ getConsumerGroup(): string;
+
+ /**
+ * Close the consumer and release all related resources.
+ *
+ * <p>Once the consumer is closed, <strong>it could not be started once again.</strong>
+ * We maintain an FSM (finite-state machine) to record the different states for each
+ * push consumer.</p>
+ */
+ close(): Promise<void>;
+}
+
+export interface LitePushConsumerOptions extends BaseClientOptions {
+ consumerGroup: string;
+ bindTopic: string;
+ messageListener: MessageListener;
+}
+
+const CONSUMER_GROUP_PATTERN = /^[a-zA-Z0-9_-]+$/;
+
+/**
+ * LitePushConsumer builder class.
+ *
+ * <p>This class provides a fluent API for configuring and creating
+ * lite push consumers with reduced overhead for lightweight scenarios.</p>
+ */
+export class LitePushConsumerBuilder {
+ private options: Partial<LitePushConsumerOptions> = {};
+
+ /**
+ * Set the bind topic for the lite push consumer.
+ *
+ * @param bindTopic - The parent topic to bind
+ * @return This builder instance
+ * @throws Error if bindTopic is blank
+ */
+ bindTopic(bindTopic: string): LitePushConsumerBuilder {
+ 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): LitePushConsumerBuilder {
+ 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, doesn't match the pattern, or starts with 'GID-'
+ */
+ setConsumerGroup(consumerGroup: string): LitePushConsumerBuilder {
+ 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 message listener.
+ *
+ * @param messageListener - Message listener implementation
+ * @return This builder instance
+ * @throws Error if messageListener is null/undefined
+ */
+ setMessageListener(messageListener: MessageListener): LitePushConsumerBuilder {
+ if (!messageListener) {
+ throw new Error('messageListener should not be null');
+ }
+ this.options.messageListener = messageListener;
+ return this;
+ }
+
+ /**
+ * Set the maximum cache message count.
+ *
+ * @param maxCacheMessageCount - Maximum number of messages to cache
+ * @return This builder instance
+ * @throws Error if maxCacheMessageCount is not positive
+ */
+ setMaxCacheMessageCount(maxCacheMessageCount: number): LitePushConsumerBuilder {
+ if (maxCacheMessageCount <= 0) {
+ throw new Error('maxCacheMessageCount should be positive');
+ }
+ // TODO: Will be used when implementing full consumer
+ return this;
+ }
+
+ /**
+ * Set the maximum cache message size in bytes.
+ *
+ * @param maxCacheMessageSizeInBytes - Maximum cache size in bytes
+ * @return This builder instance
+ * @throws Error if maxCacheMessageSizeInBytes is not positive
+ */
+ setMaxCacheMessageSizeInBytes(maxCacheMessageSizeInBytes: number): LitePushConsumerBuilder {
+ if (maxCacheMessageSizeInBytes <= 0) {
+ throw new Error('maxCacheMessageSizeInBytes should be positive');
+ }
+ // TODO: Will be used when implementing full consumer
+ return this;
+ }
+
+ /**
+ * Set the consumption thread count.
+ *
+ * @param consumptionThreadCount - Number of threads for consumption
+ * @return This builder instance
+ * @throws Error if consumptionThreadCount is not positive
+ */
+ setConsumptionThreadCount(consumptionThreadCount: number): LitePushConsumerBuilder {
+ if (consumptionThreadCount <= 0) {
+ throw new Error('consumptionThreadCount should be positive');
+ }
+ // TODO: Will be used when implementing full consumer
+ return this;
+ }
+
+ /**
+ * Finalize the build of LitePushConsumer and start.
+ *
+ * <p>This method will block until the push consumer starts successfully.
+ *
+ * <p>Especially, if this method is invoked more than once, different push consumers will be created and started.
+ *
+ * @return Promise resolving to started LitePushConsumer instance
+ * @throws Error if required parameters are not set
+ */
+ async build(): Promise<LitePushConsumer> {
+ 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.messageListener) {
+ throw new Error('messageListener has not been set yet');
+ }
+ if (!this.options.bindTopic) {
+ throw new Error('bindTopic has not been set yet');
+ }
+
+ // Build LitePushConsumerImpl
+ const options: any = {
+ endpoints: this.options.endpoints,
+ namespace: this.options.namespace ?? '',
+ consumerGroup: this.options.consumerGroup,
+ bindTopic: this.options.bindTopic,
+ messageListener: this.options.messageListener,
+ sslEnabled: this.options.sslEnabled,
+ sessionCredentials: this.options.sessionCredentials,
+ requestTimeout: this.options.requestTimeout,
+ logger: this.options.logger,
+ };
+
+ const litePushConsumer = new LitePushConsumerImpl(options);
+ await litePushConsumer.startup();
+ return litePushConsumer;
+ }
+}
diff --git a/nodejs/src/consumer/LitePushConsumerImpl.ts b/nodejs/src/consumer/LitePushConsumerImpl.ts
new file mode 100644
index 00000000..e89337c2
--- /dev/null
+++ b/nodejs/src/consumer/LitePushConsumerImpl.ts
@@ -0,0 +1,245 @@
+/**
+ * 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 } from '../../proto/apache/rocketmq/v2/definition_pb';
+import { NotifyUnsubscribeLiteCommand } from '../../proto/apache/rocketmq/v2/service_pb';
+import { Resource } from '../route/Resource';
+import { Endpoints, TopicRouteData } from '../route';
+import { PushConsumer, PushConsumerOptions } from './PushConsumer';
+import { LitePushConsumer } from './LitePushConsumer';
+import { OffsetOption } from './OffsetOption';
+import { LiteSubscriptionManager } from './LiteSubscriptionManager';
+import { FilterExpression } from './FilterExpression';
+
+export interface LitePushConsumerOptions extends PushConsumerOptions {
+ bindTopic: string;
+}
+
+/**
+ * Implementation of LitePushConsumer.
+ *
+ * <p>LitePushConsumer extends PushConsumer to provide lightweight message consumption
+ * with reduced metadata and storage overhead. It supports dynamic subscription
+ * management for lite topics.</p>
+ *
+ * <p>Key features:</p>
+ * <ul>
+ * <li>Dynamic lite topic subscription/unsubscription</li>
+ * <li>Quota management for lite subscriptions</li>
+ * <li>Reduced resource consumption compared to standard PushConsumer</li>
+ * <li>Automatic synchronization with server settings</li>
+ * </ul>
+ *
+ * <p>Note: Unlike LiteSimpleConsumer, LitePushConsumer does not perform route optimization
+ * because it uses the assignment-based message delivery mechanism managed by the server.</p>
+ */
+export class LitePushConsumerImpl extends PushConsumer implements LitePushConsumer {
+ private readonly liteSubscriptionManager: LiteSubscriptionManager;
+ private readonly bindTopic: Resource;
+
+ constructor(options: LitePushConsumerOptions) {
+ // Create subscription expressions with bind topic
+ const subscriptions = new Map<string, FilterExpression>();
+ subscriptions.set(options.bindTopic, FilterExpression.SUB_ALL);
+
+ super({
+ ...options,
+ subscriptions,
+ });
+
+ 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 push consumer
+ */
+ protected getClientType(): ClientType {
+ return ClientType.LITE_PUSH_CONSUMER;
+ }
+
+ /**
+ * Override to disable standard assignment scanning for lite consumers.
+ * LitePushConsumer uses Lite Subscription mechanism instead of assignments.
+ */
+ // eslint-disable-next-line @typescript-eslint/no-unused-vars
+ protected onTopicRouteDataUpdate(_topic: string, _topicRouteData: TopicRouteData) {
+ // Lite consumers do not use standard assignment mechanism
+ // They rely on LiteSubscriptionManager for message delivery
+ }
+
+ /**
+ * 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();
+ }
+
+ /**
+ * Subscribe to a lite topic.
+ *
+ * <p>The subscribeLite() method initiates network requests and performs quota verification,
+ * so it may fail. It's important to handle potential errors. Possible failure scenarios include:</p>
+ * <ul>
+ * <li>Network request errors, which can be retried.</li>
+ * <li>Quota verification failures (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
+ */
+ async subscribeLite(liteTopic: string): Promise<void>;
+
+ /**
+ * Subscribe to a lite topic with 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> {
+ await this.liteSubscriptionManager.subscribeLite(liteTopic, offsetOption ?? null);
+ }
+
+ /**
+ * Unsubscribe from a lite topic.
+ *
+ * <p>This method removes the subscription and notifies the server.
+ * After unsubscribing, the consumer will no longer receive messages from this topic.</p>
+ *
+ * @param liteTopic - The name of the lite topic to unsubscribe from
+ */
+ async unsubscribeLite(liteTopic: string): Promise<void> {
+ await this.liteSubscriptionManager.unsubscribeLite(liteTopic);
+ }
+
+ /**
+ * Get the lite topic set.
+ *
+ * <p>Returns an immutable set of currently subscribed lite topics.
+ * This set reflects the current state of subscriptions managed by the lite subscription manager.</p>
+ *
+ * @return Set of lite topic names
+ */
+ getLiteTopicSet(): Set<string> {
+ return this.liteSubscriptionManager.getLiteTopicSet();
+ }
+
+ /**
+ * Get the load balancing group for the consumer.
+ *
+ * <p>The consumer group is used for load balancing across multiple consumer instances.
+ * All consumers in the same group will share the message load.</p>
+ *
+ * @return Consumer group name
+ */
+ getConsumerGroup(): string {
+ return this.liteSubscriptionManager.getConsumerGroupName();
+ }
+
+ /**
+ * Handle notify unsubscribe lite command from server.
+ *
+ * <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 command - The unsubscribe command from the server
+ */
+ onNotifyUnsubscribeLiteCommand(command: NotifyUnsubscribeLiteCommand) {
+ this.liteSubscriptionManager.onNotifyUnsubscribeLiteCommand(command);
+ }
+
+ /**
+ * Handle settings command from server.
+ *
+ * <p>This method processes configuration updates from the server and synchronizes
+ * the lite subscription manager with the latest settings.</p>
+ *
+ * @param endpoints - The server endpoints
+ * @param settings - The settings configuration
+ */
+ onSettingsCommand(endpoints: any, settings: any) {
+ super.onSettingsCommand(endpoints, settings);
+ this.liteSubscriptionManager.sync(settings);
+ }
+
+ /**
+ * Close the consumer.
+ */
+ async close(): Promise<void> {
+ await this.shutdown();
+ }
+
+ /**
+ * Get the logger.
+ */
+ getLogger() {
+ return this.logger;
+ }
+
+ /**
+ * Get the RPC client manager (protected access for internal use).
+ *
+ * @internal
+ */
+ getRpcClientManager() {
+ return this.rpcClientManager;
+ }
+
+ /**
+ * Get the endpoints (protected access for internal use).
+ *
+ * @internal
+ */
+ getEndpoints(): Endpoints {
+ return this.endpoints;
+ }
+
+ /**
+ * Get the request timeout (protected access for internal use).
+ *
+ * @internal
+ */
+ getRequestTimeout(): number {
+ return this.requestTimeout;
+ }
+}
diff --git a/nodejs/src/consumer/LiteSimpleConsumer.ts b/nodejs/src/consumer/LiteSimpleConsumer.ts
new file mode 100644
index 00000000..57234bd9
--- /dev/null
+++ b/nodejs/src/consumer/LiteSimpleConsumer.ts
@@ -0,0 +1,234 @@
+/**
+ * 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, LiteSimpleConsumerOptions } from './LiteSimpleConsumerImpl';
+
+/**
+ * Similar to SimpleConsumer, but for lite topic.
+ *
+ * <p>LiteSimpleConsumer provides lightweight message consumption with reduced
+ * metadata and storage overhead. It supports dynamic subscription management
+ * for lite topics.</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>
+ * <ol>
+ * <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>
+ * </ol>
+ *
+ * @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 offsetOption to specify the consume from offset.
+ *
+ * @param liteTopic the name of the lite topic to subscribe
+ * @param offsetOption the consume from offset
+ * @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>;
+
+ /**
+ * Receive messages 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 invisible duration of messages to return from the server
+ * @return list of message view
+ */
+ receive(maxMessageNum: number, invisibleDuration: number): Promise<MessageView[]>;
+
+ /**
+ * Acknowledge a message synchronously.
+ *
+ * <p>Duplicate ack request does not take effect and throw an exception.</p>
+ *
+ * @param messageView special message view with handle want to ack
+ */
+ ack(messageView: MessageView): Promise<void>;
+
+ /**
+ * Changes the invisible duration of a specified message synchronously.
+ *
+ * <p>The origin invisible duration for a message decide by ack request.</p>
+ * <p>Duplicate change requests will refresh the next visible time of this message to consumers.</p>
+ *
+ * @param messageView the message view to change invisible time
+ * @param invisibleDuration new timestamp the message could be visible and re-consume which start from current time
+ */
+ changeInvisibleDuration(messageView: MessageView, invisibleDuration: number): Promise<void>;
+
+ /**
+ * Close the consumer and release all related resources.
+ *
+ * <p>Once consumer is closed, <strong>it could not be started once again.</strong></p>
+ */
+ close(): Promise<void>;
+}
+
+/**
+ * LiteSimpleConsumer builder class.
+ *
+ * <p>This class provides a fluent API for configuring and creating
+ * lite simple consumers with reduced overhead for lightweight scenarios.</p>
+ */
+export class LiteSimpleConsumerBuilder {
+ private options: Partial<LiteSimpleConsumerOptions> = {};
+
+ /**
+ * Set the bind topic for the lite simple consumer.
+ *
+ * @param bindTopic the parent topic that lite topics belong to
+ * @return this builder
+ */
+ 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 the client configuration options
+ * @return this builder
+ */
+ 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.
+ *
+ * <p>Note: For lite consumers, the consumer group should NOT start with 'GID-' prefix.</p>
+ *
+ * @param consumerGroup the consumer group name
+ * @return this builder
+ */
+ setConsumerGroup(consumerGroup: string): LiteSimpleConsumerBuilder {
+ if (!consumerGroup) {
+ throw new Error('consumerGroup should not be null');
+ }
+ // Validate consumer group format (should not start with GID-)
+ if (consumerGroup.startsWith('GID-')) {
+ throw new Error(
+ 'Lite consumer group should not start with "GID-" prefix. ' +
+ 'Please use a different group name without the prefix.');
+ }
+ this.options.consumerGroup = consumerGroup;
+ return this;
+ }
+
+ /**
+ * Set the await duration for long-polling.
+ *
+ * @param awaitDuration the await duration in milliseconds
+ * @return this builder
+ */
+ setAwaitDuration(awaitDuration: number): LiteSimpleConsumerBuilder {
+ if (awaitDuration <= 0) {
+ throw new Error('awaitDuration should be greater than 0');
+ }
+ this.options.awaitDuration = awaitDuration;
+ return this;
+ }
+
+ /**
+ * Start up the LiteSimpleConsumer.
+ *
+ * @return the started LiteSimpleConsumer instance
+ * @throws Error if required parameters are not set
+ */
+ async startup(): Promise<LiteSimpleConsumer> {
+ // Validate required parameters
+ 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.awaitDuration) {
+ throw new Error('awaitDuration has not been set yet');
+ }
+ if (!this.options.bindTopic) {
+ throw new Error('bindTopic has not been set yet');
+ }
+
+ // Create options
+ const options: any = {
+ endpoints: this.options.endpoints,
+ namespace: this.options.namespace ?? '',
+ consumerGroup: this.options.consumerGroup,
+ bindTopic: this.options.bindTopic,
+ awaitDuration: this.options.awaitDuration,
+ subscriptions: new Map(), // Will be set in constructor
+ sslEnabled: this.options.sslEnabled,
+ sessionCredentials: this.options.sessionCredentials,
+ requestTimeout: this.options.requestTimeout,
+ logger: this.options.logger,
+ };
+
+ // Create and start the consumer
+ const consumer = new LiteSimpleConsumerImpl(options);
+ await consumer.startup();
+
+ return consumer;
+ }
+}
diff --git a/nodejs/src/consumer/LiteSimpleConsumerImpl.ts b/nodejs/src/consumer/LiteSimpleConsumerImpl.ts
new file mode 100644
index 00000000..1b0c933c
--- /dev/null
+++ b/nodejs/src/consumer/LiteSimpleConsumerImpl.ts
@@ -0,0 +1,263 @@
+/**
+ * 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 } from '../../proto/apache/rocketmq/v2/definition_pb';
+import { NotifyUnsubscribeLiteCommand } from '../../proto/apache/rocketmq/v2/service_pb';
+import { Resource } from '../route/Resource';
+import { TopicRouteData } from '../route';
+import { MessageView } from '../message';
+import { Endpoints } from '../route';
+import { SimpleConsumer, SimpleConsumerOptions } from './SimpleConsumer';
+import { LiteSimpleConsumer } from './LiteSimpleConsumer';
+import { OffsetOption } from './OffsetOption';
+import { LiteSubscriptionManager } from './LiteSubscriptionManager';
+import { FilterExpression } from './FilterExpression';
+
+export interface LiteSimpleConsumerOptions extends SimpleConsumerOptions {
+ bindTopic: string;
+}
+
+/**
+ * Implementation of LiteSimpleConsumer.
+ *
+ * <p>LiteSimpleConsumer extends SimpleConsumer to provide lightweight message consumption
+ * with reduced metadata and storage overhead. It supports dynamic subscription
+ * management for lite topics.</p>
+ *
+ * <p>Key differences from standard SimpleConsumer:</p>
+ * <ul>
+ * <li>Uses lite topic subscription mechanism</li>
+ * <li>Optimized routing - only uses first readable master queue</li>
+ * <li>Reduced resource consumption</li>
+ * </ul>
+ */
+export class LiteSimpleConsumerImpl extends SimpleConsumer implements LiteSimpleConsumer {
+ private readonly liteSubscriptionManager: LiteSubscriptionManager;
+ private readonly bindTopic: Resource;
+
+ // eslint-disable-next-line @typescript-eslint/no-useless-constructor
+ constructor(options: LiteSimpleConsumerOptions) {
+ // Create subscription expressions with bind topic
+ const subscriptions = new Map<string, FilterExpression>();
+ subscriptions.set(options.bindTopic, FilterExpression.SUB_ALL);
+
+ super({
+ ...options,
+ subscriptions,
+ });
+
+ this.bindTopic = new Resource(options.namespace, options.bindTopic);
+
+ const groupResource = new Resource(options.namespace, options.consumerGroup);
+ this.liteSubscriptionManager = new LiteSubscriptionManager(
+ this as any, // Type assertion needed due to protected methods
+ this.bindTopic,
+ groupResource,
+ );
+ }
+
+ /**
+ * Get the client type.
+ */
+ protected getClientType(): ClientType {
+ return ClientType.LITE_SIMPLE_CONSUMER;
+ }
+
+ /**
+ * Start up the consumer.
+ */
+ async startup() {
+ await super.startup();
+ this.liteSubscriptionManager.startUp();
+ }
+
+ /**
+ * Shutdown the consumer.
+ */
+ async shutdown() {
+ this.liteSubscriptionManager.shutdown();
+ await super.shutdown();
+ }
+
+ /**
+ * Subscribe to a lite topic.
+ */
+ async subscribeLite(liteTopic: string): Promise<void>;
+ async subscribeLite(liteTopic: string, offsetOption: OffsetOption): Promise<void>;
+ async subscribeLite(liteTopic: string, offsetOption?: OffsetOption): Promise<void> {
+ await this.liteSubscriptionManager.subscribeLite(liteTopic, offsetOption ?? null);
+ }
+
+ /**
+ * Unsubscribe from a lite topic.
+ */
+ async unsubscribeLite(liteTopic: string): Promise<void> {
+ await this.liteSubscriptionManager.unsubscribeLite(liteTopic);
+ }
+
+ /**
+ * Get the lite topic set.
+ */
+ getLiteTopicSet(): Set<string> {
+ return this.liteSubscriptionManager.getLiteTopicSet();
+ }
+
+ /**
+ * Get the load balancing group for the consumer.
+ */
+ getConsumerGroup(): string {
+ return this.liteSubscriptionManager.getConsumerGroupName();
+ }
+
+ /**
+ * Handle notify unsubscribe lite command from server.
+ */
+ onNotifyUnsubscribeLiteCommand(command: NotifyUnsubscribeLiteCommand) {
+ this.liteSubscriptionManager.onNotifyUnsubscribeLiteCommand(command);
+ }
+
+ /**
+ * Handle settings command from server.
+ */
+ onSettingsCommand(endpoints: Endpoints, settings: any) {
+ super.onSettingsCommand(endpoints, settings);
+ this.liteSubscriptionManager.sync(settings);
+ }
+
+ /**
+ * Override topic route data update to optimize routing for lite consumer.
+ * For lite consumers, we only need routes to brokers, so keep only the first readable queue.
+ * This reduces memory usage and network overhead significantly.
+ */
+ protected onTopicRouteDataUpdate(topic: string, topicRouteData: TopicRouteData) {
+ // Optimize: only use the first readable master queue for lite consumers
+ const originalQueueCount = topicRouteData.getQueueCount();
+ const optimizedRouteData = this.optimizeRouteDataForLite(topicRouteData);
+ const optimizedQueueCount = optimizedRouteData.getQueueCount();
+
+ if (originalQueueCount > 1 && optimizedQueueCount === 1) {
+ this.logger.info(
+ 'Lite consumer route optimization applied: topic=%s, original queues=%d, optimized queues=%d',
+ topic,
+ originalQueueCount,
+ optimizedQueueCount,
+ );
+ }
+
+ super.onTopicRouteDataUpdate(topic, optimizedRouteData);
+ }
+
+ /**
+ * Optimize route data for lite consumer by keeping only the first readable master queue.
+ * This reduces memory usage and network overhead.
+ *
+ * Note: Similar to Java implementation, we filter for readable master queues and take the first one.
+ */
+ private optimizeRouteDataForLite(topicRouteData: TopicRouteData): TopicRouteData {
+ // Find the first readable master queue (similar to Java's isReadableMasterQueue)
+ const readableMasterQueue = topicRouteData.messageQueues.find(mq => {
+ // Check if it's a master broker (id > 0 typically indicates master)
+ // and has read permission
+ return mq.broker.id >= 0 && mq.permission === 1; // Permission.READ = 1
+ });
+
+ if (readableMasterQueue) {
+ const firstQueuePb = readableMasterQueue.toProtobuf();
+ return new TopicRouteData([ firstQueuePb ]);
+ }
+
+ // If no readable master queue found, fall back to first available queue
+ if (topicRouteData.messageQueues.length > 0) {
+ const firstQueuePb = topicRouteData.messageQueues[ 0 ].toProtobuf();
+ return new TopicRouteData([ firstQueuePb ]);
+ }
+
+ // If no queues available, return original (will handle error downstream)
+ return topicRouteData;
+ }
+
+ /**
+ * Receive messages synchronously.
+ *
+ * @param maxMessageNum max message num of server returned
+ * @param invisibleDuration set the invisible duration of messages to return from the server
+ * @return list of message view
+ */
+ async receive(maxMessageNum: number, invisibleDuration: number): Promise<MessageView[]> {
+ return super.receive(maxMessageNum, invisibleDuration);
+ }
+
+ /**
+ * Acknowledge a message synchronously.
+ *
+ * @param messageView message view to ack
+ */
+ async ack(messageView: MessageView): Promise<void> {
+ return super.ack(messageView);
+ }
+
+ /**
+ * Change invisible duration synchronously.
+ *
+ * @param messageView message view to change invisible duration
+ * @param invisibleDuration new invisible duration
+ */
+ async changeInvisibleDuration(messageView: MessageView, invisibleDuration: number): Promise<void> {
+ return super.changeInvisibleDuration(messageView, invisibleDuration);
+ }
+
+ /**
+ * Close the consumer.
+ */
+ async close(): Promise<void> {
+ await this.shutdown();
+ }
+
+ /**
+ * Get the logger.
+ */
+ getLogger() {
+ return this.logger;
+ }
+
+ /**
+ * Get the RPC client manager (protected access for internal use).
+ *
+ * @internal
+ */
+ getRpcClientManager() {
+ return this.rpcClientManager;
+ }
+
+ /**
+ * Get the endpoints (protected access for internal use).
+ *
+ * @internal
+ */
+ getEndpoints(): Endpoints {
+ return this.endpoints;
+ }
+
+ /**
+ * Get the request timeout (protected access for internal use).
+ *
+ * @internal
+ */
+ getRequestTimeout(): number {
+ return this.requestTimeout;
+ }
+}
diff --git a/nodejs/src/consumer/LiteSubscriptionManager.ts b/nodejs/src/consumer/LiteSubscriptionManager.ts
new file mode 100644
index 00000000..52e153f7
--- /dev/null
+++ b/nodejs/src/consumer/LiteSubscriptionManager.ts
@@ -0,0 +1,352 @@
+/**
+ * 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 { Code, LiteSubscriptionAction } from '../../proto/apache/rocketmq/v2/definition_pb';
+import {
+ NotifyUnsubscribeLiteCommand,
+ SyncLiteSubscriptionRequest,
+ SyncLiteSubscriptionResponse,
+} from '../../proto/apache/rocketmq/v2/service_pb';
+import { ClientException } from '../exception';
+import { Resource } from '../route';
+import { LitePushConsumerImpl } from './LitePushConsumerImpl';
+import { OffsetOption } from './OffsetOption';
+
+const SYNC_LITE_SUBSCRIPTION_INTERVAL = 30000; // 30 seconds
+
+/**
+ * Manages lite topic subscriptions for LitePushConsumer.
+ *
+ * <p>LiteSubscriptionManager handles:
+ * - Maintaining the set of subscribed lite topics
+ * - Periodic synchronization of lite subscriptions with server
+ * - Quota management and verification
+ * - Handling unsubscribe commands from server</p>
+ */
+export class LiteSubscriptionManager {
+ private readonly consumerImpl: LitePushConsumerImpl;
+ private readonly bindTopic: Resource;
+ private readonly group: Resource;
+ private readonly liteTopicSet = new Set<string>();
+ private liteSubscriptionQuota: number;
+ private maxLiteTopicSize: number;
+ private syncTimer?: NodeJS.Timeout;
+
+ constructor(
+ consumerImpl: LitePushConsumerImpl,
+ bindTopic: Resource,
+ group: Resource,
+ ) {
+ this.consumerImpl = consumerImpl;
+ this.bindTopic = bindTopic;
+ this.group = group;
+ this.liteSubscriptionQuota = 100; // Default quota
+ this.maxLiteTopicSize = 64; // Default max length
+ }
+
+ /**
+ * Start periodic sync of lite subscriptions.
+ */
+ public startUp() {
+ this.syncTimer = setInterval(
+ () => this.syncAllLiteSubscription(),
+ SYNC_LITE_SUBSCRIPTION_INTERVAL,
+ );
+ }
+
+ /**
+ * Stop the sync timer.
+ */
+ public shutdown() {
+ // Clear sync timer
+ if (this.syncTimer) {
+ clearInterval(this.syncTimer);
+ this.syncTimer = undefined;
+ }
+ // Clear lite topic set to free memory
+ this.liteTopicSet.clear();
+ }
+
+ /**
+ * Get the bind topic name.
+ */
+ public getBindTopicName(): string {
+ return this.bindTopic.getName();
+ }
+
+ /**
+ * Get the consumer group name.
+ */
+ public getConsumerGroupName(): string {
+ return this.group.getName();
+ }
+
+ /**
+ * Get a copy of the lite topic set.
+ */
+ public getLiteTopicSet(): Set<string> {
+ return new Set(this.liteTopicSet);
+ }
+
+ /**
+ * Sync settings from server.
+ */
+ public sync(settings: any) {
+ if (!settings?.hasSubscription?.()) {
+ return;
+ }
+
+ const subscription = settings.getSubscription();
+ if (subscription.hasLiteSubscriptionQuota?.()) {
+ this.liteSubscriptionQuota = subscription.getLiteSubscriptionQuota();
+ }
+ if (subscription.hasMaxLiteTopicSize?.()) {
+ this.maxLiteTopicSize = subscription.getMaxLiteTopicSize();
+ }
+ }
+
+ /**
+ * Subscribe to a lite topic.
+ */
+ public async subscribeLite(
+ liteTopic: string,
+ offsetOption?: OffsetOption | null,
+ ): Promise<void> {
+ // Check if consumer is running
+ if (!this.consumerImpl.isRunning()) {
+ throw new ClientException(500, 'Consumer is not running');
+ }
+
+ // Already subscribed
+ if (this.liteTopicSet.has(liteTopic)) {
+ return;
+ }
+
+ // Validate lite topic
+ this.validateLiteTopic(liteTopic);
+
+ // Check quota
+ this.checkLiteSubscriptionQuota(1);
+
+ try {
+ await this.syncLiteSubscription(
+ LiteSubscriptionAction.PARTIAL_ADD,
+ [ liteTopic ],
+ offsetOption ?? null,
+ );
+
+ this.liteTopicSet.add(liteTopic);
+ this.consumerImpl.getLogger().info(
+ 'SubscribeLite %s, topic=%s, group=%s, clientId=%s',
+ liteTopic,
+ this.getBindTopicName(),
+ this.getConsumerGroupName(),
+ this.consumerImpl.clientId,
+ );
+ } catch (error) {
+ this.consumerImpl.getLogger().error(
+ 'Failed to subscribeLite %s, error=%s',
+ liteTopic,
+ error,
+ );
+ throw error;
+ }
+ }
+
+ /**
+ * Unsubscribe from a lite topic.
+ */
+ public async unsubscribeLite(liteTopic: string): Promise<void> {
+ // Check if consumer is running
+ if (!this.consumerImpl.isRunning()) {
+ throw new ClientException(500, 'Consumer is not running');
+ }
+
+ // Not subscribed
+ if (!this.liteTopicSet.has(liteTopic)) {
+ return;
+ }
+
+ try {
+ await this.syncLiteSubscription(
+ LiteSubscriptionAction.PARTIAL_REMOVE,
+ [ liteTopic ],
+ null,
+ );
+
+ this.liteTopicSet.delete(liteTopic);
+ this.consumerImpl.getLogger().info(
+ 'UnsubscribeLite %s, topic=%s, group=%s, clientId=%s',
+ liteTopic,
+ this.getBindTopicName(),
+ this.getConsumerGroupName(),
+ this.consumerImpl.clientId,
+ );
+ } catch (error) {
+ this.consumerImpl.getLogger().error(
+ 'Failed to unsubscribeLite %s, error=%s',
+ liteTopic,
+ error,
+ );
+ throw error;
+ }
+ }
+
+ /**
+ * Periodic sync of all lite subscriptions.
+ */
+ private async syncAllLiteSubscription() {
+ try {
+ this.checkLiteSubscriptionQuota(0);
+ await this.syncLiteSubscription(
+ LiteSubscriptionAction.COMPLETE_ADD,
+ Array.from(this.liteTopicSet),
+ null,
+ );
+ } catch (error) {
+ this.consumerImpl.getLogger().error(
+ 'Schedule syncAllLiteSubscription error, clientId=%s, error=%s',
+ this.consumerImpl.clientId,
+ error,
+ );
+ }
+ }
+
+ /**
+ * Sync lite subscription with server.
+ */
+ private async syncLiteSubscription(
+ action: LiteSubscriptionAction,
+ liteTopics: string[],
+ offsetOption: OffsetOption | null,
+ ): Promise<void> {
+ const logger = this.consumerImpl.getLogger();
+ if (logger.debug) {
+ logger.debug(
+ 'SyncLiteSubscription: action=%s, liteTopics=[%s], offsetOption=%s',
+ LiteSubscriptionAction[action],
+ liteTopics.join(', '),
+ offsetOption?.toString() ?? 'null',
+ );
+ }
+
+ // Build the request
+ const request = new SyncLiteSubscriptionRequest()
+ .setAction(action)
+ .setTopic(this.bindTopic.toProtobuf())
+ .setGroup(this.group.toProtobuf())
+ .setLiteTopicSetList(liteTopics);
+
+ // Set offset option if provided
+ if (offsetOption) {
+ 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(),
+ );
+
+ // 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,
+ );
+ }
+ } catch (error) {
+ if (logger.error) {
+ logger.error(
+ 'SyncLiteSubscription failed: action=%s, liteTopics=[%s], clientId=%s, error=%s',
+ LiteSubscriptionAction[action],
+ liteTopics.join(', '),
+ this.consumerImpl.clientId,
+ error,
+ );
+ }
+ throw error;
+ }
+ }
+
+ /**
+ * Handle notify unsubscribe lite command from server.
+ */
+ public onNotifyUnsubscribeLiteCommand(command: NotifyUnsubscribeLiteCommand) {
+ const liteTopic = command.getLiteTopic();
+ this.consumerImpl.getLogger().info(
+ 'Notify unsubscribe lite: liteTopic=%s, group=%s, bindTopic=%s',
+ liteTopic,
+ this.getConsumerGroupName(),
+ this.getBindTopicName(),
+ );
+
+ if (liteTopic && liteTopic.trim().length > 0) {
+ this.liteTopicSet.delete(liteTopic);
+ }
+ }
+
+ /**
+ * Validate lite topic name.
+ */
+ private validateLiteTopic(liteTopic: string, maxLength?: number) {
+ const maxLen = maxLength ?? this.maxLiteTopicSize;
+
+ if (!liteTopic || liteTopic.trim().length === 0) {
+ throw new Error('liteTopic is blank');
+ }
+
+ if (liteTopic.length > maxLen) {
+ throw new Error(
+ `liteTopic length exceeded max length ${maxLen}, liteTopic: ${liteTopic}`,
+ );
+ }
+ }
+
+ /**
+ * Check lite subscription quota.
+ */
+ private checkLiteSubscriptionQuota(delta: number) {
+ if (this.liteTopicSet.size + delta > this.liteSubscriptionQuota) {
+ throw new ClientException(
+ 429,
+ `Lite subscription quota exceeded ${this.liteSubscriptionQuota}`,
+ );
+ }
+ }
+
+ /**
+ * Returns a string representation of this LiteSubscriptionManager.
+ *
+ * @return String representation
+ */
+ toString(): string {
+ return `LiteSubscriptionManager{bindTopic=${this.bindTopic.getName()}, group=${this.group.getName()}, liteTopicSet=[${Array.from(this.liteTopicSet).join(', ')}], liteSubscriptionQuota=${this.liteSubscriptionQuota}, maxLiteTopicSize=${this.maxLiteTopicSize}}`;
+ }
+}
diff --git a/nodejs/src/consumer/OffsetOption.ts b/nodejs/src/consumer/OffsetOption.ts
new file mode 100644
index 00000000..5bb5ee9f
--- /dev/null
+++ b/nodejs/src/consumer/OffsetOption.ts
@@ -0,0 +1,202 @@
+/**
+ * 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.
+ */
+
+/**
+ * OffsetOption type enum.
+ */
+export enum OffsetType {
+ POLICY = 0,
+ OFFSET = 1,
+ TAIL_N = 2,
+ TIMESTAMP = 3,
+}
+
+/**
+ * ProtoBuf OffsetOption type.
+ */
+import { OffsetOption as ProtoOffsetOption } from '../../proto/apache/rocketmq/v2/definition_pb';
+
+/**
+ * OffsetOption for specifying consume from offset.
+ *
+ * <p>OffsetOption is used to specify the starting point for message consumption.
+ * It supports various policies and custom offsets.</p>
+ */
+export class OffsetOption {
+ /** Policy constant: last offset */
+ static readonly POLICY_LAST_VALUE = BigInt(0);
+
+ /** Policy constant: min offset */
+ static readonly POLICY_MIN_VALUE = BigInt(1);
+
+ /** Policy constant: max offset */
+ static readonly POLICY_MAX_VALUE = BigInt(2);
+
+ /** Predefined option: LAST_OFFSET */
+ static readonly LAST_OFFSET = new OffsetOption(OffsetType.POLICY, OffsetOption.POLICY_LAST_VALUE);
+
+ /** Predefined option: MIN_OFFSET */
+ static readonly MIN_OFFSET = new OffsetOption(OffsetType.POLICY, OffsetOption.POLICY_MIN_VALUE);
+
+ /** Predefined option: MAX_OFFSET */
+ static readonly MAX_OFFSET = new OffsetOption(OffsetType.POLICY, OffsetOption.POLICY_MAX_VALUE);
+
+ private readonly type: OffsetType;
+ private readonly value: bigint;
+
+ private constructor(type: OffsetType, value: bigint) {
+ this.type = type;
+ this.value = value;
+ }
+
+ /**
+ * Create an OffsetOption from a specific offset.
+ *
+ * @param offset - The offset value (must be >= 0)
+ * @return OffsetOption instance
+ * @throws Error if offset is negative
+ */
+ static ofOffset(offset: number | bigint): OffsetOption {
+ const value = typeof offset === 'number' ? BigInt(offset) : offset;
+ if (value < BigInt(0)) {
+ throw new Error('offset must be greater than or equal to 0');
+ }
+ return new OffsetOption(OffsetType.OFFSET, value);
+ }
+
+ /**
+ * Create an OffsetOption from tail N messages.
+ *
+ * @param tailN - Number of messages from the tail (must be >= 0)
+ * @return OffsetOption instance
+ * @throws Error if tailN is negative
+ */
+ static ofTailN(tailN: number | bigint): OffsetOption {
+ const value = typeof tailN === 'number' ? BigInt(tailN) : tailN;
+ if (value < BigInt(0)) {
+ throw new Error('tailN must be greater than or equal to 0');
+ }
+ return new OffsetOption(OffsetType.TAIL_N, value);
+ }
+
+ /**
+ * Create an OffsetOption from a timestamp.
+ *
+ * @param timestamp - Unix timestamp in milliseconds (must be >= 0)
+ * @return OffsetOption instance
+ * @throws Error if timestamp is negative
+ */
+ static ofTimestamp(timestamp: number | bigint): OffsetOption {
+ const value = typeof timestamp === 'number' ? BigInt(timestamp) : timestamp;
+ if (value < BigInt(0)) {
+ throw new Error('timestamp must be greater than or equal to 0');
+ }
+ return new OffsetOption(OffsetType.TIMESTAMP, value);
+ }
+
+ /**
+ * Get the type of this OffsetOption.
+ *
+ * @return OffsetType
+ */
+ getType(): OffsetType {
+ return this.type;
+ }
+
+ /**
+ * Get the value of this OffsetOption.
+ *
+ * @return Value as bigint
+ */
+ getValue(): bigint {
+ return this.value;
+ }
+
+ /**
+ * Returns a string representation of this OffsetOption.
+ *
+ * @return String representation
+ */
+ toString(): string {
+ return `OffsetOption{type=${OffsetType[this.type]}, value=${this.value}}`;
+ }
+
+ /**
+ * Validate that a bigint value is within the safe number range.
+ *
+ * @param value - The bigint value to validate
+ * @throws Error if value is out of safe number range
+ */
+ private validateSafeNumberRange(value: bigint): void {
+ if (value > BigInt(Number.MAX_SAFE_INTEGER) || value < BigInt(Number.MIN_SAFE_INTEGER)) {
+ throw new Error(`Value ${value} is out of safe number range`);
+ }
+ }
+
+ /**
+ * Convert this OffsetOption to ProtoBuf OffsetOption.
+ *
+ * @return ProtoBuf OffsetOption instance
+ */
+ toProtobuf(): ProtoOffsetOption {
+ const proto = new ProtoOffsetOption();
+ switch (this.type) {
+ case OffsetType.POLICY: {
+ // Map our policy values to ProtoBuf Policy enum
+ // POLICY_LAST_VALUE (0) -> Policy.LAST (0)
+ // POLICY_MIN_VALUE (1) -> Policy.MIN (1)
+ // POLICY_MAX_VALUE (2) -> Policy.MAX (2)
+ this.validateSafeNumberRange(this.value);
+ proto.setPolicy(Number(this.value));
+ break;
+ }
+ case OffsetType.OFFSET: {
+ this.validateSafeNumberRange(this.value);
+ proto.setOffset(Number(this.value));
+ break;
+ }
+ case OffsetType.TAIL_N: {
+ this.validateSafeNumberRange(this.value);
+ proto.setTailN(Number(this.value));
+ break;
+ }
+ case OffsetType.TIMESTAMP: {
+ this.validateSafeNumberRange(this.value);
+ proto.setTimestamp(Number(this.value));
+ break;
+ }
+ default:
+ throw new Error(`Unknown OffsetType: ${this.type}`);
+ }
+ return proto;
+ }
+
+ equals(other: OffsetOption): boolean {
+ if (this === other) return true;
+ if (!other) return false;
+ return this.type === other.type && this.value === other.value;
+ }
+
+ hashCode(): number {
+ let hash = 17;
+ hash = hash * 31 + this.type;
+ // Convert bigint to number for hash calculation
+ const valueNum = Number(this.value);
+ hash = hash * 31 + valueNum;
+ return hash;
+ }
+}
diff --git a/nodejs/src/consumer/ProcessQueue.ts b/nodejs/src/consumer/ProcessQueue.ts
index 65eaca7b..3d21f802 100644
--- a/nodejs/src/consumer/ProcessQueue.ts
+++ b/nodejs/src/consumer/ProcessQueue.ts
@@ -88,6 +88,8 @@ export class ProcessQueue {
}
fetchMessageImmediately(): void {
+ (this.#consumer as any).logger?.info('ProcessQueue.fetchMessageImmediately called, mq=%s@%s@%d',
+ this.#mq.topic.name, this.#mq.broker.name, this.#mq.queueId);
this.#receiveMessageImmediately();
}
@@ -136,14 +138,25 @@ export class ProcessQueue {
);
this.#activityTime = Date.now();
+ (this.#consumer as any).logger?.debug?.('Start to receive message, mq=%s@%s@%d, batchSize=%d, attemptId=%s',
+ this.#mq.topic.name, this.#mq.broker.name, this.#mq.queueId, batchSize, attemptId);
+
this.#consumer.receiveMessage(request, this.#mq, longPollingTimeout)
.then(messages => {
+ if (messages.length > 0) {
+ (this.#consumer as any).logger?.debug?.('Received messages, count=%d, mq=%s@%s@%d',
+ messages.length, this.#mq.topic.name, this.#mq.broker.name, this.#mq.queueId);
+ }
this.#onReceiveMessageResult(messages);
})
.catch(err => {
+ (this.#consumer as any).logger?.error('Failed to receive message, mq=%s@%s@%d, error=%s',
+ this.#mq.topic.name, this.#mq.broker.name, this.#mq.queueId, err);
this.onReceiveMessageException(err, attemptId);
});
} catch (err) {
+ (this.#consumer as any).logger?.error('Exception in receiveMessage, mq=%s@%s@%d, error=%s',
+ this.#mq.topic.name, this.#mq.broker.name, this.#mq.queueId, err);
this.onReceiveMessageException(err as Error, attemptId);
}
}
@@ -245,7 +258,7 @@ export class ProcessQueue {
async eraseFifoMessage(messageView: MessageView, consumeResult: ConsumeResult): Promise<void> {
const retryPolicy = this.#consumer.getRetryPolicy();
const maxAttempts = retryPolicy?.getMaxAttempts() ?? 1;
- let attempt = messageView.deliveryAttempt ?? 1;
+ const attempt = messageView.deliveryAttempt ?? 1;
if (consumeResult === ConsumeResult.FAILURE && attempt < maxAttempts) {
const nextAttemptDelay = retryPolicy?.getNextAttemptDelay(attempt) ?? 0;
diff --git a/nodejs/src/consumer/PushConsumer.ts b/nodejs/src/consumer/PushConsumer.ts
index f3ba754c..62e751cb 100644
--- a/nodejs/src/consumer/PushConsumer.ts
+++ b/nodejs/src/consumer/PushConsumer.ts
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-import { ClientType } from '../../proto/apache/rocketmq/v2/definition_pb';
+import { ClientType, MessageType, Permission } from '../../proto/apache/rocketmq/v2/definition_pb';
import {
AckMessageRequest,
ChangeInvisibleDurationRequest,
@@ -26,7 +26,8 @@ import {
ReceiveMessageRequest,
} from '../../proto/apache/rocketmq/v2/service_pb';
import { MessageView } from '../message';
-import { MessageQueue, TopicRouteData } from '../route';
+import { Broker, MessageQueue, TopicRouteData } from '../route';
+import { MessageQueue as MessageQueuePB, Broker as BrokerPB } from '../../proto/apache/rocketmq/v2/definition_pb';
import { StatusChecker } from '../exception';
import { RetryPolicy } from '../retry';
import { createDuration, createResource } from '../util';
@@ -91,7 +92,7 @@ export class PushConsumer extends Consumer {
this.#enableFifoConsumeAccelerator = options.enableFifoConsumeAccelerator ?? false;
this.#pushSubscriptionSettings = new PushSubscriptionSettings(
- options.namespace, this.clientId, this.endpoints,
+ options.namespace, this.clientId, this.getClientType(), this.endpoints,
this.consumerGroup, this.requestTimeout, this.#subscriptionExpressions,
);
}
@@ -104,9 +105,7 @@ export class PushConsumer extends Consumer {
try {
this.#consumeService = this.#createConsumeService();
// Start scanning assignments periodically
- this.logger.info('Starting assignment scanning, clientId=%s', this.clientId);
- setTimeout(() => this.#scanAssignments(), ASSIGNMENT_SCAN_SCHEDULE_DELAY);
- this.#scanAssignmentTimer = setInterval(() => this.#scanAssignments(), ASSIGNMENT_SCAN_SCHEDULE_PERIOD);
+ this.startAssignmentScanning();
this.logger.info('Push consumer started successfully, clientId=%s', this.clientId);
} catch (err) {
this.logger.error('Failed to start push consumer, cleaning up resources, clientId=%s, error=%s',
@@ -120,6 +119,15 @@ export class PushConsumer extends Consumer {
}
}
+ /**
+ * Start assignment scanning. Can be overridden by subclasses.
+ */
+ protected startAssignmentScanning() {
+ this.logger.info('Starting assignment scanning, clientId=%s', this.clientId);
+ setTimeout(() => this.#scanAssignments(), ASSIGNMENT_SCAN_SCHEDULE_DELAY);
+ this.#scanAssignmentTimer = setInterval(() => this.#scanAssignments(), ASSIGNMENT_SCAN_SCHEDULE_PERIOD);
+ }
+
async shutdown() {
this.logger.info('Begin to shutdown the rocketmq push consumer, clientId=%s', this.clientId);
// Stop scanning assignments
@@ -148,9 +156,18 @@ export class PushConsumer extends Consumer {
return this.#pushSubscriptionSettings;
}
+ /**
+ * Get the client type for push consumer.
+ *
+ * @return The client type identifier
+ */
+ protected getClientType(): ClientType {
+ return ClientType.PUSH_CONSUMER;
+ }
+
protected wrapHeartbeatRequest() {
return new HeartbeatRequest()
- .setClientType(ClientType.PUSH_CONSUMER)
+ .setClientType(this.getClientType())
.setGroup(createResource(this.consumerGroup));
}
@@ -286,30 +303,50 @@ export class PushConsumer extends Consumer {
const request = new AckMessageRequest()
.setGroup(createResource(this.consumerGroup))
.setTopic(createResource(messageView.topic));
- request.addEntries()
+ const entry = request.addEntries()
.setMessageId(messageView.messageId)
.setReceiptHandle(messageView.receiptHandle);
+
+ // For lite consumers, must set liteTopic in ACK request
+ if (this.isLiteConsumer() && messageView.liteTopic) {
+ entry.setLiteTopic(messageView.liteTopic);
+ }
+
return request;
}
wrapChangeInvisibleDurationRequest(messageView: MessageView, invisibleDuration: number) {
- return new ChangeInvisibleDurationRequest()
+ const request = new ChangeInvisibleDurationRequest()
.setGroup(createResource(this.consumerGroup))
.setTopic(createResource(messageView.topic))
.setReceiptHandle(messageView.receiptHandle)
.setInvisibleDuration(createDuration(invisibleDuration))
.setMessageId(messageView.messageId);
+
+ // For lite consumers, must set liteTopic in request
+ if (this.isLiteConsumer() && messageView.liteTopic) {
+ request.setLiteTopic(messageView.liteTopic);
+ }
+
+ return request;
}
wrapForwardMessageToDeadLetterQueueRequest(messageView: MessageView) {
const retryPolicy = this.getRetryPolicy();
- return new ForwardMessageToDeadLetterQueueRequest()
+ const request = new ForwardMessageToDeadLetterQueueRequest()
.setGroup(createResource(this.consumerGroup))
.setTopic(createResource(messageView.topic))
.setReceiptHandle(messageView.receiptHandle)
.setMessageId(messageView.messageId)
.setDeliveryAttempt(messageView.deliveryAttempt ?? 0)
.setMaxDeliveryAttempts(retryPolicy?.getMaxAttempts() ?? 1);
+
+ // For lite consumers, must set liteTopic in request
+ if (this.isLiteConsumer() && messageView.liteTopic) {
+ request.setLiteTopic(messageView.liteTopic);
+ }
+
+ return request;
}
// --- Internal: queue size and cache threshold ---
@@ -470,4 +507,57 @@ export class PushConsumer extends Consumer {
this.#processQueueTable.set(mqKey, { mq, pq: processQueue });
return processQueue;
}
+
+ /**
+ * Create a virtual process queue for Lite consumer.
+ * Lite consumers use a special message queue with queueId=-1 to receive messages.
+ * This method creates the virtual queue and starts fetching messages immediately.
+ *
+ * @param topic - The bind topic name
+ * @param filterExpression - The filter expression for message filtering
+ */
+ protected createVirtualProcessQueueForLite(topic: string, filterExpression: FilterExpression) {
+ // Validate parameters
+ if (!topic || topic.trim().length === 0) {
+ throw new Error('topic should not be blank');
+ }
+ if (!filterExpression) {
+ throw new Error('filterExpression should not be null');
+ }
+
+ // Create a virtual broker with default endpoints
+ const brokerPb = new BrokerPB();
+ brokerPb.setName('virtual-broker');
+ brokerPb.setId(0);
+ brokerPb.setEndpoints(this.endpoints.toProtobuf());
+ const broker = new Broker(brokerPb.toObject());
+
+ // Create a virtual message queue with queueId=-1
+ const virtualMqPb = new MessageQueuePB();
+ virtualMqPb.setId(-1); // Virtual queue ID for Lite consumer
+ virtualMqPb.setTopic(createResource(topic));
+ virtualMqPb.setBroker(broker.toProtobuf());
+ virtualMqPb.setPermission(Permission.READ);
+ virtualMqPb.setAcceptMessageTypesList([ MessageType.LITE ]);
+
+ const virtualMq = new MessageQueue(virtualMqPb);
+
+ // Create process queue for the virtual message queue
+ const mqKey = this.#mqKey(virtualMq);
+ if (this.#processQueueTable.has(mqKey)) {
+ this.logger.info('Virtual process queue already exists, mq=%s, clientId=%s', mqKey, this.clientId);
+ return;
+ }
+
+ const processQueue = new ProcessQueue(this, virtualMq, filterExpression);
+ this.#processQueueTable.set(mqKey, { mq: virtualMq, pq: processQueue });
+
+ this.logger.info('Created virtual process queue for Lite consumer, mq=%s, clientId=%s',
+ mqKey, this.clientId);
+
+ // Start fetching messages immediately
+ this.logger.info('Start to fetch lite messages from virtual queue, mq=%s, clientId=%s',
+ mqKey, this.clientId);
+ processQueue.fetchMessageImmediately();
+ }
}
diff --git a/nodejs/src/consumer/PushSubscriptionSettings.ts b/nodejs/src/consumer/PushSubscriptionSettings.ts
index fe97f6b1..d02ca277 100644
--- a/nodejs/src/consumer/PushSubscriptionSettings.ts
+++ b/nodejs/src/consumer/PushSubscriptionSettings.ts
@@ -37,13 +37,14 @@ export class PushSubscriptionSettings extends Settings {
constructor(
namespace: string,
clientId: string,
+ clientType: ClientType,
accessPoint: Endpoints,
consumerGroup: string,
requestTimeout: number,
subscriptionExpressions: Map<string, FilterExpression>,
longPollingTimeout?: number,
) {
- super(namespace, clientId, ClientType.PUSH_CONSUMER, accessPoint, requestTimeout);
+ super(namespace, clientId, clientType, accessPoint, requestTimeout);
this.#group = consumerGroup;
this.#subscriptionExpressions = subscriptionExpressions;
if (longPollingTimeout !== undefined) {
diff --git a/nodejs/src/consumer/SimpleConsumer.ts b/nodejs/src/consumer/SimpleConsumer.ts
index 994ae957..1ba6654d 100644
--- a/nodejs/src/consumer/SimpleConsumer.ts
+++ b/nodejs/src/consumer/SimpleConsumer.ts
@@ -24,8 +24,11 @@ import { FilterExpression } from './FilterExpression';
import { SimpleSubscriptionSettings } from './SimpleSubscriptionSettings';
import { SubscriptionLoadBalancer } from './SubscriptionLoadBalancer';
import { Consumer, ConsumerOptions } from './Consumer';
+import { InternalErrorException } from '../exception';
const RANDOM_INDEX = Math.floor(Math.random() * Number.MAX_SAFE_INTEGER);
+const DEFAULT_MAX_RETRY_ATTEMPTS = 3;
+const RETRY_DELAY_MS = 1000;
export interface SimpleConsumerOptions extends ConsumerOptions {
/**
@@ -41,6 +44,10 @@ export interface SimpleConsumerOptions extends ConsumerOptions {
* 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;
}
export class SimpleConsumer extends Consumer {
@@ -48,6 +55,7 @@ export class SimpleConsumer extends Consumer {
readonly #subscriptionExpressions = new Map<string, FilterExpression>();
readonly #subscriptionRouteDataCache = new Map<string, SubscriptionLoadBalancer>();
readonly #awaitDuration: number;
+ readonly #maxRetryAttempts: number;
#topicIndex = RANDOM_INDEX;
constructor(options: SimpleConsumerOptions) {
@@ -62,17 +70,28 @@ export class SimpleConsumer extends Consumer {
}
}
this.#awaitDuration = options.awaitDuration ?? 30000;
- this.#simpleSubscriptionSettings = new SimpleSubscriptionSettings(options.namespace, this.clientId, this.endpoints,
- this.consumerGroup, this.requestTimeout, this.#awaitDuration, this.#subscriptionExpressions);
+ this.#maxRetryAttempts = options.maxRetryAttempts ?? DEFAULT_MAX_RETRY_ATTEMPTS;
+ this.#simpleSubscriptionSettings = new SimpleSubscriptionSettings(options.namespace, this.clientId,
+ this.getClientType(), this.endpoints, this.consumerGroup, this.requestTimeout,
+ this.#awaitDuration, this.#subscriptionExpressions);
}
protected getSettings() {
return this.#simpleSubscriptionSettings;
}
+ /**
+ * Get the client type.
+ *
+ * @return The client type identifier for simple consumer
+ */
+ protected getClientType(): ClientType {
+ return ClientType.SIMPLE_CONSUMER;
+ }
+
protected wrapHeartbeatRequest() {
return new HeartbeatRequest()
- .setClientType(ClientType.SIMPLE_CONSUMER)
+ .setClientType(this.getClientType())
.setGroup(createResource(this.consumerGroup));
}
@@ -129,7 +148,32 @@ export class SimpleConsumer extends Consumer {
const mq = loadBalancer.takeMessageQueue();
const request = this.wrapReceiveMessageRequest(maxMessageNum, mq, filterExpression,
invisibleDuration, this.#awaitDuration);
- return await this.receiveMessage(request, mq, this.#awaitDuration);
+
+ // Retry logic for temporary errors
+ let lastError: Error | undefined;
+ for (let attempt = 1; attempt <= this.#maxRetryAttempts; attempt++) {
+ try {
+ return await this.receiveMessage(request, mq, this.#awaitDuration);
+ } catch (err) {
+ lastError = err as Error;
+
+ // Only retry on internal server errors (temporary errors)
+ if (err instanceof InternalErrorException) {
+ if (attempt < this.#maxRetryAttempts) {
+ this.logger.warn('Received temporary error from server, will retry after %dms, attempt=%d/%d, topic=%s, clientId=%s, error=%s',
+ RETRY_DELAY_MS, attempt, this.#maxRetryAttempts, topic, this.clientId, err.message);
+ await new Promise(resolve => setTimeout(resolve, RETRY_DELAY_MS));
+ continue;
+ }
+ }
+
+ // For non-retryable errors or max attempts reached, throw immediately
+ throw err;
+ }
+ }
+
+ // This should never be reached, but just in case
+ throw lastError;
}
async ack(message: MessageView) {
diff --git a/nodejs/src/consumer/SimpleSubscriptionSettings.ts b/nodejs/src/consumer/SimpleSubscriptionSettings.ts
index f12c4e28..0b0645b4 100644
--- a/nodejs/src/consumer/SimpleSubscriptionSettings.ts
+++ b/nodejs/src/consumer/SimpleSubscriptionSettings.ts
@@ -30,8 +30,8 @@ export class SimpleSubscriptionSettings extends Settings {
readonly group: string;
readonly subscriptionExpressions: Map<string, FilterExpression>;
- constructor(namespace: string, clientId: string, accessPoint: Endpoints, group: string, requestTimeout: number, longPollingTimeout: number, subscriptionExpressions: Map<string, FilterExpression>) {
- super(namespace, clientId, ClientType.SIMPLE_CONSUMER, accessPoint, requestTimeout);
+ constructor(namespace: string, clientId: string, clientType: ClientType, accessPoint: Endpoints, group: string, requestTimeout: number, longPollingTimeout: number, subscriptionExpressions: Map<string, FilterExpression>) {
+ super(namespace, clientId, clientType, accessPoint, requestTimeout);
this.longPollingTimeout = longPollingTimeout;
this.group = group;
this.subscriptionExpressions = subscriptionExpressions;
diff --git a/nodejs/src/consumer/StandardConsumeService.ts b/nodejs/src/consumer/StandardConsumeService.ts
index 3121f210..bd507902 100644
--- a/nodejs/src/consumer/StandardConsumeService.ts
+++ b/nodejs/src/consumer/StandardConsumeService.ts
@@ -19,24 +19,37 @@ import { MessageView } from '../message';
import { ConsumeService } from './ConsumeService';
import { MessageListener } from './MessageListener';
import type { ProcessQueue } from './ProcessQueue';
+import { getDefaultLogger } from '../client';
export class StandardConsumeService extends ConsumeService {
+ private readonly logger = getDefaultLogger();
+
+ // eslint-disable-next-line @typescript-eslint/no-useless-constructor
constructor(clientId: string, messageListener: MessageListener) {
super(clientId, messageListener);
}
consume(pq: ProcessQueue, messageViews: MessageView[]): void {
+ this.logger.debug?.('StandardConsumeService.consume called, messageCount=%d', messageViews.length);
for (const messageView of messageViews) {
if (messageView.corrupted) {
+ this.logger.warn('Message corrupted, discarding, messageId=%s', messageView.messageId);
pq.discardMessage(messageView);
continue;
}
+ this.logger.debug?.('Consuming message, messageId=%s, topic=%s',
+ messageView.messageId, messageView.topic);
+
this.consumeMessage(messageView)
.then(result => {
+ this.logger.debug?.('Message consumed successfully, messageId=%s, result=%s',
+ messageView.messageId, result);
pq.eraseMessage(messageView, result);
})
- .catch(() => {
+ .catch(err => {
+ this.logger.error('Message consumption failed, messageId=%s, error=%s',
+ messageView.messageId, err);
// Should never reach here.
});
}
diff --git a/nodejs/src/consumer/index.ts b/nodejs/src/consumer/index.ts
index 8b1cfc1c..07e472a3 100644
--- a/nodejs/src/consumer/index.ts
+++ b/nodejs/src/consumer/index.ts
@@ -31,3 +31,9 @@ export * from './StandardConsumeService';
export * from './FifoConsumeService';
export * from './ProcessQueue';
export * from './PushConsumer';
+export * from './LitePushConsumer';
+export * from './LiteSimpleConsumer';
+export * from './OffsetOption';
+export { LitePushConsumerImpl } from './LitePushConsumerImpl';
+export { LiteSimpleConsumerImpl } from './LiteSimpleConsumerImpl';
+export { LiteSubscriptionManager } from './LiteSubscriptionManager';
diff --git a/nodejs/src/exception/InternalErrorException.ts b/nodejs/src/exception/InternalErrorException.ts
index accb043a..1f360734 100644
--- a/nodejs/src/exception/InternalErrorException.ts
+++ b/nodejs/src/exception/InternalErrorException.ts
@@ -21,5 +21,15 @@ export class InternalErrorException extends ClientException {
constructor(code: number, message: string, requestId?: string) {
super(code, message, requestId);
this.name = 'InternalErrorException';
+ // Mark this as a potentially retryable error (temporary server-side issue)
+ (this as any).retryable = true;
+ }
+
+ /**
+ * Check if this error is retryable.
+ * Internal server errors are typically temporary and can be retried.
+ */
+ isRetryable(): boolean {
+ return true;
}
}
diff --git a/nodejs/src/client/ClientId.ts b/nodejs/src/exception/LiteSubscriptionQuotaExceededException.ts
similarity index 67%
copy from nodejs/src/client/ClientId.ts
copy to nodejs/src/exception/LiteSubscriptionQuotaExceededException.ts
index 76cbfd19..046e656d 100644
--- a/nodejs/src/client/ClientId.ts
+++ b/nodejs/src/exception/LiteSubscriptionQuotaExceededException.ts
@@ -15,17 +15,14 @@
* limitations under the License.
*/
-import { hostname } from 'node:os';
+import { ClientException } from '../exception/ClientException';
/**
- * Client Identifier Helper
- * https://github.com/apache/rocketmq-clients/blob/master/docs/design.md#client-identifier
+ * Exception thrown when lite subscription quota is exceeded.
*/
-export class ClientId {
- static #hostname = hostname();
- static #index = 0n;
-
- static create() {
- return `${this.#hostname}@${process.pid}@${this.#index++}@${Date.now().toString(36)}`;
+export class LiteSubscriptionQuotaExceededException extends ClientException {
+ constructor(responseCode: number, requestId: string | null, message: string) {
+ super(responseCode, message, requestId ?? undefined);
+ this.name = 'LiteSubscriptionQuotaExceededException';
}
}
diff --git a/nodejs/src/client/ClientId.ts b/nodejs/src/exception/LiteTopicQuotaExceededException.ts
similarity index 68%
copy from nodejs/src/client/ClientId.ts
copy to nodejs/src/exception/LiteTopicQuotaExceededException.ts
index 76cbfd19..1616655a 100644
--- a/nodejs/src/client/ClientId.ts
+++ b/nodejs/src/exception/LiteTopicQuotaExceededException.ts
@@ -15,17 +15,14 @@
* limitations under the License.
*/
-import { hostname } from 'node:os';
+import { ClientException } from '../exception/ClientException';
/**
- * Client Identifier Helper
- * https://github.com/apache/rocketmq-clients/blob/master/docs/design.md#client-identifier
+ * Exception thrown when lite topic quota is exceeded during message sending.
*/
-export class ClientId {
- static #hostname = hostname();
- static #index = 0n;
-
- static create() {
- return `${this.#hostname}@${process.pid}@${this.#index++}@${Date.now().toString(36)}`;
+export class LiteTopicQuotaExceededException extends ClientException {
+ constructor(responseCode: number, message: string, requestId?: string) {
+ super(responseCode, message, requestId);
+ this.name = 'LiteTopicQuotaExceededException';
}
}
diff --git a/nodejs/src/exception/StatusChecker.ts b/nodejs/src/exception/StatusChecker.ts
index d2f08a93..bb7cdae4 100644
--- a/nodejs/src/exception/StatusChecker.ts
+++ b/nodejs/src/exception/StatusChecker.ts
@@ -19,6 +19,7 @@ import { Status, Code } from '../../proto/apache/rocketmq/v2/definition_pb';
import { BadRequestException } from './BadRequestException';
import { ForbiddenException } from './ForbiddenException';
import { InternalErrorException } from './InternalErrorException';
+import { LiteTopicQuotaExceededException } from './LiteTopicQuotaExceededException';
import { NotFoundException } from './NotFoundException';
import { PayloadTooLargeException } from './PayloadTooLargeException';
import { PaymentRequiredException } from './PaymentRequiredException';
@@ -87,6 +88,8 @@ export class StatusChecker {
case Code.VERSION_UNSUPPORTED:
case Code.VERIFY_FIFO_MESSAGE_UNSUPPORTED:
throw new UnsupportedException(status.code, status.message, requestId);
+ case Code.LITE_TOPIC_QUOTA_EXCEEDED:
+ throw new LiteTopicQuotaExceededException(status.code, status.message || '', requestId);
default:
throw new UnsupportedException(status.code, status.message, requestId);
}
diff --git a/nodejs/src/exception/index.ts b/nodejs/src/exception/index.ts
index a40b59f4..e542e96e 100644
--- a/nodejs/src/exception/index.ts
+++ b/nodejs/src/exception/index.ts
@@ -19,6 +19,8 @@ export * from './BadRequestException';
export * from './ClientException';
export * from './ForbiddenException';
export * from './InternalErrorException';
+export * from './LiteSubscriptionQuotaExceededException';
+export * from './LiteTopicQuotaExceededException';
export * from './NotFoundException';
export * from './PayloadTooLargeException';
export * from './PaymentRequiredException';
diff --git a/nodejs/src/index.ts b/nodejs/src/index.ts
index c36d39fa..28ec5f3b 100644
--- a/nodejs/src/index.ts
+++ b/nodejs/src/index.ts
@@ -21,3 +21,4 @@ export * from './message';
export * from './producer';
export * from './retry';
export * from './route';
+export * from './client';
diff --git a/nodejs/src/message/Message.ts b/nodejs/src/message/Message.ts
index 7d021810..e50c074e 100644
--- a/nodejs/src/message/Message.ts
+++ b/nodejs/src/message/Message.ts
@@ -22,9 +22,9 @@ export interface MessageOptions {
messageGroup?: string;
keys?: string[];
properties?: Map<string, string>;
- delay?: number;
deliveryTimestamp?: Date;
priority?: number;
+ liteTopic?: string;
}
export class Message {
@@ -36,9 +36,10 @@ export class Message {
properties?: Map<string, string>;
deliveryTimestamp?: Date;
priority?: number;
+ liteTopic?: string;
constructor(options: MessageOptions) {
- // Validate priority and mutual exclusivity
+ // Validate mutual exclusivity - priority
if (options.priority !== undefined) {
if (options.priority < 0) {
throw new Error('priority must be greater than or equal to 0');
@@ -49,6 +50,26 @@ export class Message {
if (options.messageGroup) {
throw new Error('priority and messageGroup should not be set at same time');
}
+ if (options.liteTopic) {
+ throw new Error('priority and liteTopic should not be set at same time');
+ }
+ }
+
+ // Validate mutual exclusivity - liteTopic
+ if (options.liteTopic !== undefined) {
+ if (options.deliveryTimestamp) {
+ throw new Error('liteTopic and deliveryTimestamp should not be set at same time');
+ }
+ if (options.messageGroup) {
+ throw new Error('liteTopic and messageGroup should not be set at same time');
+ }
+ }
+
+ // Validate mutual exclusivity - deliveryTimestamp
+ if (options.deliveryTimestamp !== undefined) {
+ if (options.messageGroup) {
+ throw new Error('deliveryTimestamp and messageGroup should not be set at same time');
+ }
}
this.topic = options.topic;
@@ -57,11 +78,17 @@ export class Message {
this.messageGroup = options.messageGroup;
this.keys = options.keys ?? [];
this.properties = options.properties;
- let deliveryTimestamp = options.deliveryTimestamp;
- if (options.delay && !deliveryTimestamp) {
- deliveryTimestamp = new Date(Date.now() + options.delay);
- }
- this.deliveryTimestamp = deliveryTimestamp;
+ this.deliveryTimestamp = options.deliveryTimestamp;
this.priority = options.priority;
+ this.liteTopic = options.liteTopic;
+ }
+
+ /**
+ * Returns a string representation of the message.
+ */
+ toString(): string {
+ return `Message{topic='${this.topic}', tag=${this.tag}, messageGroup=${this.messageGroup}, ` +
+ `liteTopic=${this.liteTopic}, priority=${this.priority}, deliveryTimestamp=${this.deliveryTimestamp}, ` +
+ `keys=[${this.keys.join(', ')}], properties=${JSON.stringify(this.properties)}}`;
}
}
diff --git a/nodejs/src/message/MessageId.ts b/nodejs/src/message/MessageId.ts
index c6523309..1afb9e7a 100644
--- a/nodejs/src/message/MessageId.ts
+++ b/nodejs/src/message/MessageId.ts
@@ -53,6 +53,24 @@ export class MessageId {
toString() {
return this.id;
}
+
+ equals(other: MessageId): boolean {
+ if (this === other) return true;
+ if (!other) return false;
+ return this.id === other.id;
+ }
+
+ hashCode(): number {
+ let hash = 0;
+ for (let i = 0; i < this.id.length; i++) {
+ const char = this.id.charCodeAt(i);
+ // eslint-disable-next-line no-bitwise
+ hash = ((hash << 5) - hash) + char;
+ // eslint-disable-next-line no-bitwise
+ hash = hash & hash; // Convert to 32bit integer
+ }
+ return hash;
+ }
}
const MAX_UINT32 = 0xFFFFFFFF;
diff --git a/nodejs/src/message/MessageView.ts b/nodejs/src/message/MessageView.ts
index d2aad738..7c191611 100644
--- a/nodejs/src/message/MessageView.ts
+++ b/nodejs/src/message/MessageView.ts
@@ -42,6 +42,8 @@ export class MessageView {
readonly offset?: number;
readonly decodeTimestamp: Date;
readonly properties = new Map<string, string>();
+ readonly liteTopic?: string;
+ readonly priority?: number;
constructor(message: MessagePB, messageQueue?: MessageQueue, transportDeliveryTimestamp?: Date) {
const systemProperties = message.getSystemProperties()!;
@@ -86,9 +88,134 @@ export class MessageView {
this.offset = systemProperties.getQueueOffset();
this.receiptHandle = systemProperties.getReceiptHandle()!;
this.transportDeliveryTimestamp = transportDeliveryTimestamp;
+ this.liteTopic = systemProperties.getLiteTopic();
+ this.priority = systemProperties.getPriority();
if (messageQueue) {
this.endpoints = messageQueue.broker.endpoints;
}
this.body = bodyBytes;
}
+
+ /**
+ * Get the lite topic, which makes sense only when the topic type is LITE.
+ *
+ * @return lite topic, which is optional, undefined means lite topic is not specified.
+ */
+ getLiteTopic(): string | undefined {
+ return this.liteTopic;
+ }
+
+ /**
+ * Get the priority of the message, which makes sense only when topic type is priority.
+ *
+ * @return message priority, which is optional, undefined means priority is not specified.
+ */
+ getPriority(): number | undefined {
+ return this.priority;
+ }
+
+ /**
+ * Get the unique id of the message.
+ *
+ * @return unique id.
+ */
+ getMessageId(): string {
+ return this.messageId;
+ }
+
+ /**
+ * Get the topic of the message, which is the first classifier for the message.
+ *
+ * @return topic of the message.
+ */
+ getTopic(): string {
+ return this.topic;
+ }
+
+ /**
+ * Get the deep copy of the message body, which means any modification of the return value does not
+ * affect the built-in message body.
+ *
+ * @return the deep copy of message body.
+ */
+ getBody(): Buffer {
+ return Buffer.from(this.body);
+ }
+
+ /**
+ * Get the deep copy of message properties, which makes the modifies of return value does
+ * not affect the message itself.
+ *
+ * @return the deep copy of message properties.
+ */
+ getProperties(): Map<string, string> {
+ return new Map(this.properties);
+ }
+
+ /**
+ * Get the tag of the message, which is the second classifier besides the topic.
+ *
+ * @return the tag of message, which is optional, undefined means tag does not exist.
+ */
+ getTag(): string | undefined {
+ return this.tag;
+ }
+
+ /**
+ * Get the key collection of the message, which means any modification of the return value does not affect the
+ * built-in message key collection.
+ *
+ * @return copy of the key collection of the message, empty array means message key is not specified.
+ */
+ getKeys(): string[] {
+ return [ ...this.keys ];
+ }
+
+ /**
+ * Get the message group, which makes sense only when the topic type is FIFO(First In, First Out).
+ *
+ * @return message group, which is optional, undefined means message group is not specified.
+ */
+ getMessageGroup(): string | undefined {
+ return this.messageGroup;
+ }
+
+ /**
+ * Get the expected delivery timestamp, which makes sense only when the topic type is delay.
+ *
+ * @return message expected delivery timestamp, which is optional, undefined means delivery timestamp is not specified.
+ */
+ getDeliveryTimestamp(): Date | undefined {
+ return this.deliveryTimestamp;
+ }
+
+ /**
+ * Get the born host of the message.
+ *
+ * @return born host of the message.
+ */
+ getBornHost(): string {
+ return this.bornHost;
+ }
+
+ /**
+ * Get the born timestamp of the message.
+ *
+ * <p>Born time means the timestamp that the message is prepared to send rather than the timestamp the
+ * Message was built.
+ *
+ * @return born timestamp of the message.
+ */
+ getBornTimestamp(): Date | undefined {
+ return this.bornTimestamp;
+ }
+
+ /**
+ * Get the delivery attempt for the message.
+ *
+ * @return delivery attempt.
+ */
+ getDeliveryAttempt(): number | undefined {
+ return this.deliveryAttempt;
+ }
}
diff --git a/nodejs/src/message/PublishingMessage.ts b/nodejs/src/message/PublishingMessage.ts
index 959d2f28..32b504f4 100644
--- a/nodejs/src/message/PublishingMessage.ts
+++ b/nodejs/src/message/PublishingMessage.ts
@@ -40,7 +40,7 @@ export class PublishingMessage extends Message {
// Generate message id.
this.messageId = MessageIdFactory.create().toString();
// Normal message.
- if (!this.messageGroup && !this.deliveryTimestamp && !txEnabled) {
+ if (!this.messageGroup && !this.liteTopic && !this.deliveryTimestamp && !this.priority && !txEnabled) {
this.messageType = MessageType.NORMAL;
return;
}
@@ -49,19 +49,29 @@ export class PublishingMessage extends Message {
this.messageType = MessageType.FIFO;
return;
}
+ // Lite message.
+ if (this.liteTopic && !txEnabled) {
+ this.messageType = MessageType.LITE;
+ return;
+ }
// Delay message.
if (this.deliveryTimestamp && !txEnabled) {
this.messageType = MessageType.DELAY;
return;
}
+ // Priority message.
+ if (this.priority && !txEnabled) {
+ this.messageType = MessageType.PRIORITY;
+ return;
+ }
// Transaction message.
if (!this.messageGroup &&
- !this.deliveryTimestamp && txEnabled) {
+ !this.liteTopic && !this.deliveryTimestamp && !this.priority && txEnabled) {
this.messageType = MessageType.TRANSACTION;
return;
}
- // Transaction semantics is conflicted with fifo/delay.
- throw new TypeError('Transactional message should not set messageGroup or deliveryTimestamp');
+ // Transaction semantics is conflicted with fifo/delay/lite/priority.
+ throw new TypeError('Transactional message should not set messageGroup, liteTopic, deliveryTimestamp or priority');
}
/**
@@ -86,6 +96,12 @@ export class PublishingMessage extends Message {
if (this.messageGroup) {
systemProperties.setMessageGroup(this.messageGroup);
}
+ if (this.liteTopic !== undefined) {
+ systemProperties.setLiteTopic(this.liteTopic);
+ }
+ if (this.priority !== undefined) {
+ systemProperties.setPriority(this.priority);
+ }
const resource = createResource(this.topic);
resource.setResourceNamespace(namespace);
diff --git a/nodejs/src/producer/Producer.ts b/nodejs/src/producer/Producer.ts
index 753c6b00..50233b4d 100644
--- a/nodejs/src/producer/Producer.ts
+++ b/nodejs/src/producer/Producer.ts
@@ -160,9 +160,18 @@ export class Producer extends BaseClient {
return this.#publishingSettings;
}
+ /**
+ * Get the client type.
+ *
+ * @return The client type identifier for producer
+ */
+ protected getClientType(): ClientType {
+ return ClientType.PRODUCER;
+ }
+
protected wrapHeartbeatRequest(): HeartbeatRequest {
return new HeartbeatRequest()
- .setClientType(ClientType.PRODUCER);
+ .setClientType(this.getClientType());
}
protected wrapNotifyClientTerminationRequest(): NotifyClientTerminationRequest {
@@ -207,11 +216,18 @@ export class Producer extends BaseClient {
const topic = pubMessages[0].topic;
const messageType = pubMessages[0].messageType;
const messageGroup = pubMessages[0].messageGroup;
+ const liteTopic = pubMessages[0].liteTopic;
const messageTypes = new Set(pubMessages.map(m => m.messageType));
if (messageTypes.size > 1) {
throw new TypeError(`Messages to send have different types=${JSON.stringify(Array.from(messageTypes))}`);
}
+ // Log Lite Topic message sending
+ if (liteTopic) {
+ this.logger.debug?.('Sending Lite Topic message, topic=%s, liteTopic=%s, messageType=%s, clientId=%s',
+ topic, liteTopic, MessageType[messageType], this.clientId);
+ }
+
// Message group must be same if message type is FIFO, or no need to proceed.
if (messageType === MessageType.FIFO) {
const messageGroups = new Set(pubMessages.map(m => m.messageGroup!));
@@ -302,6 +318,11 @@ export class Producer extends BaseClient {
this.logger.info('Resend message successfully, topic=%s, messageId(s)=%s, maxAttempts=%d, attempt=%d, endpoints=%s, clientId=%s',
topic, messageIds, maxAttempts, attempt, endpoints, this.clientId);
}
+ // Log Lite Topic message sent successfully
+ if (sendReceipts.length > 0 && messages[0].liteTopic) {
+ this.logger.debug?.('Lite Topic message sent successfully, topic=%s, liteTopic=%s, messageId=%s, clientId=%s',
+ topic, messages[0].liteTopic, sendReceipts[0].messageId, this.clientId);
+ }
// Send message(s) successfully on first attempt, return directly
return sendReceipts;
}
@@ -343,7 +364,7 @@ export class Producer extends BaseClient {
*
* @param topic - The topic associated with the scheduled message to be canceled.
* @param recallHandle - A unique handle to identify the message to recall (obtained from SendReceipt).
- * @returns Promise resolving to RecallReceipt containing the recalled message ID.
+ * @return Promise resolving to RecallReceipt containing the recalled message ID.
* @throws Error if producer is not running or recall handle is invalid.
*/
async recallMessage(topic: string, recallHandle: string): Promise<RecallReceipt> {
diff --git a/nodejs/src/producer/PublishingLoadBalancer.ts b/nodejs/src/producer/PublishingLoadBalancer.ts
index b04300ff..1d59c2a7 100644
--- a/nodejs/src/producer/PublishingLoadBalancer.ts
+++ b/nodejs/src/producer/PublishingLoadBalancer.ts
@@ -82,4 +82,24 @@ export class PublishingLoadBalancer {
const index = parseInt(`${hashCode % BigInt(this.#messageQueues.length)}`);
return this.#messageQueues[index];
}
+
+ equals(other: PublishingLoadBalancer): boolean {
+ if (this === other) return true;
+ if (!other) return false;
+ if (this.#messageQueues.length !== other.#messageQueues.length) return false;
+ for (let i = 0; i < this.#messageQueues.length; i++) {
+ if (!this.#messageQueues[i].equals(other.#messageQueues[i])) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ hashCode(): number {
+ let hash = 17;
+ for (const mq of this.#messageQueues) {
+ hash = hash * 31 + mq.hashCode();
+ }
+ return hash;
+ }
}
diff --git a/nodejs/src/route/Broker.ts b/nodejs/src/route/Broker.ts
index 3923b7a7..ffb0464b 100644
--- a/nodejs/src/route/Broker.ts
+++ b/nodejs/src/route/Broker.ts
@@ -16,6 +16,7 @@
*/
import { Broker as BrokerPB } from '../../proto/apache/rocketmq/v2/definition_pb';
+import { hashCodeOfString } from '../util';
import { Endpoints } from './Endpoints';
export class Broker implements BrokerPB.AsObject {
@@ -36,4 +37,20 @@ export class Broker implements BrokerPB.AsObject {
broker.setEndpoints(this.endpoints.toProtobuf());
return broker;
}
+
+ equals(other: Broker): boolean {
+ if (this === other) return true;
+ if (!other) return false;
+ return this.id === other.id &&
+ this.name === other.name &&
+ this.endpoints.equals(other.endpoints);
+ }
+
+ hashCode(): number {
+ let hash = 17;
+ hash = hash * 31 + this.id;
+ hash = hash * 31 + hashCodeOfString(this.name);
+ hash = hash * 31 + this.endpoints.hashCode();
+ return hash;
+ }
}
diff --git a/nodejs/src/route/Endpoints.ts b/nodejs/src/route/Endpoints.ts
index 6ac7332f..8e698521 100644
--- a/nodejs/src/route/Endpoints.ts
+++ b/nodejs/src/route/Endpoints.ts
@@ -16,6 +16,7 @@
*/
import { isIPv4, isIPv6 } from 'node:net';
+import { hashCodeOfString } from '../util';
import { Address, AddressScheme, Endpoints as EndpointsPB } from '../../proto/apache/rocketmq/v2/definition_pb';
const DEFAULT_PORT = 80;
@@ -67,4 +68,31 @@ export class Endpoints {
}
return endpoints;
}
+
+ equals(other: Endpoints): boolean {
+ if (this === other) return true;
+ if (!other) return false;
+ if (this.scheme !== other.scheme) return false;
+ if (this.facade !== other.facade) return false;
+ if (this.addressesList.length !== other.addressesList.length) return false;
+ for (let i = 0; i < this.addressesList.length; i++) {
+ const addr1 = this.addressesList[i];
+ const addr2 = other.addressesList[i];
+ if (addr1.host !== addr2.host || addr1.port !== addr2.port) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ hashCode(): number {
+ let hash = 17;
+ hash = hash * 31 + this.scheme;
+ hash = hash * 31 + hashCodeOfString(this.facade);
+ for (const addr of this.addressesList) {
+ hash = hash * 31 + hashCodeOfString(addr.host);
+ hash = hash * 31 + addr.port;
+ }
+ return hash;
+ }
}
diff --git a/nodejs/src/route/MessageQueue.ts b/nodejs/src/route/MessageQueue.ts
index 1569b75e..4f3151e3 100644
--- a/nodejs/src/route/MessageQueue.ts
+++ b/nodejs/src/route/MessageQueue.ts
@@ -19,7 +19,7 @@ import {
MessageQueue as MessageQueuePB, MessageType, Permission,
Resource,
} from '../../proto/apache/rocketmq/v2/definition_pb';
-import { createResource } from '../util';
+import { createResource, hashCodeOfString } from '../util';
import { Broker } from './Broker';
export class MessageQueue {
@@ -46,4 +46,22 @@ export class MessageQueue {
messageQueue.setAcceptMessageTypesList(this.acceptMessageTypesList);
return messageQueue;
}
+
+ equals(other: MessageQueue): boolean {
+ if (this === other) return true;
+ if (!other) return false;
+ return this.queueId === other.queueId &&
+ this.topic.name === other.topic.name &&
+ this.broker.equals(other.broker) &&
+ this.permission === other.permission;
+ }
+
+ hashCode(): number {
+ let hash = 17;
+ hash = hash * 31 + this.queueId;
+ hash = hash * 31 + hashCodeOfString(this.topic.name);
+ hash = hash * 31 + this.broker.hashCode();
+ hash = hash * 31 + this.permission;
+ return hash;
+ }
}
diff --git a/nodejs/src/route/Resource.ts b/nodejs/src/route/Resource.ts
new file mode 100644
index 00000000..2aabff73
--- /dev/null
+++ b/nodejs/src/route/Resource.ts
@@ -0,0 +1,102 @@
+/**
+ * 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 { Resource as ProtoResource } from '../../proto/apache/rocketmq/v2/definition_pb';
+import { hashCodeOfString } from '../util';
+
+/**
+ * Represents a resource with namespace and name.
+ */
+export class Resource {
+ private readonly namespace: string;
+ private readonly name: string;
+
+ constructor(namespace: string, name: string);
+ constructor(name: string);
+ constructor(namespaceOrName: string, name?: string) {
+ if (name !== undefined) {
+ this.namespace = namespaceOrName;
+ this.name = name;
+ } else {
+ this.namespace = '';
+ this.name = namespaceOrName;
+ }
+ }
+
+ /**
+ * Get the namespace.
+ */
+ getNamespace(): string {
+ return this.namespace;
+ }
+
+ /**
+ * Get the name.
+ */
+ getName(): string {
+ return this.name;
+ }
+
+ /**
+ * Convert to protobuf Resource.
+ */
+ toProtobuf(): ProtoResource {
+ const resource = new ProtoResource();
+ resource.setResourceNamespace(this.namespace);
+ resource.setName(this.name);
+ return resource;
+ }
+
+ /**
+ * Create from protobuf Resource.
+ */
+ static fromProtobuf(proto: ProtoResource): Resource {
+ const namespace = proto.getResourceNamespace();
+ const name = proto.getName();
+ return new Resource(namespace, name);
+ }
+
+ /**
+ * Returns a string representation of the resource.
+ */
+ toString(): string {
+ if (this.namespace && this.namespace.length > 0) {
+ return `Resource{name='${this.name}', namespace='${this.namespace}'}`;
+ }
+ return this.name;
+ }
+
+ /**
+ * Check equality with another Resource.
+ */
+ equals(other: Resource): boolean {
+ if (this === other) return true;
+ if (!other) return false;
+ return this.namespace === other.namespace &&
+ this.name === other.name;
+ }
+
+ /**
+ * Calculate hash code.
+ */
+ hashCode(): number {
+ let hash = 17;
+ hash = hash * 31 + hashCodeOfString(this.namespace);
+ hash = hash * 31 + hashCodeOfString(this.name);
+ return hash;
+ }
+}
diff --git a/nodejs/src/route/TopicRouteData.ts b/nodejs/src/route/TopicRouteData.ts
index 6abae444..876312f8 100644
--- a/nodejs/src/route/TopicRouteData.ts
+++ b/nodejs/src/route/TopicRouteData.ts
@@ -35,4 +35,40 @@ export class TopicRouteData {
}
return Array.from(endpointsMap.values());
}
+
+ /**
+ * Get the count of message queues.
+ * @return number of message queues
+ */
+ getQueueCount(): number {
+ return this.messageQueues.length;
+ }
+
+ /**
+ * Check if the route data is empty.
+ * @return true if no message queues available
+ */
+ isEmpty(): boolean {
+ return this.messageQueues.length === 0;
+ }
+
+ equals(other: TopicRouteData): boolean {
+ if (this === other) return true;
+ if (!other) return false;
+ if (this.messageQueues.length !== other.messageQueues.length) return false;
+ for (let i = 0; i < this.messageQueues.length; i++) {
+ if (!this.messageQueues[i].equals(other.messageQueues[i])) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ hashCode(): number {
+ let hash = 17;
+ for (const mq of this.messageQueues) {
+ hash = hash * 31 + mq.hashCode();
+ }
+ return hash;
+ }
}
diff --git a/nodejs/src/route/index.ts b/nodejs/src/route/index.ts
index 571e4f92..790c213c 100644
--- a/nodejs/src/route/index.ts
+++ b/nodejs/src/route/index.ts
@@ -19,3 +19,4 @@ export * from './Broker';
export * from './Endpoints';
export * from './MessageQueue';
export * from './TopicRouteData';
+export * from './Resource';
diff --git a/nodejs/src/util/index.ts b/nodejs/src/util/index.ts
index 7b4243fe..eb2af3a8 100644
--- a/nodejs/src/util/index.ts
+++ b/nodejs/src/util/index.ts
@@ -81,3 +81,22 @@ export function calculateStringSipHash24(value: string) {
const hash = siphash24(Buffer.from(value), SIP_HASH_24_KEY);
return Buffer.from(hash).readBigUInt64BE();
}
+
+/**
+ * Calculate hash code for a string (Java String.hashCode() compatible).
+ * This is used for implementing equals/hashCode contract in TypeScript classes.
+ *
+ * @param str - The string to calculate hash code for
+ * @return Hash code as a 32-bit integer
+ */
+export function hashCodeOfString(str: string): number {
+ let hash = 0;
+ for (let i = 0; i < str.length; i++) {
+ const char = str.charCodeAt(i);
+ // eslint-disable-next-line no-bitwise
+ hash = ((hash << 5) - hash) + char;
+ // eslint-disable-next-line no-bitwise
+ hash = hash & hash; // Convert to 32bit integer
+ }
+ return hash;
+}
diff --git a/nodejs/test/consumer/LitePushConsumerBuilder.test.ts b/nodejs/test/consumer/LitePushConsumerBuilder.test.ts
new file mode 100644
index 00000000..ca02efe4
--- /dev/null
+++ b/nodejs/test/consumer/LitePushConsumerBuilder.test.ts
@@ -0,0 +1,222 @@
+/**
+ * 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.
+ */
+
+/**
+ * Test for LitePushConsumerBuilder
+ */
+
+import { describe, it } from 'node:test';
+import * as assert from 'node:assert';
+import { LitePushConsumerBuilder } from '../../src';
+import { BaseClientOptions } from '../../src/client';
+import { ConsumeResult } from '../../src';
+
+describe('LitePushConsumerBuilder', () => {
+ const createMockClientConfig = (): BaseClientOptions => {
+ return {
+ endpoints: '127.0.0.1:8080',
+ namespace: 'test-namespace',
+ };
+ };
+
+ const createMockMessageListener = () => {
+ return {
+ // eslint-disable-next-line @typescript-eslint/no-unused-vars
+ consume: async (_messageView: any) => {
+ // Mock listener
+ return ConsumeResult.SUCCESS;
+ },
+ };
+ };
+
+ describe('bindTopic()', () => {
+ it('should accept valid topic', () => {
+ const builder = new LitePushConsumerBuilder();
+ const result = builder.bindTopic('test-topic');
+ assert.strictEqual(result, builder);
+ });
+
+ it('should throw error for blank topic', () => {
+ const builder = new LitePushConsumerBuilder();
+ assert.throws(() => {
+ builder.bindTopic('');
+ }, /bindTopic should not be blank/);
+ });
+
+ it('should throw error for null topic', () => {
+ const builder = new LitePushConsumerBuilder();
+ assert.throws(() => {
+ builder.bindTopic(null as any);
+ }, /bindTopic should not be blank/);
+ });
+ });
+
+ describe('setConsumerGroup()', () => {
+ it('should accept valid consumer group', () => {
+ const builder = new LitePushConsumerBuilder();
+ const result = builder.setConsumerGroup('test-group');
+ assert.strictEqual(result, builder);
+ });
+
+ it('should throw error for null consumer group', () => {
+ const builder = new LitePushConsumerBuilder();
+ assert.throws(() => {
+ builder.setConsumerGroup(null as any);
+ }, /consumerGroup should not be null/);
+ });
+
+ it('should throw error for invalid consumer group format', () => {
+ const builder = new LitePushConsumerBuilder();
+ assert.throws(() => {
+ builder.setConsumerGroup('invalid@group');
+ }, /does not match the pattern/);
+ });
+ });
+
+ describe('setMessageListener()', () => {
+ it('should accept valid message listener', () => {
+ const builder = new LitePushConsumerBuilder();
+ const listener = createMockMessageListener();
+ const result = builder.setMessageListener(listener);
+ assert.strictEqual(result, builder);
+ });
+
+ it('should throw error for null message listener', () => {
+ const builder = new LitePushConsumerBuilder();
+ assert.throws(() => {
+ builder.setMessageListener(null as any);
+ }, /messageListener should not be null/);
+ });
+ });
+
+ describe('setMaxCacheMessageCount()', () => {
+ it('should accept positive value', () => {
+ const builder = new LitePushConsumerBuilder();
+ const result = builder.setMaxCacheMessageCount(512);
+ assert.strictEqual(result, builder);
+ });
+
+ it('should throw error for zero value', () => {
+ const builder = new LitePushConsumerBuilder();
+ assert.throws(() => {
+ builder.setMaxCacheMessageCount(0);
+ }, /maxCacheMessageCount should be positive/);
+ });
+
+ it('should throw error for negative value', () => {
+ const builder = new LitePushConsumerBuilder();
+ assert.throws(() => {
+ builder.setMaxCacheMessageCount(-1);
+ }, /maxCacheMessageCount should be positive/);
+ });
+ });
+
+ describe('setMaxCacheMessageSizeInBytes()', () => {
+ it('should accept positive value', () => {
+ const builder = new LitePushConsumerBuilder();
+ const result = builder.setMaxCacheMessageSizeInBytes(32 * 1024 * 1024);
+ assert.strictEqual(result, builder);
+ });
+
+ it('should throw error for zero value', () => {
+ const builder = new LitePushConsumerBuilder();
+ assert.throws(() => {
+ builder.setMaxCacheMessageSizeInBytes(0);
+ }, /maxCacheMessageSizeInBytes should be positive/);
+ });
+ });
+
+ describe('setConsumptionThreadCount()', () => {
+ it('should accept positive value', () => {
+ const builder = new LitePushConsumerBuilder();
+ const result = builder.setConsumptionThreadCount(10);
+ assert.strictEqual(result, builder);
+ });
+
+ it('should throw error for zero value', () => {
+ const builder = new LitePushConsumerBuilder();
+ assert.throws(() => {
+ builder.setConsumptionThreadCount(0);
+ }, /consumptionThreadCount should be positive/);
+ });
+ });
+
+ describe('build()', () => {
+ it('should throw error when clientConfiguration not set', async () => {
+ const builder = new LitePushConsumerBuilder();
+ builder.setConsumerGroup('test-group');
+ builder.setMessageListener(createMockMessageListener());
+ builder.bindTopic('test-topic');
+
+ await assert.rejects(() => builder.build(), /clientConfiguration has not been set yet/);
+ });
+
+ it('should accept valid configuration and build', async () => {
+ const builder = new LitePushConsumerBuilder();
+ builder.setClientConfiguration(createMockClientConfig());
+ builder.setConsumerGroup('test-group');
+ builder.setMessageListener(createMockMessageListener());
+ builder.bindTopic('test-topic');
+
+ // Should reject with "not implemented" error since we haven't implemented the full consumer yet
+ await assert.rejects(
+ () => builder.build(),
+ /LitePushConsumerImpl not yet implemented/,
+ );
+ });
+
+ it('should throw error when consumerGroup not set', async () => {
+ const builder = new LitePushConsumerBuilder();
+ builder.setClientConfiguration(createMockClientConfig());
+ builder.setMessageListener(createMockMessageListener());
+ builder.bindTopic('test-topic');
+
+ await assert.rejects(() => builder.build(), /consumerGroup has not been set yet/);
+ });
+
+ it('should throw error when messageListener not set', async () => {
+ const builder = new LitePushConsumerBuilder();
+ builder.setClientConfiguration(createMockClientConfig());
+ builder.setConsumerGroup('test-group');
+ builder.bindTopic('test-topic');
+
+ await assert.rejects(() => builder.build(), /messageListener has not been set yet/);
+ });
+
+ it('should throw error when bindTopic not set', async () => {
+ const builder = new LitePushConsumerBuilder();
+ builder.setClientConfiguration(createMockClientConfig());
+ builder.setConsumerGroup('test-group');
+ builder.setMessageListener(createMockMessageListener());
+
+ await assert.rejects(() => builder.build(), /bindTopic has not been set yet/);
+ });
+
+ it('should throw not implemented error when all params set', async () => {
+ const builder = new LitePushConsumerBuilder();
+ builder.setClientConfiguration(createMockClientConfig());
+ builder.setConsumerGroup('test-group');
+ builder.setMessageListener(createMockMessageListener());
+ builder.bindTopic('test-topic');
+
+ await assert.rejects(
+ () => builder.build(),
+ /LitePushConsumerImpl not yet implemented/,
+ );
+ });
+ });
+});
diff --git a/nodejs/test/consumer/OffsetOption.test.ts b/nodejs/test/consumer/OffsetOption.test.ts
new file mode 100644
index 00000000..a70a5e19
--- /dev/null
+++ b/nodejs/test/consumer/OffsetOption.test.ts
@@ -0,0 +1,114 @@
+/**
+ * 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.
+ */
+
+/**
+ * Test for OffsetOption
+ */
+
+import { describe, it } from 'node:test';
+import * as assert from 'node:assert';
+import { OffsetOption, OffsetType } from '../../src/consumer/OffsetOption';
+
+describe('OffsetOption', () => {
+ describe('Predefined constants', () => {
+ it('should have LAST_OFFSET constant', () => {
+ assert.strictEqual(OffsetOption.LAST_OFFSET.getType(), OffsetType.POLICY);
+ assert.strictEqual(OffsetOption.LAST_OFFSET.getValue(), BigInt(0));
+ });
+
+ it('should have MIN_OFFSET constant', () => {
+ assert.strictEqual(OffsetOption.MIN_OFFSET.getType(), OffsetType.POLICY);
+ assert.strictEqual(OffsetOption.MIN_OFFSET.getValue(), BigInt(1));
+ });
+
+ it('should have MAX_OFFSET constant', () => {
+ assert.strictEqual(OffsetOption.MAX_OFFSET.getType(), OffsetType.POLICY);
+ assert.strictEqual(OffsetOption.MAX_OFFSET.getValue(), BigInt(2));
+ });
+ });
+
+ describe('ofOffset()', () => {
+ it('should create offset from number', () => {
+ const option = OffsetOption.ofOffset(100);
+ assert.strictEqual(option.getType(), OffsetType.OFFSET);
+ assert.strictEqual(option.getValue(), BigInt(100));
+ });
+
+ it('should create offset from bigint', () => {
+ const option = OffsetOption.ofOffset(BigInt(100));
+ assert.strictEqual(option.getType(), OffsetType.OFFSET);
+ assert.strictEqual(option.getValue(), BigInt(100));
+ });
+
+ it('should throw error for negative offset', () => {
+ assert.throws(() => {
+ OffsetOption.ofOffset(-1);
+ }, /offset must be greater than or equal to 0/);
+ });
+ });
+
+ describe('ofTailN()', () => {
+ it('should create tailN from number', () => {
+ const option = OffsetOption.ofTailN(50);
+ assert.strictEqual(option.getType(), OffsetType.TAIL_N);
+ assert.strictEqual(option.getValue(), BigInt(50));
+ });
+
+ it('should create tailN from bigint', () => {
+ const option = OffsetOption.ofTailN(BigInt(50));
+ assert.strictEqual(option.getType(), OffsetType.TAIL_N);
+ assert.strictEqual(option.getValue(), BigInt(50));
+ });
+
+ it('should throw error for negative tailN', () => {
+ assert.throws(() => {
+ OffsetOption.ofTailN(-1);
+ }, /tailN must be greater than or equal to 0/);
+ });
+ });
+
+ describe('ofTimestamp()', () => {
+ it('should create timestamp from number', () => {
+ const timestamp = Date.now();
+ const option = OffsetOption.ofTimestamp(timestamp);
+ assert.strictEqual(option.getType(), OffsetType.TIMESTAMP);
+ assert.strictEqual(option.getValue(), BigInt(timestamp));
+ });
+
+ it('should create timestamp from bigint', () => {
+ const timestamp = BigInt(Date.now());
+ const option = OffsetOption.ofTimestamp(timestamp);
+ assert.strictEqual(option.getType(), OffsetType.TIMESTAMP);
+ assert.strictEqual(option.getValue(), timestamp);
+ });
+
+ it('should throw error for negative timestamp', () => {
+ assert.throws(() => {
+ OffsetOption.ofTimestamp(-1);
+ }, /timestamp must be greater than or equal to 0/);
+ });
+ });
+
+ describe('toString()', () => {
+ it('should return string representation', () => {
+ const option = OffsetOption.ofOffset(100);
+ const str = option.toString();
+ assert.ok(str.includes('type=OFFSET'));
+ assert.ok(str.includes('value=100'));
+ });
+ });
+});
diff --git a/nodejs/test/exception/LiteSubscriptionQuotaExceededException.test.ts b/nodejs/test/exception/LiteSubscriptionQuotaExceededException.test.ts
new file mode 100644
index 00000000..24ee998a
--- /dev/null
+++ b/nodejs/test/exception/LiteSubscriptionQuotaExceededException.test.ts
@@ -0,0 +1,81 @@
+/**
+ * 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.
+ */
+
+/**
+ * Test for LiteSubscriptionQuotaExceededException
+ */
+
+import { describe, it } from 'node:test';
+import * as assert from 'node:assert';
+import { LiteSubscriptionQuotaExceededException } from '../../src/exception/LiteSubscriptionQuotaExceededException';
+
+describe('LiteSubscriptionQuotaExceededException', () => {
+ it('should create exception with correct properties', () => {
+ const responseCode = 429;
+ const requestId = 'test-request-id';
+ const message = 'Lite subscription quota exceeded 100';
+
+ const exception = new LiteSubscriptionQuotaExceededException(
+ responseCode,
+ requestId,
+ message,
+ );
+
+ assert.strictEqual(exception.code, responseCode);
+ assert.ok(exception.message.includes(requestId));
+ assert.ok(exception.message.includes(message));
+ assert.strictEqual(exception.name, 'LiteSubscriptionQuotaExceededException');
+ });
+
+ it('should create exception with null requestId', () => {
+ const responseCode = 429;
+ const message = 'Quota exceeded';
+
+ const exception = new LiteSubscriptionQuotaExceededException(
+ responseCode,
+ null,
+ message,
+ );
+
+ assert.strictEqual(exception.code, responseCode);
+ assert.ok(exception.message.includes(message));
+ });
+
+ it('should be instance of Error', () => {
+ const exception = new LiteSubscriptionQuotaExceededException(
+ 429,
+ null,
+ 'Test error',
+ );
+
+ assert.ok(exception instanceof Error);
+ assert.ok(exception instanceof LiteSubscriptionQuotaExceededException);
+ });
+
+ it('should have proper stack trace', () => {
+ try {
+ throw new LiteSubscriptionQuotaExceededException(
+ 429,
+ 'request-123',
+ 'Test quota exceeded error',
+ );
+ } catch (error: any) {
+ assert.ok(error.stack);
+ assert.ok(error.stack.includes('LiteSubscriptionQuotaExceededException'));
+ }
+ });
+});
diff --git a/nodejs/test/message/LiteMessage.test.ts b/nodejs/test/message/LiteMessage.test.ts
new file mode 100644
index 00000000..a964f00d
--- /dev/null
+++ b/nodejs/test/message/LiteMessage.test.ts
@@ -0,0 +1,105 @@
+/**
+ * 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.
+ */
+
+/**
+ * Test for Lite Message
+ */
+
+import { describe, it } from 'node:test';
+import * as assert from 'node:assert';
+import { Message } from '../../src/message/Message';
+
+describe('Lite Message', () => {
+ const fakeTopic = 'test-topic';
+ const fakeBody = Buffer.from('test body');
+
+ it('should create message with liteTopic', () => {
+ const message = new Message({
+ topic: fakeTopic,
+ body: fakeBody,
+ liteTopic: 'lite-topic-name',
+ });
+
+ assert.strictEqual(message.liteTopic, 'lite-topic-name');
+ });
+
+ it('should create message without liteTopic (undefined)', () => {
+ const message = new Message({
+ topic: fakeTopic,
+ body: fakeBody,
+ });
+
+ assert.strictEqual(message.liteTopic, undefined);
+ });
+
+ it('should throw error when liteTopic and priority are set together', () => {
+ assert.throws(() => {
+ new Message({
+ topic: fakeTopic,
+ body: fakeBody,
+ liteTopic: 'lite-topic-name',
+ priority: 5,
+ });
+ }, /priority and liteTopic should not be set at same time/);
+ });
+
+ it('should throw error when liteTopic and messageGroup are set together', () => {
+ assert.throws(() => {
+ new Message({
+ topic: fakeTopic,
+ body: fakeBody,
+ liteTopic: 'lite-topic-name',
+ messageGroup: 'test-group',
+ });
+ }, /liteTopic and messageGroup should not be set at same time/);
+ });
+
+ it('should throw error when liteTopic and deliveryTimestamp are set together', () => {
+ assert.throws(() => {
+ new Message({
+ topic: fakeTopic,
+ body: fakeBody,
+ liteTopic: 'lite-topic-name',
+ deliveryTimestamp: new Date(Date.now() + 60000),
+ });
+ }, /liteTopic and deliveryTimestamp should not be set at same time/);
+ });
+
+ it('should accept empty string liteTopic', () => {
+ const message = new Message({
+ topic: fakeTopic,
+ body: fakeBody,
+ liteTopic: '',
+ });
+
+ assert.strictEqual(message.liteTopic, '');
+ });
+
+ it('should include liteTopic in toString output', () => {
+ const message = new Message({
+ topic: fakeTopic,
+ body: fakeBody,
+ liteTopic: 'test-lite-topic',
+ tag: 'test-tag',
+ });
+
+ const str = message.toString();
+ assert.ok(str.includes('liteTopic=test-lite-topic'));
+ assert.ok(str.includes('tag=test-tag'));
+ assert.ok(str.includes(fakeTopic));
+ });
+});
diff --git a/nodejs/test/message/PublishingLiteMessage.test.ts b/nodejs/test/message/PublishingLiteMessage.test.ts
new file mode 100644
index 00000000..ca20319c
--- /dev/null
+++ b/nodejs/test/message/PublishingLiteMessage.test.ts
@@ -0,0 +1,59 @@
+/**
+ * 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.
+ */
+
+/**
+ * Test for Publishing Lite Message
+ */
+
+import { describe, it } from 'node:test';
+import * as assert from 'node:assert';
+import { Message } from '../../src/message/Message';
+
+describe('PublishingMessage with LITE type', () => {
+ const fakeTopic = 'test-topic';
+ const fakeBody = Buffer.from('test body');
+
+ it('should create message with liteTopic', () => {
+ const message = new Message({
+ topic: fakeTopic,
+ body: fakeBody,
+ liteTopic: 'lite-topic-name',
+ });
+
+ assert.strictEqual(message.liteTopic, 'lite-topic-name');
+ });
+
+ it('should create message without liteTopic (undefined)', () => {
+ const message = new Message({
+ topic: fakeTopic,
+ body: fakeBody,
+ });
+
+ assert.strictEqual(message.liteTopic, undefined);
+ });
+
+ it('should throw error when liteTopic and priority are set together', () => {
+ assert.throws(() => {
+ new Message({
+ topic: fakeTopic,
+ body: fakeBody,
+ liteTopic: 'lite-topic-name',
+ priority: 5,
+ });
+ }, /priority and liteTopic should not be set at same time/);
+ });
+});
diff --git a/protos b/protos
index 5c9f8419..68c2cc94 160000
--- a/protos
+++ b/protos
@@ -1 +1 @@
-Subproject commit 5c9f84199bffa79b2ed73beb37774ca92e749c19
+Subproject commit 68c2cc9442928f769f8938515a05af6fa05c9993