fix(hub connection): concurrent connect() remain in invalid state c…

…ausing connecting/disconnecting (#68)

CHANGELOG.md

@@ -5,6 +5,10 @@
 - **hub connection:** expose `HubConnection.connectionId`
 - **deps:** export `DesiredConnectionStatus`
 
+### Bug Fixes
+
+- **hub connection:** concurrent `connect()` remain in invalid state causing connecting/disconnecting
+
 ### BREAKING CHANGES
 
 - **deps:** changed `@microsoft/signalr: ^7.0.0` as `peerDependency`

package.json

@@ -25,7 +25,7 @@
     "build": "nx build",
     "lint": "nx lint",
     "test": "nx test",
-    "test-ui": "vitest --ui",
+    "tdd": "vitest --ui",
     "__CI__": "",
     "_pre-release-build": "echo hook for pre-release-build",
     "release": "npm publish ./dist --access=public"

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

@@ -50,15 +50,40 @@ describe("HubConnection Specs", () => {
 
 				describe("and connected successfully", () => {
 
-					it("should have status as connected", () => new Promise<void>(done => {
-						conn$$ = SUT.connect().pipe(
-							withLatestFrom(SUT.connectionState$, (_x, y) => y),
-						).subscribe({
-							next: state => expect(state.status).toBe(ConnectionStatus.connected),
-							complete: done
-						});
-						// todo: convert to promise
-					}));
+					it("should have status as connected", async () => {
+						await lastValueFrom(SUT.connect());
+
+						const state = await lastValueFrom(SUT.connectionState$.pipe(
+							delay(20),
+							first(),
+						));
+
+						expect(state.status).toBe(ConnectionStatus.connected);
+					});
+
+				});
+
+				describe("and invoked again concurrently", () => {
+
+					beforeEach(() => {
+						hubBackend.connection.start = vi.fn().mockReturnValue(promiseDelayResolve(5));
+						hubBackend.connection.stop = vi.fn().mockReturnValue(promiseDelayResolve(5));
+					});
+
+					it("should connect once", async () =>  {
+						const c1$ = lastValueFrom(SUT.connect());
+						const c2$ = lastValueFrom(SUT.connect());
+
+						await Promise.all([c1$, c2$]);
+
+						const state = await lastValueFrom(SUT.connectionState$.pipe(
+							delay(20),
+							first(),
+						));
+
+						expect(state.status).toBe(ConnectionStatus.connected);
+						expect(hubBackend.connection.start).toHaveBeenCalledTimes(1);
+					});
 
 				});
 
@@ -217,7 +242,7 @@ describe("HubConnection Specs", () => {
 
 			describe("when disconnect is invoked", () => {
 
-				it("should have status as disconnected", ()  => {
+				it("should have status as disconnected", () => {
 					const test$ = SUT.disconnect().pipe(
 						tap(() => hubBackend.disconnect()),
 						withLatestFrom(SUT.connectionState$, (_x, y) => y),

src/lib/hub-connection.ts

@@ -126,8 +126,8 @@ export class HubConnection<THub> {
 	connect(data?: () => Dictionary<string>): Observable<void> {
 		// console.warn("[connect] init", data);
 		this.desiredState$.next(DesiredConnectionStatus.connected);
-		if (this.internalConnStatus$.value === InternalConnectionStatus.connected) {
-			console.warn(`${this.source} session already connected`);
+		if (this._connectionState$.value.status !== ConnectionStatus.disconnected) {
+			console.warn(`${this.source} session already connecting/connected`);
 			return emptyNext();
 		}
 		if (data) {