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 42a965ca feat(consumer): implement LitePushConsumer with FIFO and
standard consume services (#1294)
42a965ca is described below
commit 42a965cab34230120112478e4765599a27c17b6e
Author: Drizzle <[email protected]>
AuthorDate: Mon Jul 6 19:45:01 2026 +0800
feat(consumer): implement LitePushConsumer with FIFO and standard consume
services (#1294)
- Add LitePushConsumerImpl extending PushConsumer for lightweight message
consumption
- Implement LiteFifoConsumeService with liteTopic-based grouping and
suspend support
- Implement LiteStandardConsumeService with local retry via eraseFifoMessage
- Refactor ConsumeResult from enum to class-based design with
ConsumeResultSuspend extension
- Optimize FifoConsumeService: add parameter validation, improve Map
grouping logic
- Enhance ProcessQueue: add FIFO message discard/erase methods with local
retry
- Align LitePushConsumerBuilder with Java client: implement
maxCacheMessageCount,
maxCacheMessageSizeInBytes, consumptionThreadCount configurations
- Add unit tests for LiteFifoConsumeService, ConsumeResult, and
LitePushConsumerBuilder
- Update MessageView to support liteTopic field for lite consumer routing
Co-authored-by: drizzle.zk <[email protected]>
---
nodejs/package-lock.json | 4 +-
nodejs/package.json | 2 +-
nodejs/proto/apache/rocketmq/v2/definition.proto | 2 +-
nodejs/src/consumer/ConsumeResult.ts | 65 +++++++-
nodejs/src/consumer/FifoConsumeService.ts | 171 ++++++++++-----------
nodejs/src/consumer/LiteFifoConsumeService.ts | 103 +++++++++++++
nodejs/src/consumer/LitePushConsumer.ts | 32 +++-
nodejs/src/consumer/LitePushConsumerImpl.ts | 36 +++++
nodejs/src/consumer/LiteStandardConsumeService.ts | 73 +++++++++
nodejs/src/consumer/ProcessQueue.ts | 71 ++++++++-
nodejs/src/consumer/PushConsumer.ts | 7 +-
nodejs/src/consumer/index.ts | 2 +
nodejs/src/message/MessageView.ts | 13 +-
nodejs/test/consumer/ConsumeResult.test.ts | 43 ++++++
nodejs/test/consumer/FifoConsumeService.test.ts | 2 +-
.../test/consumer/LiteFifoConsumeService.test.ts | 157 +++++++++++++++++++
.../test/consumer/LitePushConsumerBuilder.test.ts | 23 ++-
nodejs/test/message/LiteMessage.test.ts | 6 +-
nodejs/test/message/PriorityMessage.test.ts | 4 +-
nodejs/test/message/PublishingLiteMessage.test.ts | 2 +-
20 files changed, 694 insertions(+), 124 deletions(-)
diff --git a/nodejs/package-lock.json b/nodejs/package-lock.json
index 8a1e2b2a..2cba0785 100644
--- a/nodejs/package-lock.json
+++ b/nodejs/package-lock.json
@@ -1,12 +1,12 @@
{
"name": "rocketmq-client-nodejs",
- "version": "1.0.5",
+ "version": "1.0.7",
"lockfileVersion": 3,
"requires": true,
"packages": {
"": {
"name": "rocketmq-client-nodejs",
- "version": "1.0.5",
+ "version": "1.0.7",
"license": "Apache-2.0",
"dependencies": {
"@grpc/grpc-js": "^1.9.1",
diff --git a/nodejs/package.json b/nodejs/package.json
index 49917098..eda38358 100644
--- a/nodejs/package.json
+++ b/nodejs/package.json
@@ -1,6 +1,6 @@
{
"name": "rocketmq-client-nodejs",
- "version": "1.0.6",
+ "version": "1.0.7",
"description": "RocketMQ Node.js Client",
"main": "dist/index.js",
"types": "dist/index.d.ts",
diff --git a/nodejs/proto/apache/rocketmq/v2/definition.proto
b/nodejs/proto/apache/rocketmq/v2/definition.proto
index 516474df..a87dee3b 100644
--- a/nodejs/proto/apache/rocketmq/v2/definition.proto
+++ b/nodejs/proto/apache/rocketmq/v2/definition.proto
@@ -617,4 +617,4 @@ message OffsetOption {
MIN = 1;
MAX = 2;
}
-}
\ No newline at end of file
+}
diff --git a/nodejs/src/consumer/ConsumeResult.ts
b/nodejs/src/consumer/ConsumeResult.ts
index e3d2aafa..9dc70a9e 100644
--- a/nodejs/src/consumer/ConsumeResult.ts
+++ b/nodejs/src/consumer/ConsumeResult.ts
@@ -15,7 +15,66 @@
* limitations under the License.
*/
-export enum ConsumeResult {
- SUCCESS = 'SUCCESS',
- FAILURE = 'FAILURE',
+/**
+ * Consume result for push consumer.
+ *
+ * <p>Designed for push consumer specifically. This is a class-based result
+ * (matching the Java client) so that it can be extended with suspend
results.</p>
+ */
+export class ConsumeResult {
+ /**
+ * Consume message successfully.
+ */
+ static readonly SUCCESS = new ConsumeResult('SUCCESS');
+
+ /**
+ * Failed to consume message.
+ */
+ static readonly FAILURE = new ConsumeResult('FAILURE');
+
+ readonly name: string;
+
+ protected constructor(name: string) {
+ this.name = name;
+ }
+
+ toString(): string {
+ return this.name;
+ }
+}
+
+const MIN_SUSPEND_TIME_MS = 50;
+
+/**
+ * Suspend consume result.
+ *
+ * <p>When returned by the message listener, the consumer will suspend
+ * consumption of the current message (and related messages in FIFO mode)
+ * for the specified duration by changing the invisible duration.</p>
+ */
+export class ConsumeResultSuspend extends ConsumeResult {
+ readonly suspendTimeMs: number;
+
+ private constructor(suspendTimeMs: number) {
+ super('SUSPEND');
+ if (suspendTimeMs < MIN_SUSPEND_TIME_MS) {
+ throw new Error(`suspend time cannot be less than
${MIN_SUSPEND_TIME_MS}ms, got ${suspendTimeMs}ms`);
+ }
+ this.suspendTimeMs = suspendTimeMs;
+ }
+
+ /**
+ * Create a suspend result with the given suspend time in milliseconds.
+ *
+ * @param suspendTimeMs - Suspend time in milliseconds
+ * @return {ConsumeResultSuspend} ConsumeResultSuspend instance
+ * @throws {Error} if suspendTimeMs is less than 50ms
+ */
+ static of(suspendTimeMs: number): ConsumeResultSuspend {
+ return new ConsumeResultSuspend(suspendTimeMs);
+ }
+
+ toString(): string {
+ return `SUSPEND(${this.suspendTimeMs}ms)`;
+ }
}
diff --git a/nodejs/src/consumer/FifoConsumeService.ts
b/nodejs/src/consumer/FifoConsumeService.ts
index 58bdffd2..f7bfae74 100644
--- a/nodejs/src/consumer/FifoConsumeService.ts
+++ b/nodejs/src/consumer/FifoConsumeService.ts
@@ -22,125 +22,110 @@ import type { ProcessQueue } from './ProcessQueue';
import { ILogger, getDefaultLogger } from '../client/Logger';
export class FifoConsumeService extends ConsumeService {
- readonly #enableAccelerator: boolean;
- readonly #groupProcessing = new Map<string /* messageGroup */,
Promise<void>>();
+ readonly #enableFifoConsumeAccelerator: boolean;
readonly #logger: ILogger;
- constructor(clientId: string, messageListener: MessageListener,
enableAccelerator?: boolean) {
+ constructor(clientId: string, messageListener: MessageListener,
enableFifoConsumeAccelerator?: boolean) {
super(clientId, messageListener);
- this.#enableAccelerator = enableAccelerator ?? false;
+ this.#enableFifoConsumeAccelerator = enableFifoConsumeAccelerator ?? false;
this.#logger = getDefaultLogger();
}
consume(pq: ProcessQueue, messageViews: MessageView[]): void {
- // Use iterative consumption when accelerator is disabled or only one
message
- if (!this.#enableAccelerator || messageViews.length <= 1) {
- this.#consumeIteratively(pq, messageViews, 0);
+ if (!pq || !messageViews) {
+ this.#logger.error('[Bug] Invalid arguments for consume, pq=%s,
messageViews=%s, clientId=%s',
+ pq, messageViews, this.clientId);
+ return;
+ }
+ if (!this.#enableFifoConsumeAccelerator || messageViews.length <= 1) {
+ this.consumeIteratively(pq, messageViews, 0).catch(() => {
+ // Error already logged, ignore
+ });
return;
}
- this.#consumeWithAccelerator(pq, messageViews);
- }
-
- /**
- * FIFO consume accelerator mode:
- * - Messages with the same messageGroup are consumed sequentially
- * - Messages with different messageGroups are consumed in parallel
- * - Messages without messageGroup are consumed in parallel
- */
- #consumeWithAccelerator(pq: ProcessQueue, messageViews: MessageView[]): void
{
- // Group messages by messageGroup
- const groupedMessages = new Map<string, MessageView[]>();
- const ungroupedMessages: MessageView[] = [];
+ // Group messages by group key. Default to null-key group for unkeyed
messages.
+ const messageViewsGroupByGroupKey = new Map<string, MessageView[]>();
for (const messageView of messageViews) {
- if (messageView.corrupted) {
- pq.discardFifoMessage(messageView);
- continue;
- }
-
- const messageGroup = messageView.messageGroup || '';
- if (messageGroup) {
- const group = groupedMessages.get(messageGroup) || [];
- group.push(messageView);
- groupedMessages.set(messageGroup, group);
- } else {
- ungroupedMessages.push(messageView);
+ const groupKey = this.getMessageGroupKey(messageView);
+ let group = messageViewsGroupByGroupKey.get(groupKey);
+ if (!group) {
+ group = [];
+ messageViewsGroupByGroupKey.set(groupKey, group);
}
+ group.push(messageView);
}
- // Log parallel consumption info
- const groupCount = groupedMessages.size + (ungroupedMessages.length > 0 ?
1 : 0);
this.#logger.debug?.('FifoConsumeService parallel consume,
messageViewsNum=%d, groupNum=%d',
- messageViews.length, groupCount);
-
- // Process grouped messages (each group sequentially, groups in parallel)
- for (const [ , messages ] of groupedMessages.entries()) {
- this.#processMessageGroup(pq, messages);
- }
+ messageViews.length, messageViewsGroupByGroupKey.size);
- // Process ungrouped messages in parallel
- for (const messageView of ungroupedMessages) {
- this.consumeMessage(messageView)
- .then(result => pq.eraseFifoMessage(messageView, result))
- .catch(() => {
- // Error already logged, continue with next message
- });
+ for (const list of messageViewsGroupByGroupKey.values()) {
+ this.consumeIteratively(pq, list, 0).catch(() => {
+ // Error already logged, ignore
+ });
}
}
/**
- * Process messages within the same group sequentially
+ * Get the group key for the given message view.
+ * Subclasses can override this method to provide different grouping logic.
+ *
+ * @param messageView - The message view
+ * @return {string} The group key, empty string means no grouping
*/
- async #processMessageGroup(pq: ProcessQueue, messages: MessageView[]):
Promise<void> {
- // Check if there's already processing happening for this group
- const messageGroup = messages[0]?.messageGroup || 'NO_GROUP';
- const existingPromise = this.#groupProcessing.get(messageGroup);
- const processTask = (async () => {
- // Wait for previous processing to complete if any
- if (existingPromise) {
- await existingPromise.catch(() => {
- // Ignore previous errors, continue processing
- });
- }
-
- // Process messages in sequence
- for (const messageView of messages) {
- try {
- const result = await this.consumeMessage(messageView);
- await pq.eraseFifoMessage(messageView, result);
- } catch (error) {
- this.#logger.error('Failed to process FIFO message, messageGroup=%s,
messageId=%s, error=%s',
- messageGroup, messageView.messageId, error);
- // Continue with next message even if current fails
- }
- }
- })();
-
- // Store the promise for this group
- this.#groupProcessing.set(messageGroup, processTask);
-
- // Clean up when done
- processTask.finally(() => {
- this.#groupProcessing.delete(messageGroup);
- });
+ protected getMessageGroupKey(messageView: MessageView): string {
+ return messageView.messageGroup || '';
}
- #consumeIteratively(pq: ProcessQueue, messageViews: MessageView[], index:
number): void {
- if (index >= messageViews.length) {
- return;
+ /**
+ * Consume messages iteratively using the provided message view array.
+ * This method handles corrupted messages and continues processing the next
valid message.
+ *
+ * @param pq - The process queue
+ * @param messageViews - The message views to consume
+ * @param startIndex - The index to start consuming from
+ * @return {Promise<void>} Promise that resolves when all messages are
consumed
+ */
+ protected consumeIteratively(pq: ProcessQueue, messageViews: MessageView[],
startIndex: number): Promise<void> {
+ const next = this.getNextValidMessage(pq, messageViews, startIndex);
+ if (!next) {
+ return Promise.resolve();
}
+ const { messageView, nextIndex } = next;
- const messageView = messageViews[index];
+ return this.consumeMessage(messageView)
+ .then(result => pq.eraseFifoMessage(messageView, result))
+ .then(() => this.consumeIteratively(pq, messageViews, nextIndex))
+ .catch(() => this.consumeIteratively(pq, messageViews, nextIndex));
+ }
- if (messageView.corrupted) {
- pq.discardFifoMessage(messageView);
- this.#consumeIteratively(pq, messageViews, index + 1);
- return;
+ /**
+ * Get the next valid message from the message view array.
+ * Skip corrupted messages and return the first valid one.
+ * Returns null if there are no more messages.
+ *
+ * @param pq - The process queue
+ * @param messageViews - The message views
+ * @param startIndex - The index to start searching from
+ * @return {{ messageView: MessageView; nextIndex: number } | null} The next
valid message and the index after it, or null
+ */
+ protected getNextValidMessage(
+ pq: ProcessQueue,
+ messageViews: MessageView[],
+ startIndex: number,
+ ): { messageView: MessageView; nextIndex: number } | null {
+ for (let i = startIndex; i < messageViews.length; i++) {
+ const messageView = messageViews[i];
+ if (messageView.corrupted) {
+ // Discard corrupted message.
+ this.#logger.error('Message is corrupted for FIFO consumption, prepare
to discard it, mq=%s, '
+ + 'messageId=%s, clientId=%s',
+ pq.getMessageQueue(), messageView.messageId, this.clientId);
+ pq.discardFifoMessage(messageView);
+ continue;
+ }
+ return { messageView, nextIndex: i + 1 };
}
-
- this.consumeMessage(messageView)
- .then(result => pq.eraseFifoMessage(messageView, result))
- .then(() => this.#consumeIteratively(pq, messageViews, index + 1))
- .catch(() => this.#consumeIteratively(pq, messageViews, index + 1));
+ return null;
}
}
diff --git a/nodejs/src/consumer/LiteFifoConsumeService.ts
b/nodejs/src/consumer/LiteFifoConsumeService.ts
new file mode 100644
index 00000000..29f39ab3
--- /dev/null
+++ b/nodejs/src/consumer/LiteFifoConsumeService.ts
@@ -0,0 +1,103 @@
+/**
+ * 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 { MessageView } from '../message';
+import { ConsumeResultSuspend } from './ConsumeResult';
+import { FifoConsumeService } from './FifoConsumeService';
+import { MessageListener } from './MessageListener';
+import type { ProcessQueue } from './ProcessQueue';
+import { ILogger, getDefaultLogger } from '../client/Logger';
+
+/**
+ * FIFO consume service for lite push consumer.
+ *
+ * <p>Similar to {@link FifoConsumeService}, but groups messages by lite topic
+ * instead of message group. This ensures messages with the same lite topic are
+ * consumed sequentially, while messages with different lite topics can be
+ * consumed in parallel when the accelerator is enabled.</p>
+ *
+ * <p>When the message listener returns {@link ConsumeResultSuspend}, all
+ * remaining messages with the same lite topic in the current batch are also
+ * suspended, while messages with different lite topics continue to be
consumed.</p>
+ */
+export class LiteFifoConsumeService extends FifoConsumeService {
+ readonly #logger: ILogger;
+
+ constructor(clientId: string, messageListener: MessageListener,
enableFifoConsumeAccelerator?: boolean) {
+ super(clientId, messageListener, enableFifoConsumeAccelerator);
+ this.#logger = getDefaultLogger();
+ }
+
+ /**
+ * Use lite topic as the group key for FIFO consumption.
+ *
+ * @param messageView - The message view
+ * @return {string} The lite topic, empty string if not present
+ */
+ protected getMessageGroupKey(messageView: MessageView): string {
+ return messageView.liteTopic || '';
+ }
+
+ protected consumeIteratively(pq: ProcessQueue, messageViews: MessageView[],
startIndex: number): Promise<void> {
+ if (!pq || !messageViews) {
+ this.#logger.error('[Bug] Invalid arguments for consumeIteratively,
pq=%s, messageViews=%s, clientId=%s',
+ pq, messageViews, this.clientId);
+ return Promise.resolve();
+ }
+
+ const next = this.getNextValidMessage(pq, messageViews, startIndex);
+ if (!next) {
+ return Promise.resolve();
+ }
+ const { messageView, nextIndex } = next;
+
+ return this.consumeMessage(messageView)
+ .then(async result => {
+ await pq.eraseFifoMessage(messageView, result);
+ return result;
+ })
+ .then(async result => {
+ if (result instanceof ConsumeResultSuspend) {
+ const currentLiteTopic = messageView.liteTopic ?? '';
+ const continuation: MessageView[] = [];
+ const suspendTasks: Array<Promise<void>> = [];
+
+ // Suspend all messages with the same liteTopic in this batch.
+ for (let i = nextIndex; i < messageViews.length; i++) {
+ const msg = messageViews[i];
+ if ((msg.liteTopic ?? '') === currentLiteTopic) {
+ suspendTasks.push(pq.eraseFifoMessage(msg, result));
+ } else {
+ continuation.push(msg);
+ }
+ }
+
+ this.#logger.debug?.('LiteFifoConsumeService suspend, liteTopic=%s,
suspendCount=%d, continuationCount=%d, '
+ + 'clientId=%s', currentLiteTopic, suspendTasks.length,
continuation.length, this.clientId);
+
+ await Promise.all(suspendTasks);
+ return this.consumeIteratively(pq, continuation, 0);
+ }
+ return this.consumeIteratively(pq, messageViews, nextIndex);
+ })
+ .catch(err => {
+ this.#logger.error('[Bug] Exception raised in lite FIFO consumption
callback, clientId=%s, error=%s',
+ this.clientId, err);
+ return this.consumeIteratively(pq, messageViews, nextIndex);
+ });
+ }
+}
diff --git a/nodejs/src/consumer/LitePushConsumer.ts
b/nodejs/src/consumer/LitePushConsumer.ts
index 3517150e..a0df4ff1 100644
--- a/nodejs/src/consumer/LitePushConsumer.ts
+++ b/nodejs/src/consumer/LitePushConsumer.ts
@@ -91,6 +91,10 @@ export interface LitePushConsumerOptions extends
BaseClientOptions {
consumerGroup: string;
bindTopic: string;
messageListener: MessageListener;
+ maxCacheMessageCount?: number;
+ maxCacheMessageSizeInBytes?: number;
+ consumptionThreadCount?: number;
+ enableFifoConsumeAccelerator?: boolean;
}
const CONSUMER_GROUP_PATTERN = /^[a-zA-Z0-9_-]+$/;
@@ -178,7 +182,7 @@ export class LitePushConsumerBuilder {
if (maxCacheMessageCount <= 0) {
throw new Error('maxCacheMessageCount should be positive');
}
- // TODO: Will be used when implementing full consumer
+ this.options.maxCacheMessageCount = maxCacheMessageCount;
return this;
}
@@ -193,7 +197,7 @@ export class LitePushConsumerBuilder {
if (maxCacheMessageSizeInBytes <= 0) {
throw new Error('maxCacheMessageSizeInBytes should be positive');
}
- // TODO: Will be used when implementing full consumer
+ this.options.maxCacheMessageSizeInBytes = maxCacheMessageSizeInBytes;
return this;
}
@@ -208,7 +212,25 @@ export class LitePushConsumerBuilder {
if (consumptionThreadCount <= 0) {
throw new Error('consumptionThreadCount should be positive');
}
- // TODO: Will be used when implementing full consumer
+ // Note: Node.js uses a single-threaded event loop model, so this
configuration
+ // has no effect on actual consumption concurrency. It is retained solely
for
+ // API compatibility with the other clients. Consumption concurrency in
Node.js
+ // is naturally managed by the event loop and Promise-based async
scheduling.
+ this.options.consumptionThreadCount = consumptionThreadCount;
+ return this;
+ }
+
+ /**
+ * Set enable fifo consume accelerator.
+ *
+ * <p>If enabled, messages with different messageGroups are consumed in
parallel
+ * while messages within the same messageGroup are consumed sequentially.</p>
+ *
+ * @param enableFifoConsumeAccelerator - Whether to enable FIFO consume
accelerator
+ * @return This builder instance
+ */
+ setEnableFifoConsumeAccelerator(enableFifoConsumeAccelerator: boolean):
LitePushConsumerBuilder {
+ this.options.enableFifoConsumeAccelerator = enableFifoConsumeAccelerator;
return this;
}
@@ -247,6 +269,10 @@ export class LitePushConsumerBuilder {
sessionCredentials: this.options.sessionCredentials,
requestTimeout: this.options.requestTimeout,
logger: this.options.logger,
+ maxCacheMessageCount: this.options.maxCacheMessageCount,
+ maxCacheMessageSizeInBytes: this.options.maxCacheMessageSizeInBytes,
+ consumptionThreadCount: this.options.consumptionThreadCount,
+ enableFifoConsumeAccelerator: this.options.enableFifoConsumeAccelerator,
};
const litePushConsumer = new LitePushConsumerImpl(options);
diff --git a/nodejs/src/consumer/LitePushConsumerImpl.ts
b/nodejs/src/consumer/LitePushConsumerImpl.ts
index fc977323..37440b51 100644
--- a/nodejs/src/consumer/LitePushConsumerImpl.ts
+++ b/nodejs/src/consumer/LitePushConsumerImpl.ts
@@ -24,6 +24,10 @@ import { LitePushConsumer } from './LitePushConsumer';
import { OffsetOption } from './OffsetOption';
import { LiteSubscriptionManager } from './LiteSubscriptionManager';
import { FilterExpression } from './FilterExpression';
+import { LiteFifoConsumeService } from './LiteFifoConsumeService';
+import { LiteStandardConsumeService } from './LiteStandardConsumeService';
+import { ConsumeService } from './ConsumeService';
+import { MessageListener } from './MessageListener';
export interface LitePushConsumerOptions extends PushConsumerOptions {
bindTopic: string;
@@ -50,6 +54,8 @@ export interface LitePushConsumerOptions extends
PushConsumerOptions {
export class LitePushConsumerImpl extends PushConsumer implements
LitePushConsumer {
private readonly liteSubscriptionManager: LiteSubscriptionManager;
private readonly bindTopic: Resource;
+ readonly #messageListener: MessageListener;
+ readonly #enableFifoConsumeAccelerator: boolean;
constructor(options: LitePushConsumerOptions) {
// Create subscription expressions with bind topic
@@ -61,6 +67,9 @@ export class LitePushConsumerImpl extends PushConsumer
implements LitePushConsum
subscriptions,
} as PushConsumerOptions);
+ this.#messageListener = options.messageListener;
+ this.#enableFifoConsumeAccelerator = options.enableFifoConsumeAccelerator
?? false;
+
this.bindTopic = new Resource(options.namespace, options.bindTopic);
const groupResource = new Resource(options.namespace,
options.consumerGroup);
@@ -71,6 +80,27 @@ export class LitePushConsumerImpl extends PushConsumer
implements LitePushConsum
);
}
+ /**
+ * Create the consume service for lite push consumer.
+ *
+ * <p>Lite push consumer uses lite-topic-specific consume services to ensure
+ * correct grouping and retry behavior.</p>
+ */
+ protected createConsumeService(): ConsumeService {
+ if (!this.#messageListener) {
+ throw new Error('messageListener should not be null');
+ }
+ if (this.getPushConsumerSettings().isFifo()) {
+ this.logger.info(
+ 'Create lite FIFO consume service, consumerGroup=%s, clientId=%s,
enableFifoConsumeAccelerator=%s',
+ this.consumerGroup, this.clientId, this.#enableFifoConsumeAccelerator,
+ );
+ return new LiteFifoConsumeService(this.clientId, this.#messageListener,
this.#enableFifoConsumeAccelerator);
+ }
+ this.logger.info('Create lite standard consume service, consumerGroup=%s,
clientId=%s', this.consumerGroup, this.clientId);
+ return new LiteStandardConsumeService(this.clientId,
this.#messageListener);
+ }
+
/**
* Get the client type.
*
@@ -137,6 +167,9 @@ export class LitePushConsumerImpl extends PushConsumer
implements LitePushConsum
async subscribeLite(liteTopic: string, offsetOption: OffsetOption):
Promise<void>;
async subscribeLite(liteTopic: string, offsetOption?: OffsetOption):
Promise<void> {
+ if (!liteTopic || liteTopic.trim().length === 0) {
+ throw new Error('liteTopic should not be blank');
+ }
await this.liteSubscriptionManager.subscribeLite(liteTopic, offsetOption
?? null);
}
@@ -149,6 +182,9 @@ export class LitePushConsumerImpl extends PushConsumer
implements LitePushConsum
* @param liteTopic - The name of the lite topic to unsubscribe from
*/
async unsubscribeLite(liteTopic: string): Promise<void> {
+ if (!liteTopic || liteTopic.trim().length === 0) {
+ throw new Error('liteTopic should not be blank');
+ }
await this.liteSubscriptionManager.unsubscribeLite(liteTopic);
}
diff --git a/nodejs/src/consumer/LiteStandardConsumeService.ts
b/nodejs/src/consumer/LiteStandardConsumeService.ts
new file mode 100644
index 00000000..f3d93ced
--- /dev/null
+++ b/nodejs/src/consumer/LiteStandardConsumeService.ts
@@ -0,0 +1,73 @@
+/**
+ * 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 { MessageView } from '../message';
+import { ConsumeService } from './ConsumeService';
+import { MessageListener } from './MessageListener';
+import type { ProcessQueue } from './ProcessQueue';
+import { ILogger, getDefaultLogger } from '../client/Logger';
+
+/**
+ * Standard consume service for lite push consumer.
+ *
+ * <p>Similar to {@link StandardConsumeService}, but uses
+ * {@link ProcessQueue#eraseFifoMessage} for local retry on failure, avoiding
+ * frequent server requests.</p>
+ */
+export class LiteStandardConsumeService extends ConsumeService {
+ readonly #logger: ILogger;
+
+ constructor(clientId: string, messageListener: MessageListener) {
+ super(clientId, messageListener);
+ this.#logger = getDefaultLogger();
+ }
+
+ consume(pq: ProcessQueue, messageViews: MessageView[]): void {
+ if (!pq || !messageViews) {
+ this.#logger.error('[Bug] Invalid arguments for consume, pq=%s,
messageViews=%s, clientId=%s', pq, messageViews, this.clientId);
+ return;
+ }
+
+ this.#logger.debug?.('LiteStandardConsumeService.consume called,
messageCount=%d, clientId=%s', messageViews.length, this.clientId);
+
+ for (const messageView of messageViews) {
+ if (messageView.corrupted) {
+ this.#logger.error('Message is corrupted for lite standard
consumption, prepare to discard it, mq=%s, '
+ + 'messageId=%s, clientId=%s',
+ pq.getMessageQueue(), messageView.messageId, this.clientId);
+ pq.discardMessage(messageView);
+ continue;
+ }
+
+ this.#logger.debug?.('Consuming lite message, messageId=%s, topic=%s,
liteTopic=%s, clientId=%s',
+ messageView.messageId, messageView.topic, messageView.liteTopic,
this.clientId);
+
+ this.consumeMessage(messageView)
+ .then(result => {
+ this.#logger.debug?.('Lite message consumed successfully,
messageId=%s, result=%s, clientId=%s',
+ messageView.messageId, result, this.clientId);
+ // Use eraseFifoMessage for local retry on failure, avoiding
frequent server requests.
+ pq.eraseFifoMessage(messageView, result);
+ })
+ .catch(err => {
+ // Should never reach here.
+ this.#logger.error('[Bug] Exception raised in lite standard
consumption callback, clientId=%s, error=%s',
+ this.clientId, err);
+ });
+ }
+ }
+}
diff --git a/nodejs/src/consumer/ProcessQueue.ts
b/nodejs/src/consumer/ProcessQueue.ts
index 3d21f802..5fc83387 100644
--- a/nodejs/src/consumer/ProcessQueue.ts
+++ b/nodejs/src/consumer/ProcessQueue.ts
@@ -20,7 +20,7 @@ import { Code } from
'../../proto/apache/rocketmq/v2/definition_pb';
import { MessageView } from '../message';
import { MessageQueue } from '../route';
import { TooManyRequestsException } from '../exception';
-import { ConsumeResult } from './ConsumeResult';
+import { ConsumeResult, ConsumeResultSuspend } from './ConsumeResult';
import { FilterExpression } from './FilterExpression';
import type { PushConsumer } from './PushConsumer';
@@ -73,6 +73,9 @@ export class ProcessQueue {
}
cacheMessages(messageList: MessageView[]): void {
+ if (!messageList || messageList.length === 0) {
+ return;
+ }
for (const messageView of messageList) {
this.#cachedMessages.push(messageView);
this.#cachedMessagesBytes += messageView.body.length;
@@ -187,6 +190,11 @@ export class ProcessQueue {
}
eraseMessage(messageView: MessageView, consumeResult: ConsumeResult): void {
+ if (!messageView) {
+ return;
+ }
+ consumeResult = this.#convertSuspendResultIfNeeded(consumeResult);
+ this.#statsConsumptionResult(consumeResult);
const task = consumeResult === ConsumeResult.SUCCESS
? this.#ackMessage(messageView)
: this.#nackMessage(messageView);
@@ -256,22 +264,51 @@ export class ProcessQueue {
}
async eraseFifoMessage(messageView: MessageView, consumeResult:
ConsumeResult): Promise<void> {
+ if (!messageView) {
+ return;
+ }
+ consumeResult = this.#convertSuspendResultIfNeeded(consumeResult);
+ this.#statsConsumptionResult(consumeResult);
const retryPolicy = this.#consumer.getRetryPolicy();
const maxAttempts = retryPolicy?.getMaxAttempts() ?? 1;
const attempt = messageView.deliveryAttempt ?? 1;
+ const messageId = messageView.messageId;
+ const clientId = (this.#consumer as any).clientId;
if (consumeResult === ConsumeResult.FAILURE && attempt < maxAttempts) {
const nextAttemptDelay = retryPolicy?.getNextAttemptDelay(attempt) ?? 0;
+ // Increment delivery attempt before redelivering
+ messageView.incrementAndGetDeliveryAttempt();
+ (this.#consumer as any).logger?.debug(
+ 'Prepare to redeliver the fifo message because of the consumption
failure, maxAttempts=%d, '
+ + 'attempt=%d, mq=%s, messageId=%s, nextAttemptDelay=%d, clientId=%s',
+ maxAttempts, messageView.deliveryAttempt, this.#mq, messageId,
nextAttemptDelay, clientId,
+ );
// Redeliver the fifo message
const result = await
this.#consumer.getConsumeService().consumeMessage(messageView,
nextAttemptDelay);
await this.eraseFifoMessage(messageView, result);
+ return;
+ }
+
+ if (consumeResult === ConsumeResult.SUCCESS) {
+ await this.#ackMessage(messageView);
+ } else if (consumeResult instanceof ConsumeResultSuspend) {
+ const suspendTimeMs = consumeResult.suspendTimeMs;
+ (this.#consumer as any).logger?.info(
+ 'Suspend consumption, consumerGroup=%s, topic=%s, liteTopic=%s,
messageId=%s, result=%s',
+ this.#consumer.getConsumerGroup(), messageView.topic,
messageView.liteTopic ?? '',
+ messageId, consumeResult,
+ );
+ await this.#changeInvisibleDuration(messageView, suspendTimeMs);
} else {
- const task = consumeResult === ConsumeResult.SUCCESS
- ? this.#ackMessage(messageView)
- : this.#forwardToDeadLetterQueue(messageView);
- await task;
- this.#evictCache(messageView);
+ (this.#consumer as any).logger?.info(
+ 'Failed to consume fifo message finally, run out of attempt times,
maxAttempts=%d, attempt=%d, '
+ + 'mq=%s, messageId=%s, clientId=%s',
+ maxAttempts, attempt, this.#mq, messageId, clientId,
+ );
+ await this.#forwardToDeadLetterQueue(messageView);
}
+ this.#evictCache(messageView);
}
async #forwardToDeadLetterQueue(messageView: MessageView, attempt = 1):
Promise<void> {
@@ -325,6 +362,28 @@ export class ProcessQueue {
}
}
+ #convertSuspendResultIfNeeded(consumeResult: ConsumeResult): ConsumeResult {
+ if (consumeResult instanceof ConsumeResultSuspend) {
+ if (!(this.#consumer as any).isLiteConsumer?.()) {
+ (this.#consumer as any).logger?.warn(
+ 'Only LitePushConsumer support ConsumeResultSuspend! Convert to
ConsumeResult.FAILURE, '
+ + 'consumerGroup=%s, consumerType=%s',
+ this.#consumer.getConsumerGroup(), this.#consumer.constructor.name,
+ );
+ return ConsumeResult.FAILURE;
+ }
+ }
+ return consumeResult;
+ }
+
+ #statsConsumptionResult(consumeResult: ConsumeResult): void {
+ if (consumeResult === ConsumeResult.SUCCESS) {
+ this.#consumer.incrementConsumptionOkQuantity();
+ return;
+ }
+ this.#consumer.incrementConsumptionErrorQuantity();
+ }
+
cachedMessagesCount(): number {
return this.#cachedMessages.length;
}
diff --git a/nodejs/src/consumer/PushConsumer.ts
b/nodejs/src/consumer/PushConsumer.ts
index 62e751cb..b0afaf70 100644
--- a/nodejs/src/consumer/PushConsumer.ts
+++ b/nodejs/src/consumer/PushConsumer.ts
@@ -103,7 +103,7 @@ export class PushConsumer extends Consumer {
await super.startup();
this.logger.info('Super startup completed, clientId=%s', this.clientId);
try {
- this.#consumeService = this.#createConsumeService();
+ this.#consumeService = this.createConsumeService();
// Start scanning assignments periodically
this.startAssignmentScanning();
this.logger.info('Push consumer started successfully, clientId=%s',
this.clientId);
@@ -181,7 +181,7 @@ export class PushConsumer extends Consumer {
// No-op for push consumer; assignments are queried separately
}
- #createConsumeService(): ConsumeService {
+ protected createConsumeService(): ConsumeService {
if (this.#pushSubscriptionSettings.isFifo()) {
return new FifoConsumeService(this.clientId, this.#messageListener,
this.#enableFifoConsumeAccelerator);
}
@@ -323,9 +323,10 @@ export class PushConsumer extends Consumer {
.setInvisibleDuration(createDuration(invisibleDuration))
.setMessageId(messageView.messageId);
- // For lite consumers, must set liteTopic in request
+ // For lite consumers, must set liteTopic and suspend in request
if (this.isLiteConsumer() && messageView.liteTopic) {
request.setLiteTopic(messageView.liteTopic);
+ request.setSuspend(true);
}
return request;
diff --git a/nodejs/src/consumer/index.ts b/nodejs/src/consumer/index.ts
index eadc3bd7..ff7a14ff 100644
--- a/nodejs/src/consumer/index.ts
+++ b/nodejs/src/consumer/index.ts
@@ -29,6 +29,8 @@ export * from './ConsumeTask';
export * from './ConsumeService';
export * from './StandardConsumeService';
export * from './FifoConsumeService';
+export * from './LiteFifoConsumeService';
+export * from './LiteStandardConsumeService';
export * from './ProcessQueue';
export * from './PushConsumer';
export * from './LitePushConsumer';
diff --git a/nodejs/src/message/MessageView.ts
b/nodejs/src/message/MessageView.ts
index 7c191611..0e17d233 100644
--- a/nodejs/src/message/MessageView.ts
+++ b/nodejs/src/message/MessageView.ts
@@ -36,7 +36,7 @@ export class MessageView {
readonly keys: string[];
readonly bornHost: string;
readonly bornTimestamp?: Date;
- readonly deliveryAttempt?: number;
+ deliveryAttempt?: number;
readonly endpoints: Endpoints;
readonly receiptHandle: string;
readonly offset?: number;
@@ -218,4 +218,15 @@ export class MessageView {
getDeliveryAttempt(): number | undefined {
return this.deliveryAttempt;
}
+
+ /**
+ * Increment the delivery attempt and return the new value.
+ * Used by FIFO retry logic to track redelivery attempts.
+ *
+ * @return the incremented delivery attempt
+ */
+ incrementAndGetDeliveryAttempt(): number {
+ this.deliveryAttempt = (this.deliveryAttempt ?? 0) + 1;
+ return this.deliveryAttempt;
+ }
}
diff --git a/nodejs/test/consumer/ConsumeResult.test.ts
b/nodejs/test/consumer/ConsumeResult.test.ts
new file mode 100644
index 00000000..25db9ed5
--- /dev/null
+++ b/nodejs/test/consumer/ConsumeResult.test.ts
@@ -0,0 +1,43 @@
+/**
+ * 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 { describe, it } from 'node:test';
+import * as assert from 'node:assert';
+import { ConsumeResult, ConsumeResultSuspend } from
'../../src/consumer/ConsumeResult';
+
+describe('ConsumeResult', () => {
+ it('should have SUCCESS and FAILURE singletons', () => {
+ assert.strictEqual(ConsumeResult.SUCCESS.name, 'SUCCESS');
+ assert.strictEqual(ConsumeResult.FAILURE.name, 'FAILURE');
+ assert.strictEqual(ConsumeResult.SUCCESS.toString(), 'SUCCESS');
+ assert.strictEqual(ConsumeResult.FAILURE.toString(), 'FAILURE');
+ assert.strictEqual(ConsumeResult.SUCCESS, ConsumeResult.SUCCESS);
+ });
+
+ it('should create ConsumeResultSuspend with valid suspend time', () => {
+ const suspend = ConsumeResultSuspend.of(100);
+ assert.strictEqual(suspend.name, 'SUSPEND');
+ assert.strictEqual(suspend.suspendTimeMs, 100);
+ assert.strictEqual(suspend.toString(), 'SUSPEND(100ms)');
+ });
+
+ it('should reject suspend time less than 50ms', () => {
+ assert.throws(() => {
+ ConsumeResultSuspend.of(49);
+ }, /suspend time cannot be less than 50ms/);
+ });
+});
diff --git a/nodejs/test/consumer/FifoConsumeService.test.ts
b/nodejs/test/consumer/FifoConsumeService.test.ts
index c60e3025..25b48655 100644
--- a/nodejs/test/consumer/FifoConsumeService.test.ts
+++ b/nodejs/test/consumer/FifoConsumeService.test.ts
@@ -20,7 +20,7 @@
*/
import { describe, it } from 'node:test';
-import assert = require('node:assert');
+import * as assert from 'node:assert';
import { FifoConsumeService } from '../../src/consumer/FifoConsumeService';
import { MessageListener } from '../../src/consumer/MessageListener';
import { ConsumeResult } from '../../src/consumer/ConsumeResult';
diff --git a/nodejs/test/consumer/LiteFifoConsumeService.test.ts
b/nodejs/test/consumer/LiteFifoConsumeService.test.ts
new file mode 100644
index 00000000..0697168e
--- /dev/null
+++ b/nodejs/test/consumer/LiteFifoConsumeService.test.ts
@@ -0,0 +1,157 @@
+/**
+ * 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 FIFO Consume Service
+ */
+
+import { describe, it } from 'node:test';
+import * as assert from 'node:assert';
+import { LiteFifoConsumeService } from
'../../src/consumer/LiteFifoConsumeService';
+import { MessageListener } from '../../src/consumer/MessageListener';
+import { ConsumeResult, ConsumeResultSuspend } from
'../../src/consumer/ConsumeResult';
+import type { ProcessQueue } from '../../src/consumer/ProcessQueue';
+import { MessageView } from '../../src/message';
+
+describe('LiteFifoConsumeService', () => {
+ const createMessageListener = (result: ConsumeResult =
ConsumeResult.SUCCESS): MessageListener => {
+ return {
+ // eslint-disable-next-line @typescript-eslint/no-unused-vars
+ async consume(_messageView: MessageView): Promise<ConsumeResult> {
+ return result;
+ },
+ };
+ };
+
+ const createMockProcessQueue = (): ProcessQueue => {
+ const erased: Array<{ messageView: MessageView; result: ConsumeResult }> =
[];
+ return {
+ eraseFifoMessage: async (messageView: MessageView, result:
ConsumeResult) => {
+ erased.push({ messageView, result });
+ },
+ discardFifoMessage: () => { /* noop */ },
+ getErased: () => erased,
+ } as unknown as ProcessQueue;
+ };
+
+ const createMessageView = (id: string, liteTopic: string): MessageView => {
+ return {
+ messageId: id,
+ liteTopic,
+ corrupted: false,
+ } as unknown as MessageView;
+ };
+
+ it('should support enableFifoConsumeAccelerator option', () => {
+ const messageListener = createMessageListener();
+
+ // Create with accelerator enabled
+ const serviceWithAccelerator = new LiteFifoConsumeService('test-client',
messageListener, true);
+ assert.ok(serviceWithAccelerator, 'Should create service with accelerator
enabled');
+
+ // Create with accelerator disabled (default)
+ const serviceWithoutAccelerator = new
LiteFifoConsumeService('test-client', messageListener, false);
+ assert.ok(serviceWithoutAccelerator, 'Should create service with
accelerator disabled');
+
+ // Create without specifying (default to false)
+ const serviceDefault = new LiteFifoConsumeService('test-client',
messageListener);
+ assert.ok(serviceDefault, 'Should create service with default accelerator
setting');
+ });
+
+ it('should group messages by lite topic', () => {
+ const messageListener = createMessageListener();
+ const service = new LiteFifoConsumeService('test-client', messageListener,
true);
+
+ // Expose protected method for testing
+ const getMessageGroupKey = (service as
any).getMessageGroupKey.bind(service);
+
+ const messageWithLiteTopic = {
+ liteTopic: 'lite-topic-a',
+ messageGroup: 'fifo-group',
+ } as MessageView;
+ assert.strictEqual(getMessageGroupKey(messageWithLiteTopic),
'lite-topic-a',
+ 'Should use lite topic as group key');
+
+ const messageWithoutLiteTopic = {
+ liteTopic: undefined,
+ messageGroup: 'fifo-group',
+ } as MessageView;
+ assert.strictEqual(getMessageGroupKey(messageWithoutLiteTopic), '',
+ 'Should return empty string when lite topic is absent');
+ });
+
+ it('should suspend remaining messages with the same lite topic', async () =>
{
+ const suspendResult = ConsumeResultSuspend.of(100);
+ const messageListener = createMessageListener(suspendResult);
+ const service = new LiteFifoConsumeService('test-client', messageListener,
false);
+ const pq = createMockProcessQueue();
+
+ const msg1 = createMessageView('1', 'topic-a');
+ const msg2 = createMessageView('2', 'topic-a');
+ const msg3 = createMessageView('3', 'topic-b');
+ const msg4 = createMessageView('4', 'topic-a');
+
+ const consumeIteratively = (service as
any).consumeIteratively.bind(service);
+ await consumeIteratively(pq, [ msg1, msg2, msg3, msg4 ], 0);
+
+ const erased = (pq as any).getErased();
+ // msg1 consumed and suspended, msg2/msg4 suspended without consume, msg3
consumed in continuation
+ assert.strictEqual(erased.length, 4, 'All messages should be erased');
+ assert.strictEqual(erased[0].messageView.messageId, '1');
+ assert.strictEqual(erased[0].result, suspendResult);
+ assert.strictEqual(erased[1].messageView.messageId, '2');
+ assert.strictEqual(erased[1].result, suspendResult);
+ assert.strictEqual(erased[2].messageView.messageId, '4');
+ assert.strictEqual(erased[2].result, suspendResult);
+ assert.strictEqual(erased[3].messageView.messageId, '3');
+ assert.strictEqual(erased[3].result, suspendResult);
+ });
+
+ it('should continue consuming messages with different lite topic after
suspend', async () => {
+ const suspendResult = ConsumeResultSuspend.of(100);
+ let callCount = 0;
+ const messageListener: MessageListener = {
+ async consume(messageView: MessageView): Promise<ConsumeResult> {
+ callCount++;
+ if (messageView.liteTopic === 'topic-a') {
+ return suspendResult;
+ }
+ return ConsumeResult.SUCCESS;
+ },
+ };
+ const service = new LiteFifoConsumeService('test-client', messageListener,
false);
+ const pq = createMockProcessQueue();
+
+ const msg1 = createMessageView('1', 'topic-a');
+ const msg2 = createMessageView('2', 'topic-b');
+ const msg3 = createMessageView('3', 'topic-a');
+
+ const consumeIteratively = (service as
any).consumeIteratively.bind(service);
+ await consumeIteratively(pq, [ msg1, msg2, msg3 ], 0);
+
+ const erased = (pq as any).getErased();
+ // msg1 consumed and suspended, msg3 suspended without consume, msg2
consumed in continuation
+ assert.strictEqual(callCount, 2, 'Only msg1 and msg2 should be consumed by
listener');
+ assert.strictEqual(erased.length, 3);
+ assert.strictEqual(erased[0].messageView.messageId, '1');
+ assert.strictEqual(erased[0].result, suspendResult);
+ assert.strictEqual(erased[1].messageView.messageId, '3');
+ assert.strictEqual(erased[1].result, suspendResult);
+ assert.strictEqual(erased[2].messageView.messageId, '2');
+ assert.strictEqual(erased[2].result, ConsumeResult.SUCCESS);
+ });
+});
diff --git a/nodejs/test/consumer/LitePushConsumerBuilder.test.ts
b/nodejs/test/consumer/LitePushConsumerBuilder.test.ts
index ca02efe4..9ee543b8 100644
--- a/nodejs/test/consumer/LitePushConsumerBuilder.test.ts
+++ b/nodejs/test/consumer/LitePushConsumerBuilder.test.ts
@@ -155,6 +155,20 @@ describe('LitePushConsumerBuilder', () => {
});
});
+ describe('setEnableFifoConsumeAccelerator()', () => {
+ it('should accept true value', () => {
+ const builder = new LitePushConsumerBuilder();
+ const result = builder.setEnableFifoConsumeAccelerator(true);
+ assert.strictEqual(result, builder);
+ });
+
+ it('should accept false value', () => {
+ const builder = new LitePushConsumerBuilder();
+ const result = builder.setEnableFifoConsumeAccelerator(false);
+ assert.strictEqual(result, builder);
+ });
+ });
+
describe('build()', () => {
it('should throw error when clientConfiguration not set', async () => {
const builder = new LitePushConsumerBuilder();
@@ -172,10 +186,10 @@ describe('LitePushConsumerBuilder', () => {
builder.setMessageListener(createMockMessageListener());
builder.bindTopic('test-topic');
- // Should reject with "not implemented" error since we haven't
implemented the full consumer yet
+ // Should reject because startup() requires a real RocketMQ server
connection
await assert.rejects(
() => builder.build(),
- /LitePushConsumerImpl not yet implemented/,
+ /Failed to start|Connection|ECONNREFUSED|timeout|Timeout|Error/,
);
});
@@ -206,16 +220,17 @@ describe('LitePushConsumerBuilder', () => {
await assert.rejects(() => builder.build(), /bindTopic has not been set
yet/);
});
- it('should throw not implemented error when all params set', async () => {
+ it('should throw connection error when all params set but no server',
async () => {
const builder = new LitePushConsumerBuilder();
builder.setClientConfiguration(createMockClientConfig());
builder.setConsumerGroup('test-group');
builder.setMessageListener(createMockMessageListener());
builder.bindTopic('test-topic');
+ // Should reject because startup() requires a real RocketMQ server
connection
await assert.rejects(
() => builder.build(),
- /LitePushConsumerImpl not yet implemented/,
+ /Failed to start|Connection|ECONNREFUSED|timeout|Timeout|Error/,
);
});
});
diff --git a/nodejs/test/message/LiteMessage.test.ts
b/nodejs/test/message/LiteMessage.test.ts
index a964f00d..e58c874c 100644
--- a/nodejs/test/message/LiteMessage.test.ts
+++ b/nodejs/test/message/LiteMessage.test.ts
@@ -54,7 +54,7 @@ describe('Lite Message', () => {
liteTopic: 'lite-topic-name',
priority: 5,
});
- }, /priority and liteTopic should not be set at same time/);
+ }, /priority is mutually exclusive with/);
});
it('should throw error when liteTopic and messageGroup are set together', ()
=> {
@@ -65,7 +65,7 @@ describe('Lite Message', () => {
liteTopic: 'lite-topic-name',
messageGroup: 'test-group',
});
- }, /liteTopic and messageGroup should not be set at same time/);
+ }, /liteTopic is mutually exclusive with/);
});
it('should throw error when liteTopic and deliveryTimestamp are set
together', () => {
@@ -76,7 +76,7 @@ describe('Lite Message', () => {
liteTopic: 'lite-topic-name',
deliveryTimestamp: new Date(Date.now() + 60000),
});
- }, /liteTopic and deliveryTimestamp should not be set at same time/);
+ }, /liteTopic is mutually exclusive with/);
});
it('should accept empty string liteTopic', () => {
diff --git a/nodejs/test/message/PriorityMessage.test.ts
b/nodejs/test/message/PriorityMessage.test.ts
index 1721f133..104306b1 100644
--- a/nodejs/test/message/PriorityMessage.test.ts
+++ b/nodejs/test/message/PriorityMessage.test.ts
@@ -64,7 +64,7 @@ describe('Priority Message', () => {
priority: 1,
deliveryTimestamp: new Date(Date.now() + 60000),
});
- }, /priority and deliveryTimestamp should not be set at same time/);
+ }, /priority is mutually exclusive with/);
});
it('should throw error when priority and messageGroup are set together', ()
=> {
@@ -75,7 +75,7 @@ describe('Priority Message', () => {
priority: 1,
messageGroup: 'test-group',
});
- }, /priority and messageGroup should not be set at same time/);
+ }, /priority is mutually exclusive with/);
});
it('should accept priority = 0', () => {
diff --git a/nodejs/test/message/PublishingLiteMessage.test.ts
b/nodejs/test/message/PublishingLiteMessage.test.ts
index ca20319c..4c7b8d43 100644
--- a/nodejs/test/message/PublishingLiteMessage.test.ts
+++ b/nodejs/test/message/PublishingLiteMessage.test.ts
@@ -54,6 +54,6 @@ describe('PublishingMessage with LITE type', () => {
liteTopic: 'lite-topic-name',
priority: 5,
});
- }, /priority and liteTopic should not be set at same time/);
+ }, /priority is mutually exclusive with/);
});
});
