Skip to content

Commit d3b297b

Browse files
stephenlautierstephenlautier
authored
fix(hub connection): correctly close the websocket connection when disconnect is called right after connect (#54)
### Features - **hub connection:** ability to configure microsoft `HubConnection` directly via `configureSignalRHubConnection` ### Bug Fixes - **hub connection:** correctly close the websocket connection when `disconnect` is called right after `connect` - **hub connection:** `connectionState$` `connecting` will also be set when connecting not only when reconnecting
1 parent 4e94c71 commit d3b297b

File tree

5 files changed

+81
-36
lines changed

5 files changed

+81
-36
lines changed

‎CHANGELOG.md‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,14 @@
1+
## [3.1.0](https://github.com/sketch7/signalr-client/compare/3.0.0...3.1.0) (2022-11-22)
2+
3+
### Features
4+
5+
- **hub connection:** ability to configure microsoft `HubConnection` directly via `configureSignalRHubConnection`
6+
7+
### Bug Fixes
8+
9+
- **hub connection:** correctly close the websocket connection when `disconnect` is called right after `connect`
10+
- **hub connection:** `connectionState$` `connecting` will also be set when connecting not only when reconnecting
11+
112
## [3.0.0](https://github.com/sketch7/signalr-client/compare/2.0.0...3.0.0) (2021-04-30)
213

314
### Features

‎package.json‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
{
22
"name": "@ssv/signalr-client",
3-
"version": "3.0.0",
3+
"version": "3.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/hub-connection.connection.spec.ts‎

Lines changed: 29 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -65,13 +65,14 @@ describe("HubConnection Specs", () => {
6565

6666
beforeEach(() => {
6767
hubBackend.connection.start = jest.fn().mockReturnValue(promiseDelayResolve(5));
68+
hubBackend.connection.stop = jest.fn().mockReturnValue(promiseDelayResolve(5));
6869
});
6970

7071

7172
describe("and connects successfully", () => {
7273

7374

74-
75+
// connect -> WHILE CONNECTING -> disconnect
7576
it("should have status disconnected", done => {
7677
const connect$ = SUT.connect();
7778
const state$ = SUT.connectionState$.pipe(
@@ -81,13 +82,39 @@ describe("HubConnection Specs", () => {
8182
withLatestFrom(SUT.connectionState$, (_x, y) => y),
8283
tap(state => {
8384
expect(hubBackend.connection.start).toHaveBeenCalledTimes(1);
85+
expect(hubBackend.connection.stop).toHaveBeenCalledTimes(1);
8486
expect(state.status).toBe(ConnectionStatus.disconnected);
8587
done();
8688
})
8789
);
8890
conn$$ = merge(connect$, state$).subscribe();
8991
});
9092

93+
describe("and connects with different data", () => {
94+
95+
96+
// connect -> WHILE CONNECTING -> disconnect -> connect with different data
97+
it("should have status connected", done => {
98+
const connect$ = SUT.connect();
99+
const state$ = SUT.connectionState$.pipe(
100+
first(),
101+
switchMap(() => SUT.disconnect()),
102+
delay(2), // ensure start is in flight
103+
switchMap(() => SUT.connect(() => ({ second: "true" }))),
104+
withLatestFrom(SUT.connectionState$, (_x, y) => y),
105+
tap(state => {
106+
expect(hubBackend.connection.start).toHaveBeenCalledTimes(2);
107+
expect(hubBackend.connection.stop).toHaveBeenCalledTimes(1);
108+
expect(state.status).toBe(ConnectionStatus.connected);
109+
done();
110+
})
111+
);
112+
conn$$ = merge(connect$, state$).subscribe();
113+
});
114+
115+
116+
117+
});
91118

92119

93120
});
@@ -184,7 +211,7 @@ describe("HubConnection Specs", () => {
184211
tap(state => {
185212
expect(state.status).toBe(ConnectionStatus.disconnected);
186213
expect(hubStartSpy).toBeCalledTimes(1);
187-
expect(hubStopSpy).not.toBeCalled();
214+
// expect(hubStopSpy).not.toBeCalled();
188215
done();
189216
}),
190217
);

‎src/hub-connection.model.ts‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
import { IHttpConnectionOptions, IHubProtocol } from "@microsoft/signalr";
1+
import { HubConnection, IHttpConnectionOptions, IHubProtocol } from "@microsoft/signalr";
22

33
import { Dictionary } from "./utils/dictionary";
44

@@ -38,6 +38,10 @@ export interface HubConnectionOptions {
3838
/** @internal */
3939
getData?: () => Dictionary<string>;
4040
protocol?: IHubProtocol;
41+
/**
42+
* Configures the SignalR Hub connection after it has been built (raw) in order to access/configure `serverTimeoutInMilliseconds`, `keepAliveIntervalInMilliseconds` etc...
43+
*/
44+
configureSignalRHubConnection?: (hubConnection: HubConnection) => void;
4145
}
4246

4347
export interface ConnectionOptions extends IHttpConnectionOptions {

‎src/hub-connection.ts‎

Lines changed: 35 additions & 32 deletions
Original file line numberDiff line numberDiff line change
@@ -4,9 +4,9 @@ import {
44
} from "rxjs/operators";
55
import {
66
HubConnection as SignalRHubConnection,
7-
HubConnectionBuilder as SignalRHubConnectionBuilder
7+
HubConnectionBuilder as SignalRHubConnectionBuilder,
88
} from "@microsoft/signalr";
9-
import { from as fromPromise, BehaviorSubject, Observable, Observer, timer, throwError, Subscription, merge } from "rxjs";
9+
import { from, BehaviorSubject, Observable, Observer, timer, throwError, Subject } from "rxjs";
1010

1111
import {
1212
ConnectionState, ConnectionStatus, HubConnectionOptions,
@@ -20,6 +20,7 @@ import { emptyNext } from "./utils/rxjs";
2020
const errorReasonName = "error";
2121
const disconnectedState = Object.freeze<ConnectionState>({ status: ConnectionStatus.disconnected });
2222
const connectedState = Object.freeze<ConnectionState>({ status: ConnectionStatus.connected });
23+
const connectingState = Object.freeze<ConnectionState>({ status: ConnectionStatus.connecting });
2324

2425
// todo: rename HubClient?
2526
export class HubConnection<THub> {
@@ -36,7 +37,7 @@ export class HubConnection<THub> {
3637
private desiredState$ = new BehaviorSubject<DesiredConnectionStatus>(DesiredConnectionStatus.disconnected);
3738
private internalConnStatus$ = new BehaviorSubject<InternalConnectionStatus>(InternalConnectionStatus.disconnected);
3839
private connectionBuilder = new SignalRHubConnectionBuilder();
39-
private effects$$ = Subscription.EMPTY;
40+
private readonly _destroy$ = new Subject<void>();
4041

4142
private waitUntilConnect$ = this.connectionState$.pipe(
4243
distinctUntilChanged((x, y) => x.status === y.status),
@@ -62,7 +63,9 @@ export class HubConnection<THub> {
6263
if (connectionOpts.protocol) {
6364
this.connectionBuilder = this.connectionBuilder.withHubProtocol(connectionOpts.protocol);
6465
}
66+
6567
this.hubConnection = this.connectionBuilder.build();
68+
connectionOpts.configureSignalRHubConnection?.(this.hubConnection);
6669
this.hubConnection.onclose(err => {
6770
this.internalConnStatus$.next(InternalConnectionStatus.disconnected);
6871
if (err) {
@@ -81,42 +84,40 @@ export class HubConnection<THub> {
8184
tap(() => this.internalConnStatus$.next(InternalConnectionStatus.ready)),
8285
filter(() => prevConnectionStatus === InternalConnectionStatus.connected),
8386
switchMap(() => this.openConnection())
84-
))
87+
)),
88+
takeUntil(this._destroy$),
8589
);
8690
const desiredDisconnected$ = this.desiredState$.pipe(
8791
filter(status => status === DesiredConnectionStatus.disconnected),
88-
// tap(status => console.warn(">>>> disconnected$", { internalConnStatus$: this.internalConnStatus$.value, desiredStatus: status })),
92+
// tap(status => console.warn(">>>> [desiredDisconnected$] desired disconnected", { internalConnStatus$: this.internalConnStatus$.value, desiredStatus: status })),
8993
tap(() => {
90-
switch (this.internalConnStatus$.value) {
91-
case InternalConnectionStatus.connected:
92-
this._disconnect();
93-
break;
94-
case InternalConnectionStatus.ready:
95-
this._connectionState$.next(disconnectedState);
96-
break;
97-
// default:
98-
// console.error("desiredDisconnected$ - State unhandled", this.internalConnStatus$.value);
99-
// break;
94+
if (this._connectionState$.value.status !== ConnectionStatus.disconnected) {
95+
// console.warn(">>>> [desiredDisconnected$] _disconnect");
96+
// note: ideally delayWhen disconnect first, though for some reason obs not bubbling
97+
this._disconnect()
10098
}
101-
})
99+
}),
100+
tap(() => this._connectionState$.next(disconnectedState)),
101+
takeUntil(this._destroy$),
102102
);
103103

104-
const reconnectOnDisconnect = this._connectionState$.pipe(
104+
const reconnectOnDisconnect$ = this._connectionState$.pipe(
105105
// tap(x => console.warn(">>>> _connectionState$ state changed", x)),
106106
filter(x => x.status === ConnectionStatus.disconnected && x.reason === errorReasonName),
107107
// tap(x => console.warn(">>>> reconnecting...", x)),
108-
switchMap(() => this.connect())
108+
switchMap(() => this.connect()),
109+
takeUntil(this._destroy$),
109110
);
110111

111-
this.effects$$ = merge(
112+
[
112113
desiredDisconnected$,
113-
reconnectOnDisconnect,
114+
reconnectOnDisconnect$,
114115
connection$
115-
).subscribe();
116+
].forEach((x: Observable<unknown>) => x.subscribe());
116117
}
117118

118119
connect(data?: () => Dictionary<string>): Observable<void> {
119-
// console.info("triggered connect", data);
120+
// console.warn("[connect] init", data);
120121
this.desiredState$.next(DesiredConnectionStatus.connected);
121122
if (this.internalConnStatus$.value === InternalConnectionStatus.connected) {
122123
console.warn(`${this.source} session already connected`);
@@ -142,7 +143,7 @@ export class HubConnection<THub> {
142143
}
143144

144145
disconnect(): Observable<void> {
145-
// console.info("triggered disconnect");
146+
// console.warn("[disconnect] init");
146147
this.desiredState$.next(DesiredConnectionStatus.disconnected);
147148
return this.untilDisconnects$();
148149
}
@@ -188,25 +189,26 @@ export class HubConnection<THub> {
188189
}
189190

190191
send(methodName: keyof THub | "StreamUnsubscribe", ...args: unknown[]): Observable<void> {
191-
return fromPromise(this.hubConnection.send(methodName.toString(), ...args));
192+
return from(this.hubConnection.send(methodName.toString(), ...args));
192193
}
193194

194195
invoke<TResult>(methodName: keyof THub, ...args: unknown[]): Observable<TResult> {
195-
return fromPromise<Promise<TResult>>(this.hubConnection.invoke(methodName.toString(), ...args));
196+
return from<Promise<TResult>>(this.hubConnection.invoke(methodName.toString(), ...args));
196197
}
197198

198199
dispose(): void {
199200
this.disconnect();
200201
this.desiredState$.complete();
201202
this._connectionState$.complete();
202203
this.internalConnStatus$.complete();
203-
this.effects$$.unsubscribe();
204+
this._destroy$.next();
205+
this._destroy$.complete();
204206
}
205207

206208
private _disconnect(): Observable<void> {
207-
// console.info("triggered _disconnect", this.internalConnStatus$.value);
208-
return this.internalConnStatus$.value === InternalConnectionStatus.connected
209-
? fromPromise(this.hubConnection.stop())
209+
// console.warn("[_disconnect] init", this.internalConnStatus$.value, this._connectionState$.value);
210+
return this._connectionState$.value.status !== ConnectionStatus.disconnected
211+
? from(this.hubConnection.stop())
210212
: emptyNext();
211213
}
212214

@@ -225,11 +227,12 @@ export class HubConnection<THub> {
225227
}
226228

227229
private openConnection() {
228-
// console.info("triggered openConnection");
230+
// console.warn("[openConnection]");
229231
return emptyNext().pipe(
230232
// tap(x => console.warn(">>>> openConnection - attempting to connect", x)),
231-
switchMap(() => fromPromise(this.hubConnection.start())),
232-
// tap(x => console.warn(">>>> openConnection - connection established", x)),
233+
tap(() => this._connectionState$.next(connectingState)),
234+
switchMap(() => from(this.hubConnection.start())),
235+
// tap(x => console.warn(">>>> [openConnection] - connection established", x)),
233236
retryWhen(errors => errors.pipe(
234237
scan((errorCount: number) => ++errorCount, 0),
235238
delayWhen((retryCount: number) => {

0 commit comments

Comments
 (0)