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 3357f9c0 fix(nodejs): align retry units, endpoints parsing (#1373)
3357f9c0 is described below
commit 3357f9c077bfc15caa506a6271ecdf0e288f85a0
Author: zhaohai <[email protected]>
AuthorDate: Tue Sep 22 10:50:19 2026 +0800
fix(nodejs): align retry units, endpoints parsing (#1373)
* fix(nodejs): correct retry backoff units, endpoints parsing and status
mappings
- Retry delays are milliseconds (the protobuf contract's unit was
misimplemented); inherited backoff preserves sub-second precision from
the server Duration (seconds * 1000 + nanos / 1e6)
- Endpoints string parsing rewritten: strip the optional http(s)://
prefix, parse bracketed and bare IPv6 addresses, split the port via
lastIndexOf(':')
- StatusChecker new mappings: ILLEGAL_LITE_TOPIC -> BadRequestException,
MESSAGE_BODY_EMPTY -> PayloadEmptyException,
LITE_SUBSCRIPTION_QUOTA_EXCEEDED -> LiteSubscriptionQuotaExceededException
- TelemetrySession: fix the reconnect timer leak; release() clears the
timer and guards reconnection, error/end events dedupe through a
single reconnect schedule
- Add tsconfig.test.json for compiling src and test together
Tests: rename regression suite to test/fixes.test.ts (13 offline cases,
all pass); tsc clean for both the prod and the test config
* fix(nodejs): keep auto-reconnect alive after telemetry session refresh
refresh() tore down the old stream through release(), which permanently
set #released and thus made #scheduleRenewStream() a no-op forever. Any
later stream error or completion was silently never reconnected.
Refresh now only cancels a pending reconnect timer and closes the stream,
leaving release() as the sole shutdown path.
---
nodejs/src/client/TelemetrySession.ts | 50 +++++--
nodejs/src/consumer/PushSubscriptionSettings.ts | 24 +++-
.../PayloadEmptyException.ts} | 10 +-
nodejs/src/exception/StatusChecker.ts | 7 +
nodejs/src/exception/index.ts | 1 +
nodejs/src/retry/CustomizedBackoffRetryPolicy.ts | 81 +++++++++++
nodejs/src/retry/ExponentialBackoffRetryPolicy.ts | 17 ++-
nodejs/src/retry/RetryPolicy.ts | 2 +-
nodejs/src/retry/index.ts | 1 +
nodejs/src/route/Endpoints.ts | 30 +++-
nodejs/test/fixes.test.ts | 156 +++++++++++++++++++++
11 files changed, 352 insertions(+), 27 deletions(-)
diff --git a/nodejs/src/client/TelemetrySession.ts
b/nodejs/src/client/TelemetrySession.ts
index d0fcf231..782481fb 100644
--- a/nodejs/src/client/TelemetrySession.ts
+++ b/nodejs/src/client/TelemetrySession.ts
@@ -27,6 +27,8 @@ export class TelemetrySession {
#logger: ILogger;
#stream: ClientDuplexStream<TelemetryCommand, TelemetryCommand>;
#isRefreshing = false;
+ #released = false;
+ #reconnectTimer?: NodeJS.Timeout;
constructor(baseClient: BaseClient, endpoints: Endpoints, logger: ILogger) {
this.#endpoints = endpoints;
@@ -38,8 +40,20 @@ export class TelemetrySession {
}
release() {
+ if (this.#released) {
+ return;
+ }
+ this.#released = true;
+ if (this.#reconnectTimer) {
+ clearTimeout(this.#reconnectTimer);
+ this.#reconnectTimer = undefined;
+ }
this.#logger.info('Begin to release telemetry session, endpoints=%s,
clientId=%s',
this.#endpoints, this.#baseClient.clientId);
+ this.#closeStream();
+ }
+
+ #closeStream() {
try {
this.#stream.end();
this.#stream.removeAllListeners();
@@ -70,7 +84,11 @@ export class TelemetrySession {
this.#endpoints, this.#baseClient.clientId);
try {
- this.release();
+ if (this.#reconnectTimer) {
+ clearTimeout(this.#reconnectTimer);
+ this.#reconnectTimer = undefined;
+ }
+ this.#closeStream();
this.#renewStream(false);
} catch (err) {
this.#logger.error('Failed to refresh telemetry session, endpoints=%s,
clientId=%s, error=%s',
@@ -150,21 +168,35 @@ export class TelemetrySession {
}
}
+ /**
+ * Schedule a telemetry stream renewal. The timer is tracked so that it can
+ * be cancelled on release(), and a pending timer doubles as a guard to
+ * prevent error/end events from scheduling duplicate reconnects.
+ */
+ #scheduleRenewStream() {
+ if (this.#released || this.#reconnectTimer) {
+ return;
+ }
+ this.#reconnectTimer = setTimeout(() => {
+ this.#reconnectTimer = undefined;
+ if (this.#released) {
+ return;
+ }
+ this.#renewStream(false);
+ }, 1000);
+ }
+
#onError(err: Error) {
this.#logger.error('Exception raised from stream response observer,
endpoints=%s, clientId=%s, error=%s',
this.#endpoints, this.#baseClient.clientId, err);
- this.release();
- setTimeout(() => {
- this.#renewStream(false);
- }, 1000);
+ this.#closeStream();
+ this.#scheduleRenewStream();
}
#onEnd() {
this.#logger.info('Receive completion for stream response observer,
endpoints=%s, clientId=%s',
this.#endpoints, this.#baseClient.clientId);
- this.release();
- setTimeout(() => {
- this.#renewStream(false);
- }, 1000);
+ this.#closeStream();
+ this.#scheduleRenewStream();
}
}
diff --git a/nodejs/src/consumer/PushSubscriptionSettings.ts
b/nodejs/src/consumer/PushSubscriptionSettings.ts
index d02ca277..2f3b1a8c 100644
--- a/nodejs/src/consumer/PushSubscriptionSettings.ts
+++ b/nodejs/src/consumer/PushSubscriptionSettings.ts
@@ -21,9 +21,10 @@ import {
Subscription,
RetryPolicy as RetryPolicyPB,
} from '../../proto/apache/rocketmq/v2/definition_pb';
+import { Duration } from 'google-protobuf/google/protobuf/duration_pb';
import { Endpoints } from '../route';
import { Settings, UserAgent } from '../client';
-import { ExponentialBackoffRetryPolicy, RetryPolicy } from '../retry';
+import { CustomizedBackoffRetryPolicy, ExponentialBackoffRetryPolicy,
RetryPolicy } from '../retry';
import { createDuration, createResource } from '../util';
import { FilterExpression } from './FilterExpression';
@@ -102,20 +103,29 @@ export class PushSubscriptionSettings extends Settings {
}
const backoffPolicy = settings.getBackoffPolicy();
if (backoffPolicy) {
+ // Convert protobuf Duration (seconds + nanos) to milliseconds without
+ // losing sub-second precision.
+ const toMillis = (duration?: Duration) =>
+ duration ? duration.getSeconds() * 1000 + duration.getNanos() / 1e6 :
0;
switch (backoffPolicy.getStrategyCase()) {
case RetryPolicyPB.StrategyCase.EXPONENTIAL_BACKOFF: {
- const exponential =
backoffPolicy.getExponentialBackoff()!.toObject();
+ const exponential = backoffPolicy.getExponentialBackoff()!;
this.retryPolicy = new ExponentialBackoffRetryPolicy(
backoffPolicy.getMaxAttempts(),
- exponential.initial?.seconds,
- exponential.max?.seconds,
- exponential.multiplier,
+ toMillis(exponential.getInitial()),
+ toMillis(exponential.getMax()),
+ exponential.getMultiplier(),
);
break;
}
- case RetryPolicyPB.StrategyCase.CUSTOMIZED_BACKOFF:
- // CustomizedBackoffRetryPolicy not yet implemented in Node.js
+ case RetryPolicyPB.StrategyCase.CUSTOMIZED_BACKOFF: {
+ const customizedBackoff = backoffPolicy.getCustomizedBackoff()!;
+ const durations = customizedBackoff.getNextList().map((duration:
Duration) => toMillis(duration));
+ if (durations.length > 0) {
+ this.retryPolicy = new CustomizedBackoffRetryPolicy(durations,
backoffPolicy.getMaxAttempts());
+ }
break;
+ }
default:
break;
}
diff --git a/nodejs/src/retry/index.ts
b/nodejs/src/exception/PayloadEmptyException.ts
similarity index 75%
copy from nodejs/src/retry/index.ts
copy to nodejs/src/exception/PayloadEmptyException.ts
index 79d29d0f..8625e6f8 100644
--- a/nodejs/src/retry/index.ts
+++ b/nodejs/src/exception/PayloadEmptyException.ts
@@ -15,5 +15,11 @@
* limitations under the License.
*/
-export * from './ExponentialBackoffRetryPolicy';
-export * from './RetryPolicy';
+import { ClientException } from './ClientException';
+
+export class PayloadEmptyException extends ClientException {
+ constructor(code: number, message: string, requestId?: string) {
+ super(code, message, requestId);
+ this.name = 'PayloadEmptyException';
+ }
+}
diff --git a/nodejs/src/exception/StatusChecker.ts
b/nodejs/src/exception/StatusChecker.ts
index bb7cdae4..dfd68167 100644
--- a/nodejs/src/exception/StatusChecker.ts
+++ b/nodejs/src/exception/StatusChecker.ts
@@ -19,8 +19,10 @@ import { Status, Code } from
'../../proto/apache/rocketmq/v2/definition_pb';
import { BadRequestException } from './BadRequestException';
import { ForbiddenException } from './ForbiddenException';
import { InternalErrorException } from './InternalErrorException';
+import { LiteSubscriptionQuotaExceededException } from
'./LiteSubscriptionQuotaExceededException';
import { LiteTopicQuotaExceededException } from
'./LiteTopicQuotaExceededException';
import { NotFoundException } from './NotFoundException';
+import { PayloadEmptyException } from './PayloadEmptyException';
import { PayloadTooLargeException } from './PayloadTooLargeException';
import { PaymentRequiredException } from './PaymentRequiredException';
import { ProxyTimeoutException } from './ProxyTimeoutException';
@@ -44,6 +46,7 @@ export class StatusChecker {
case Code.ILLEGAL_MESSAGE_KEY:
case Code.ILLEGAL_MESSAGE_GROUP:
case Code.ILLEGAL_MESSAGE_PROPERTY_KEY:
+ case Code.ILLEGAL_LITE_TOPIC:
case Code.INVALID_TRANSACTION_ID:
case Code.ILLEGAL_MESSAGE_ID:
case Code.ILLEGAL_FILTER_EXPRESSION:
@@ -71,6 +74,8 @@ export class StatusChecker {
case Code.PAYLOAD_TOO_LARGE:
case Code.MESSAGE_BODY_TOO_LARGE:
throw new PayloadTooLargeException(status.code, status.message,
requestId);
+ case Code.MESSAGE_BODY_EMPTY:
+ throw new PayloadEmptyException(status.code, status.message,
requestId);
case Code.TOO_MANY_REQUESTS:
throw new TooManyRequestsException(status.code, status.message,
requestId);
case Code.REQUEST_HEADER_FIELDS_TOO_LARGE:
@@ -90,6 +95,8 @@ export class StatusChecker {
throw new UnsupportedException(status.code, status.message, requestId);
case Code.LITE_TOPIC_QUOTA_EXCEEDED:
throw new LiteTopicQuotaExceededException(status.code, status.message
|| '', requestId);
+ case Code.LITE_SUBSCRIPTION_QUOTA_EXCEEDED:
+ throw new LiteSubscriptionQuotaExceededException(status.code,
requestId ?? null, status.message || '');
default:
throw new UnsupportedException(status.code, status.message, requestId);
}
diff --git a/nodejs/src/exception/index.ts b/nodejs/src/exception/index.ts
index e542e96e..381db8ce 100644
--- a/nodejs/src/exception/index.ts
+++ b/nodejs/src/exception/index.ts
@@ -22,6 +22,7 @@ export * from './InternalErrorException';
export * from './LiteSubscriptionQuotaExceededException';
export * from './LiteTopicQuotaExceededException';
export * from './NotFoundException';
+export * from './PayloadEmptyException';
export * from './PayloadTooLargeException';
export * from './PaymentRequiredException';
export * from './ProxyTimeoutException';
diff --git a/nodejs/src/retry/CustomizedBackoffRetryPolicy.ts
b/nodejs/src/retry/CustomizedBackoffRetryPolicy.ts
new file mode 100644
index 00000000..204130c5
--- /dev/null
+++ b/nodejs/src/retry/CustomizedBackoffRetryPolicy.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.
+ */
+
+import assert from 'node:assert';
+import { Duration } from 'google-protobuf/google/protobuf/duration_pb';
+import {
+ RetryPolicy as RetryPolicyPB,
+ CustomizedBackoff,
+} from '../../proto/apache/rocketmq/v2/definition_pb';
+import { createDuration } from '../util';
+import { RetryPolicy } from './RetryPolicy';
+
+/**
+ * Backoff policy whose durations are customized by the server side,
+ * mirroring Java's CustomizedBackoffRetryPolicy: the Nth attempt waits
+ * durations[N - 1], and once attempts run past the list, the last
+ * duration applies.
+ */
+export class CustomizedBackoffRetryPolicy implements RetryPolicy {
+ #maxAttempts: number;
+ // Backoff durations in milliseconds.
+ #durations: number[];
+
+ constructor(durations: number[], maxAttempts: number) {
+ assert(Array.isArray(durations) && durations.length > 0, 'durations must
not be empty');
+ this.#durations = durations;
+ this.#maxAttempts = maxAttempts;
+ }
+
+ getMaxAttempts(): number {
+ return this.#maxAttempts;
+ }
+
+ getNextAttemptDelay(attempt: number): number {
+ assert(attempt > 0, 'attempt must be positive');
+ return attempt > this.#durations.length
+ ? this.#durations[this.#durations.length - 1]
+ : this.#durations[attempt - 1];
+ }
+
+ inheritBackoff(retryPolicy: RetryPolicyPB): RetryPolicy {
+ assert(retryPolicy.getStrategyCase() ===
RetryPolicyPB.StrategyCase.CUSTOMIZED_BACKOFF,
+ 'strategy must be customized backoff');
+ return new CustomizedBackoffRetryPolicy(
+
CustomizedBackoffRetryPolicy.durationsToMillis(retryPolicy.getCustomizedBackoff()!),
+ this.#maxAttempts);
+ }
+
+ toProtobuf(): RetryPolicyPB {
+ const customizedBackoff = new CustomizedBackoff();
+ for (const duration of this.#durations) {
+ customizedBackoff.addNext(createDuration(duration));
+ }
+ return new RetryPolicyPB()
+ .setMaxAttempts(this.#maxAttempts)
+ .setCustomizedBackoff(customizedBackoff);
+ }
+
+ /**
+ * Convert the protobuf CustomizedBackoff durations (seconds + nanos) to
+ * milliseconds without losing sub-second precision.
+ */
+ static durationsToMillis(customizedBackoff: CustomizedBackoff): number[] {
+ return customizedBackoff.getNextList().map((duration: Duration) =>
+ duration.getSeconds() * 1000 + duration.getNanos() / 1e6);
+ }
+}
diff --git a/nodejs/src/retry/ExponentialBackoffRetryPolicy.ts
b/nodejs/src/retry/ExponentialBackoffRetryPolicy.ts
index 3fc8997b..e851d0e8 100644
--- a/nodejs/src/retry/ExponentialBackoffRetryPolicy.ts
+++ b/nodejs/src/retry/ExponentialBackoffRetryPolicy.ts
@@ -16,16 +16,16 @@
*/
import assert from 'node:assert';
-import { Duration } from 'google-protobuf/google/protobuf/duration_pb';
import {
RetryPolicy as RetryPolicyPB,
ExponentialBackoff,
} from '../../proto/apache/rocketmq/v2/definition_pb';
+import { createDuration } from '../util';
import { RetryPolicy } from './RetryPolicy';
export class ExponentialBackoffRetryPolicy implements RetryPolicy {
#maxAttempts: number;
- // seconds
+ // milliseconds
#initialBackoff: number;
#maxBackoff: number;
#backoffMultiplier: number;
@@ -58,9 +58,14 @@ export class ExponentialBackoffRetryPolicy implements
RetryPolicy {
assert(retryPolicy.getStrategyCase() ===
RetryPolicyPB.StrategyCase.EXPONENTIAL_BACKOFF,
'strategy must be exponential backoff');
const backoff = retryPolicy.getExponentialBackoff()!.toObject();
+ // Convert protobuf Duration (seconds + nanos) to milliseconds without
+ // losing sub-second precision (e.g. an initial backoff of 0.5s used to
+ // be truncated to plain 0, causing immediate retries).
+ const toMillis = (duration?: { seconds?: number; nanos?: number }) =>
+ duration ? (duration.seconds ?? 0) * 1000 + (duration.nanos ?? 0) / 1e6
: 0;
return new ExponentialBackoffRetryPolicy(this.#maxAttempts,
- backoff.initial?.seconds,
- backoff.max?.seconds,
+ toMillis(backoff.initial),
+ toMillis(backoff.max),
backoff.multiplier);
}
@@ -69,8 +74,8 @@ export class ExponentialBackoffRetryPolicy implements
RetryPolicy {
.setMaxAttempts(this.#maxAttempts)
.setExponentialBackoff(
new ExponentialBackoff()
- .setInitial(new Duration().setSeconds(this.#initialBackoff))
- .setMax(new Duration().setSeconds(this.#maxBackoff))
+ .setInitial(createDuration(this.#initialBackoff))
+ .setMax(createDuration(this.#maxBackoff))
.setMultiplier(this.#backoffMultiplier));
}
}
diff --git a/nodejs/src/retry/RetryPolicy.ts b/nodejs/src/retry/RetryPolicy.ts
index 0e559d34..f16dc359 100644
--- a/nodejs/src/retry/RetryPolicy.ts
+++ b/nodejs/src/retry/RetryPolicy.ts
@@ -32,7 +32,7 @@ export interface RetryPolicy {
* Get await time after current attempts, the attempt index starts at 1.
*
* @param attempt current attempt.
- * @return await time in seconds.
+ * @return await time in milliseconds.
*/
getNextAttemptDelay(attempt: number): number;
diff --git a/nodejs/src/retry/index.ts b/nodejs/src/retry/index.ts
index 79d29d0f..119a55e5 100644
--- a/nodejs/src/retry/index.ts
+++ b/nodejs/src/retry/index.ts
@@ -15,5 +15,6 @@
* limitations under the License.
*/
+export * from './CustomizedBackoffRetryPolicy';
export * from './ExponentialBackoffRetryPolicy';
export * from './RetryPolicy';
diff --git a/nodejs/src/route/Endpoints.ts b/nodejs/src/route/Endpoints.ts
index df08382a..840faf5e 100644
--- a/nodejs/src/route/Endpoints.ts
+++ b/nodejs/src/route/Endpoints.ts
@@ -35,7 +35,33 @@ export class Endpoints {
const splits = endpoints.split(';');
this.addressesList = [];
for (const endpoint of splits) {
- const [ host, port ] = endpoint.split(':');
+ // Strip the optional http:// or https:// prefix, mirroring the Java
client.
+ const candidate = endpoint.trim().replace(/^https?:\/\//, '');
+ let host: string;
+ let port: number;
+ if (candidate.startsWith('[')) {
+ // Bracketed IPv6 address, e.g. [::1]:10911 or [fe80::1]
+ const match = candidate.match(/^\[([^\]]+)\](?::(\d+))?$/);
+ if (!match) {
+ throw new TypeError(`Invalid IPv6 endpoint: ${endpoint}`);
+ }
+ host = match[1];
+ port = match[2] ? parseInt(match[2], 10) : DEFAULT_PORT;
+ } else if (isIPv6(candidate)) {
+ // Bare IPv6 address without brackets, e.g. ::1
+ host = candidate;
+ port = DEFAULT_PORT;
+ } else {
+ // IPv4 address or domain name, e.g. 127.0.0.1:10911, example.com:80
+ const index = candidate.lastIndexOf(':');
+ if (index > 0) {
+ host = candidate.substring(0, index);
+ port = parseInt(candidate.substring(index + 1), 10) ||
DEFAULT_PORT;
+ } else {
+ host = candidate;
+ port = DEFAULT_PORT;
+ }
+ }
if (isIPv4(host)) {
this.scheme = AddressScheme.IPV4;
} else if (isIPv6(host)) {
@@ -43,7 +69,7 @@ export class Endpoints {
} else {
this.scheme = AddressScheme.DOMAIN_NAME;
}
- this.addressesList.push({ host, port: parseInt(port) || DEFAULT_PORT
});
+ this.addressesList.push({ host, port });
}
} else {
this.scheme = endpoints.scheme;
diff --git a/nodejs/test/fixes.test.ts b/nodejs/test/fixes.test.ts
new file mode 100644
index 00000000..d230cbc1
--- /dev/null
+++ b/nodejs/test/fixes.test.ts
@@ -0,0 +1,156 @@
+/**
+ * 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.
+ */
+
+/**
+ * Regression tests for defect fixes aligned with the Java client:
+ * - Retry backoff unit contract (milliseconds, sub-second precision kept)
+ * - CustomizedBackoffRetryPolicy implementation
+ * - Endpoints string parsing (IPv6 / http(s) prefix)
+ * - StatusChecker mappings (ILLEGAL_LITE_TOPIC / MESSAGE_BODY_EMPTY /
+ * LITE_SUBSCRIPTION_QUOTA_EXCEEDED)
+ */
+
+import { describe, it } from 'node:test';
+import * as assert from 'node:assert';
+import { Duration } from 'google-protobuf/google/protobuf/duration_pb';
+import {
+ ExponentialBackoffRetryPolicy,
+ CustomizedBackoffRetryPolicy,
+} from '../src/retry';
+import { Endpoints } from '../src/route';
+import {
+ StatusChecker,
+ BadRequestException,
+ PayloadEmptyException,
+ LiteSubscriptionQuotaExceededException,
+} from '../src/exception';
+import { Code, RetryPolicy as RetryPolicyPB, ExponentialBackoff,
CustomizedBackoff } from '../proto/apache/rocketmq/v2/definition_pb';
+import { Status } from '../proto/apache/rocketmq/v2/definition_pb';
+
+function statusOf(code: Code): Status.AsObject {
+ return { code, message: 'mock message', requestId: '' } as unknown as
Status.AsObject;
+}
+
+describe('ExponentialBackoffRetryPolicy (ms contract)', () => {
+ it('should return delays in milliseconds', () => {
+ const policy = new ExponentialBackoffRetryPolicy(3, 1000, 10000, 2);
+ assert.strictEqual(policy.getNextAttemptDelay(1), 1000);
+ assert.strictEqual(policy.getNextAttemptDelay(2), 2000);
+ assert.strictEqual(policy.getNextAttemptDelay(3), 4000);
+ assert.strictEqual(policy.getNextAttemptDelay(4), 8000);
+ // capped by maxBackoff
+ assert.strictEqual(policy.getNextAttemptDelay(5), 10000);
+ });
+
+ it('should keep sub-second precision when inheriting backoff', () => {
+ const retryPolicy = new RetryPolicyPB().setExponentialBackoff(
+ new ExponentialBackoff()
+ .setInitial(new Duration().setSeconds(0).setNanos(500000000)) // 0.5s
+ .setMax(new Duration().setSeconds(3).setNanos(250000000)) // 3.25s
+ .setMultiplier(2));
+ const policy = new
ExponentialBackoffRetryPolicy(3).inheritBackoff(retryPolicy) as
ExponentialBackoffRetryPolicy;
+ assert.strictEqual(policy.getNextAttemptDelay(1), 500);
+ assert.strictEqual(policy.getNextAttemptDelay(2), 1000);
+ assert.strictEqual(policy.getNextAttemptDelay(3), 2000);
+ // capped by max backoff (3.25s, sub-second precision kept)
+ assert.strictEqual(policy.getNextAttemptDelay(4), 3250);
+ });
+
+ it('should serialize back to protobuf without losing precision', () => {
+ const policy = new ExponentialBackoffRetryPolicy(3, 500, 3250, 2);
+ const pb = policy.toProtobuf();
+ const exponential = pb.getExponentialBackoff()!;
+ assert.strictEqual(exponential.getInitial()!.getSeconds(), 0);
+ assert.strictEqual(exponential.getInitial()!.getNanos(), 500000000);
+ assert.strictEqual(exponential.getMax()!.getSeconds(), 3);
+ assert.strictEqual(exponential.getMax()!.getNanos(), 250000000);
+ });
+});
+
+describe('CustomizedBackoffRetryPolicy', () => {
+ it('should return the Nth duration and clamp to the last one', () => {
+ const policy = new CustomizedBackoffRetryPolicy([1000, 5000, 10000], 5);
+ assert.strictEqual(policy.getNextAttemptDelay(1), 1000);
+ assert.strictEqual(policy.getNextAttemptDelay(2), 5000);
+ assert.strictEqual(policy.getNextAttemptDelay(3), 10000);
+ assert.strictEqual(policy.getNextAttemptDelay(4), 10000);
+ assert.strictEqual(policy.getMaxAttempts(), 5);
+ });
+
+ it('should inherit customized backoff from protobuf', () => {
+ const customizedBackoff = new CustomizedBackoff();
+ customizedBackoff.addNext(new Duration().setSeconds(1));
+ customizedBackoff.addNext(new Duration().setNanos(500000000));
+ const retryPolicy = new RetryPolicyPB()
+ .setMaxAttempts(4)
+ .setCustomizedBackoff(customizedBackoff);
+ const policy = new CustomizedBackoffRetryPolicy([100],
3).inheritBackoff(retryPolicy);
+ assert.strictEqual(policy.getMaxAttempts(), 3);
+ assert.strictEqual(policy.getNextAttemptDelay(1), 1000);
+ assert.strictEqual(policy.getNextAttemptDelay(2), 500);
+ });
+});
+
+describe('Endpoints parsing', () => {
+ it('should parse IPv4 endpoints with port', () => {
+ const endpoints = new Endpoints('127.0.0.1:10911');
+ assert.deepStrictEqual(endpoints.addressesList, [{ host: '127.0.0.1',
port: 10911 }]);
+ assert.strictEqual(endpoints.scheme, 1); // IPV4
+ });
+
+ it('should parse domain endpoints', () => {
+ const endpoints = new Endpoints('example.com:443');
+ assert.deepStrictEqual(endpoints.addressesList, [{ host: 'example.com',
port: 443 }]);
+ assert.strictEqual(endpoints.scheme, 3); // DOMAIN_NAME
+ });
+
+ it('should parse bracketed IPv6 endpoints with port', () => {
+ const endpoints = new Endpoints('[::1]:10911');
+ assert.deepStrictEqual(endpoints.addressesList, [{ host: '::1', port:
10911 }]);
+ assert.strictEqual(endpoints.scheme, 2); // IPV6
+ });
+
+ it('should parse bare IPv6 endpoints without port', () => {
+ const endpoints = new Endpoints('1050:0000:0000:0000:0005:0600:300c:326b');
+ assert.strictEqual(endpoints.addressesList[0].host,
'1050:0000:0000:0000:0005:0600:300c:326b');
+ assert.strictEqual(endpoints.addressesList[0].port, 80);
+ assert.strictEqual(endpoints.scheme, 2); // IPV6
+ });
+
+ it('should strip http(s) prefixes and parse multiple endpoints', () => {
+ const endpoints = new
Endpoints('http://127.0.0.1:10911;https://example.com:8080');
+ assert.strictEqual(endpoints.addressesList.length, 2);
+ assert.deepStrictEqual(endpoints.addressesList[0], { host: '127.0.0.1',
port: 10911 });
+ assert.deepStrictEqual(endpoints.addressesList[1], { host: 'example.com',
port: 8080 });
+ });
+});
+
+describe('StatusChecker mappings vs Java', () => {
+ it('should map ILLEGAL_LITE_TOPIC to BadRequestException', () => {
+ assert.throws(() =>
StatusChecker.check(statusOf(Code.ILLEGAL_LITE_TOPIC)), BadRequestException);
+ });
+
+ it('should map MESSAGE_BODY_EMPTY to PayloadEmptyException', () => {
+ assert.throws(() =>
StatusChecker.check(statusOf(Code.MESSAGE_BODY_EMPTY)), PayloadEmptyException);
+ });
+
+ it('should map LITE_SUBSCRIPTION_QUOTA_EXCEEDED to
LiteSubscriptionQuotaExceededException', () => {
+ assert.throws(
+ () =>
StatusChecker.check(statusOf(Code.LITE_SUBSCRIPTION_QUOTA_EXCEEDED)),
+ LiteSubscriptionQuotaExceededException);
+ });
+});
