Skip to content

Commit 20740f9

Browse files
joseph118joseph118
authored
feat(hub connection): throttle reconnect on disconnect (#70)
1 parent e6df816 commit 20740f9

7 files changed

+181
-30
lines changed

‎CHANGELOG.md

+13-1
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,16 @@
1-
## [5.0.1](https://github.com/sketch7/signalr-client/compare/5.0.0...5.0.1) (2023-09-11)
1+
## [5.1.0](https://github.com/sketch7/signalr-client/compare/5.0.1...5.1.0) (2025-01-21)
2+
3+
### Features
4+
5+
- **hub connection:**
6+
- throttle reconnect on disconnect and stop trying after max attempts
7+
- add `autoReconnectRecoverIntervalMS` to reset max attempts after a duration
8+
9+
### Refactor
10+
11+
- **hub connection:** convert `on` + `stream` `retryWhen` (deprecated) to `retry`
12+
13+
## [5.0.1](https://github.com/sketch7/signalr-client/compare/5.0.0...5.0.1) (2024-09-11)
214

315
### Refactor
416

‎package-lock.json

+2-2
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎package.json

+1-1
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
{
22
"name": "@ssv/signalr-client",
3-
"version": "5.0.1",
3+
"version": "5.1.0",
44
"versionSuffix": "",
55
"description": "SignalR client library built on top of @microsoft/signalr. This gives you more features and easier to use.",
66
"homepage": "https://github.com/sketch7/signalr-client",

‎src/lib/hub-connection.connection.spec.ts

+66-6
Original file line numberDiff line numberDiff line change
@@ -1,18 +1,30 @@
11
import { HubConnection } from "./hub-connection";
2-
import { Subscription, lastValueFrom, merge, first, switchMap, tap, skip, delay, withLatestFrom } from "rxjs";
2+
import {
3+
Subscription, lastValueFrom, merge, first, switchMap, tap, skip, delay, withLatestFrom, takeWhile, filter, finalize, Observable
4+
} from "rxjs";
35
import type { Mock, MockInstance } from "vitest";
46

5-
import { HeroHub, createSUT } from "./testing/hub-connection.util";
6-
import { ConnectionStatus } from "./hub-connection.model";
7+
import { AUTO_RECONNECT_RECOVER_INTERVAL, HeroHub, RETRY_MAXIMUM_ATTEMPTS, createSUT } from "./testing/hub-connection.util";
8+
import { ConnectionState, ConnectionStatus } from "./hub-connection.model";
79
import { MockSignalRHubConnectionBuilder, MockSignalRHubBackend } from "./testing";
810

911
import * as signalr from "@microsoft/signalr";
12+
1013
function promiseDelayResolve(ms: number) {
1114
return new Promise(r => setTimeout(r, ms));
1215
}
1316
function promiseDelayReject(ms: number, reason?: unknown) {
1417
return new Promise((_, reject) => setTimeout(() => reject(reason), ms));
1518
}
19+
function exhaustHubRetryAttempts$(sut: HubConnection<HeroHub>, hubBackend: MockSignalRHubBackend): Observable<ConnectionState> {
20+
let retryCount = 0;
21+
return sut.connectionState$.pipe(
22+
filter(state => state.status === ConnectionStatus.connected),
23+
takeWhile(() => retryCount < RETRY_MAXIMUM_ATTEMPTS),
24+
tap(() => retryCount++),
25+
tap(() => hubBackend.disconnect(new Error("Disconnected by the server to exhaust max attempts"))),
26+
);
27+
}
1628

1729
describe("HubConnection Specs", () => {
1830

@@ -70,7 +82,7 @@ describe("HubConnection Specs", () => {
7082
hubBackend.connection.stop = vi.fn().mockReturnValue(promiseDelayResolve(5));
7183
});
7284

73-
it("should connect once", async () => {
85+
it("should connect once", async () => {
7486
const c1$ = lastValueFrom(SUT.connect());
7587
const c2$ = lastValueFrom(SUT.connect());
7688

@@ -253,7 +265,7 @@ describe("HubConnection Specs", () => {
253265

254266
});
255267

256-
describe("when disconnects", () => {
268+
describe("when disconnects from server", () => {
257269

258270
beforeEach(() => {
259271
// todo: check if redundant
@@ -265,7 +277,7 @@ describe("HubConnection Specs", () => {
265277
const reconnect$ = SUT.connectionState$.pipe(
266278
first(),
267279
tap(state => expect(state.status).toBe(ConnectionStatus.connected)),
268-
tap(() => hubBackend.disconnect(new Error("Disconnected by the server"))),
280+
tap(() => hubBackend.disconnect(new Error("Disconnected by the server for auto reconnect"))),
269281
switchMap(() => SUT.connectionState$.pipe(first(x => x.status === ConnectionStatus.connected))),
270282
tap(state => {
271283
expect(state.status).toBe(ConnectionStatus.connected);
@@ -277,6 +289,54 @@ describe("HubConnection Specs", () => {
277289
return lastValueFrom(reconnect$);
278290
});
279291

292+
describe("and server is encountering issues", () => {
293+
294+
it("should stop reconnecting after maximum attempts", () => {
295+
const reconnect$ = exhaustHubRetryAttempts$(SUT, hubBackend).pipe(
296+
finalize(() => {
297+
expect(SUT.connectionState.status).toBe(ConnectionStatus.disconnected);
298+
expect(hubStartSpy).toBeCalledTimes(3);
299+
expect(hubStopSpy).not.toBeCalled();
300+
})
301+
);
302+
303+
return lastValueFrom(reconnect$);
304+
});
305+
306+
it("should reset maximum attempts after trigger disconnect + connect", () => {
307+
const resetMaxAttempts$ = exhaustHubRetryAttempts$(SUT, hubBackend).pipe(
308+
tap(() => SUT.disconnect()),
309+
switchMap(() => SUT.connect()),
310+
switchMap(() => SUT.connectionState$.pipe(
311+
first(state => state.status === ConnectionStatus.connected),
312+
tap(() => hubBackend.disconnect(new Error("Disconnected by the server to re-trigger auto reconnect"))),
313+
switchMap(() => SUT.connectionState$.pipe(first(x => x.status === ConnectionStatus.connected))),
314+
)),
315+
tap(state => {
316+
expect(hubStartSpy).toBeCalledTimes(4);
317+
expect(state.status).toBe(ConnectionStatus.connected);
318+
}),
319+
);
320+
321+
return lastValueFrom(resetMaxAttempts$);
322+
});
323+
324+
it("should reset maximum attempts after exceeding the recover duration", () => {
325+
const resetMaxAttempts$ = exhaustHubRetryAttempts$(SUT, hubBackend).pipe(
326+
delay(AUTO_RECONNECT_RECOVER_INTERVAL),
327+
switchMap(() => SUT.connectionState$.pipe(
328+
first(state => state.status === ConnectionStatus.connected),
329+
)),
330+
tap(state => {
331+
expect(hubStartSpy).toBeCalledTimes(3);
332+
expect(state.status).toBe(ConnectionStatus.connected);
333+
}),
334+
);
335+
336+
return lastValueFrom(resetMaxAttempts$);
337+
});
338+
});
339+
280340
});
281341

282342
});

‎src/lib/hub-connection.model.ts

+6
Original file line numberDiff line numberDiff line change
@@ -51,6 +51,12 @@ export interface ConnectionOptions extends IHttpConnectionOptions {
5151

5252
export interface ReconnectionStrategyOptions {
5353
maximumAttempts?: number;
54+
/**
55+
* Resets maximum attempts when exhausted after the given duration.
56+
* The duration is restarted for each connection attempt unless a Date is provided.
57+
* Supports number in MS or date. Defaults to 15 minutes.
58+
*/
59+
autoReconnectRecoverInterval?: number | Date;
5460
customStrategy?: (retryOptions: ReconnectionStrategyOptions, retryCount: number) => number;
5561
randomBackOffStrategy?: RandomStrategyOptions;
5662
randomStrategy?: RandomStrategyOptions;

‎src/lib/hub-connection.ts

+78-12
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,11 @@
1-
import { from, BehaviorSubject, Observable, Observer, timer, throwError, Subject } from "rxjs";
1+
import { from, BehaviorSubject, Observable, Observer, timer, throwError, Subject, EMPTY, merge, } from "rxjs";
22
import {
33
tap, map, filter, switchMap, skipUntil, delay, first,
4-
retryWhen, delayWhen, distinctUntilChanged, takeUntil,retry
4+
delayWhen, distinctUntilChanged, takeUntil, retry,
5+
scan,
6+
catchError,
7+
skip,
8+
take,
59
} from "rxjs/operators";
610
import {
711
HubConnection as SignalRHubConnection,
@@ -25,6 +29,8 @@ const connectingState = Object.freeze<ConnectionState>({ status: ConnectionStatu
2529
// todo: rename HubClient?
2630
export class HubConnection<THub> {
2731

32+
get connectionState(): ConnectionState { return this._connectionState$.value; }
33+
2834
/** Gets the connection state. */
2935
get connectionState$(): Observable<ConnectionState> { return this._connectionState$.asObservable(); }
3036

@@ -108,11 +114,7 @@ export class HubConnection<THub> {
108114
takeUntil(this._destroy$),
109115
);
110116

111-
const reconnectOnDisconnect$ = this._connectionState$.pipe(
112-
// tap(x => console.warn(">>>> _connectionState$ state changed", x)),
113-
filter(x => x.status === ConnectionStatus.disconnected && x.reason === errorReasonName),
114-
// tap(x => console.warn(">>>> reconnecting...", x)),
115-
switchMap(() => this.connect()),
117+
const reconnectOnDisconnect$ = this.reconnectOnDisconnect$().pipe(
116118
takeUntil(this._destroy$),
117119
);
118120

@@ -226,7 +228,7 @@ export class HubConnection<THub> {
226228
);
227229
}
228230

229-
private untilDesiredDisconnects$() {
231+
private untilDesiredDisconnects$(): Observable<void> {
230232
return this.desiredState$.pipe(
231233
first(state => state === DesiredConnectionStatus.disconnected),
232234
map(() => undefined),
@@ -268,10 +270,11 @@ export class HubConnection<THub> {
268270
private activateStreamWithRetry<TResult>(stream$: Observable<TResult>): Observable<TResult> {
269271
return this.waitUntilConnect$.pipe(
270272
switchMap(() => stream$.pipe(
271-
retryWhen((errors: Observable<unknown>) => errors.pipe(
272-
delay(1), // workaround - when connection disconnects, stream errors fires before `signalr.onClose`
273-
delayWhen(() => this.waitUntilConnect$)
274-
))
273+
retry({
274+
delay: () => timer(1).pipe( // workaround - when connection disconnects, stream errors fires before `signalr.onClose`
275+
delayWhen(() => this.waitUntilConnect$)
276+
)
277+
})
275278
))
276279
);
277280
}
@@ -288,4 +291,67 @@ export class HubConnection<THub> {
288291
return data;
289292
}
290293

294+
private reconnectOnDisconnect$(): Observable<void> {
295+
const onServerErrorDisconnect$ = this._connectionState$.pipe(
296+
filter(x => x.status === ConnectionStatus.disconnected && x.reason === errorReasonName),
297+
);
298+
299+
// this is a fallback for when max attempts are reached and will emit to reset the max attempt after a duration
300+
const maxAttemptReset$ = onServerErrorDisconnect$.pipe(
301+
switchMap(() => this._connectionState$.pipe(
302+
switchMap(() => timer(this.retry.autoReconnectRecoverInterval || 900000)), // 15 minutes default
303+
take(1),
304+
takeUntil(
305+
this.connectionState$.pipe(
306+
filter(x => x.status === ConnectionStatus.connected)
307+
)
308+
),
309+
)),
310+
// tap(() => console.error(`${this.source} [reconnectOnDisconnect$] :: resetting max attempts`)),
311+
);
312+
313+
const onDisconnect$ = this.desiredState$.pipe(
314+
filter(state => state === DesiredConnectionStatus.disconnected),
315+
);
316+
317+
return merge(
318+
onDisconnect$,
319+
maxAttemptReset$,
320+
).pipe(
321+
switchMap(() => onServerErrorDisconnect$.pipe(
322+
scan(attempts => attempts += 1, 0),
323+
map(retryCount => ({
324+
retryCount,
325+
nextRetryMs: retryCount ? getReconnectionDelay(this.retry, retryCount) : 0
326+
})),
327+
switchMap(({ retryCount, nextRetryMs }) => {
328+
if (this.retry.maximumAttempts && retryCount > this.retry.maximumAttempts) {
329+
return throwError(() => new Error(errorCodes.retryLimitsReached));
330+
}
331+
332+
const delay$ = !nextRetryMs
333+
? emptyNext()
334+
: timer(nextRetryMs).pipe(
335+
map(() => undefined)
336+
);
337+
return delay$.pipe(
338+
// tap(() => console.error(`${this.source} [reconnectOnDisconnect$] :: retrying`, {
339+
// retryCount,
340+
// nextRetryMs,
341+
// maximumAttempts: this.retry.maximumAttempts,
342+
// })),
343+
switchMap(() => this.connect()),
344+
// tap(() => console.error(">>>> [reconnectOnDisconnect$] connected")),
345+
takeUntil(this.untilDesiredDisconnects$()),
346+
);
347+
}),
348+
catchError(() => EMPTY),
349+
takeUntil(this.desiredState$.pipe(
350+
skip(1),
351+
filter(state => state === DesiredConnectionStatus.disconnected),
352+
)),
353+
)),
354+
);
355+
}
356+
291357
}

‎src/lib/testing/hub-connection.util.ts

+15-8
Original file line numberDiff line numberDiff line change
@@ -1,27 +1,34 @@
11
import { HubConnection } from "../hub-connection";
22
import { vi } from "vitest";
3+
import { ReconnectionStrategyOptions } from "../hub-connection.model";
34
// jest.genMockFromModule("@microsoft/signalr");
45
vi.mock("@microsoft/signalr");
56

67
let nextUniqueId = 0;
78

9+
export const RETRY_MAXIMUM_ATTEMPTS = 3;
10+
export const AUTO_RECONNECT_RECOVER_INTERVAL = 2000;
11+
812
export interface HeroHub {
913
UpdateHero: string;
1014
}
1115

12-
export function createSUT(): HubConnection<HeroHub> {
16+
export function createSUT(retryOptions: ReconnectionStrategyOptions = {}): HubConnection<HeroHub> {
17+
const retry = {
18+
backOffStrategy: {
19+
delayRetriesMs: 10,
20+
maxDelayRetriesMs: 10
21+
},
22+
maximumAttempts: RETRY_MAXIMUM_ATTEMPTS,
23+
autoReconnectRecoverInterval: AUTO_RECONNECT_RECOVER_INTERVAL,
24+
...retryOptions,
25+
};
1326
return new HubConnection<HeroHub>({
1427
key: `hero-${nextUniqueId++}`,
1528
endpointUri: "/hero",
1629
defaultData: () => ({ tenant: "kowalski", power: "2000" }),
1730
options: {
18-
retry: {
19-
maximumAttempts: 3,
20-
backOffStrategy: {
21-
delayRetriesMs: 10,
22-
maxDelayRetriesMs: 10
23-
}
24-
},
31+
retry,
2532
}
2633
});
2734
}

0 commit comments

Comments
 (0)