fix: Reconnect monitor cleanup bug and other comments

aws-amplify · Sep 14, 2022 · 1a605e9 · 1a605e9
1 parent 86f55ee
commit 1a605e9
Show file tree

Hide file tree

Showing 5 changed files with 31 additions and 19 deletions.
diff --git a/packages/pubsub/__tests__/AWSAppSyncRealTimeProvider.test.ts b/packages/pubsub/__tests__/AWSAppSyncRealTimeProvider.test.ts
@@ -510,7 +510,7 @@ describe('AWSAppSyncRealTimeProvider', () => {
 						payload: { data: {} },
 					});
 					expect(loggerSpy).toBeCalledWith(
-						'ERROR',
+						'DEBUG',
 						'Connection failed: {"data":{}}'
 					);
 					await fakeWebSocketInterface?.waitUntilConnectionStateIn([
@@ -981,7 +981,7 @@ describe('AWSAppSyncRealTimeProvider', () => {
 						await delay(10);
 
 						expect(loggerSpy).toBeCalledWith(
-							'ERROR',
+							'DEBUG',
 							'AppSync Realtime subscription init error: Error: No credentials'
 						);
 					});
@@ -1013,7 +1013,7 @@ describe('AWSAppSyncRealTimeProvider', () => {
 						await delay(10);
 
 						expect(loggerSpy).toBeCalledWith(
-							'ERROR',
+							'DEBUG',
 							'AppSync Realtime subscription init error: Error: No credentials'
 						);
 
@@ -1077,7 +1077,7 @@ describe('AWSAppSyncRealTimeProvider', () => {
 						await delay(10);
 
 						expect(loggerSpy).toBeCalledWith(
-							'ERROR',
+							'DEBUG',
 							'AppSync Realtime subscription init error: Error: No federated jwt'
 						);
 					});
@@ -1171,7 +1171,7 @@ describe('AWSAppSyncRealTimeProvider', () => {
 						await delay(10);
 
 						expect(loggerSpy).toBeCalledWith(
-							'ERROR',
+							'DEBUG',
 							'AppSync Realtime subscription init error: Error: No auth token specified'
 						);
 					});

diff --git a/packages/pubsub/__tests__/PubSub-unit-test.ts b/packages/pubsub/__tests__/PubSub-unit-test.ts
@@ -16,7 +16,7 @@ import {
 	AWSIoTProvider,
 	mqttTopicMatch,
 } from '../src/Providers';
-// import Amplify from '../../src/';
+
 import {
 	Credentials,
 	Hub,

diff --git a/packages/pubsub/src/Providers/AWSAppSyncRealTimeProvider/index.ts b/packages/pubsub/src/Providers/AWSAppSyncRealTimeProvider/index.ts
@@ -124,7 +124,7 @@ export class AWSAppSyncRealTimeProvider extends AbstractPubSubProvider {
 
 					// Trigger reconnection when the connection is disrupted
 					if (connectionState === ConnectionState.ConnectionDisrupted) {
-						this.reconnectionMonitor.record(ReconnectEvent.RECONNECT);
+						this.reconnectionMonitor.record(ReconnectEvent.START_RECONNECT);
 					} else if (connectionState !== ConnectionState.Connecting) {
 						// Trigger connected to halt reconnection attempts
 						this.reconnectionMonitor.record(ReconnectEvent.HALT_RECONNECT);
@@ -198,7 +198,7 @@ export class AWSAppSyncRealTimeProvider extends AbstractPubSubProvider {
 								observer,
 								subscriptionId,
 							}).catch<any>(err => {
-								logger.error(
+								logger.debug(
 									`${CONTROL_MSG.REALTIME_SUBSCRIPTION_INIT_ERROR}: ${err}`
 								);
 
@@ -346,7 +346,7 @@ export class AWSAppSyncRealTimeProvider extends AbstractPubSubProvider {
 				this.connectionState !==
 				ConnectionState.ConnectionDisruptedPendingNetwork
 			) {
-				logger.error(`${CONTROL_MSG.CONNECTION_FAILED}: ${message}`);
+				logger.debug(`${CONTROL_MSG.CONNECTION_FAILED}: ${message}`);
 				this.connectionStateMonitor.record(CONNECTION_CHANGE.CLOSED);
 
 				const { subscriptionFailedCallback } =
@@ -556,7 +556,7 @@ export class AWSAppSyncRealTimeProvider extends AbstractPubSubProvider {
 					subscriptionState,
 				});
 
-				logger.error(
+				logger.debug(
 					`${CONTROL_MSG.CONNECTION_FAILED}: ${JSON.stringify(payload)}`
 				);
 
@@ -664,7 +664,7 @@ export class AWSAppSyncRealTimeProvider extends AbstractPubSubProvider {
 					this.socketStatus = SOCKET_STATUS.READY;
 					this.promiseArray = [];
 				} catch (err) {
-					logger.error(err);
+					logger.debug('Notifying connection exited with', err);
 					this.promiseArray.forEach(({ rej }) => rej(err));
 					this.promiseArray = [];
 					if (

diff --git a/packages/pubsub/src/Providers/MqttOverWSProvider.ts b/packages/pubsub/src/Providers/MqttOverWSProvider.ts
@@ -58,7 +58,7 @@ export interface MqttProviderOptions extends ProviderOptions {
 export type MqttProvidertOptions = MqttProviderOptions;
 
 class ClientsQueue {
-	private promises: Map<string, Promise<any> | any> = new Map();
+	private promises: Map<string, Promise<any>> = new Map();
 
 	async get(clientId: string, clientFactory?: (input: string) => Promise<any>) {
 		const cachedPromise = this.promises.get(clientId);
@@ -67,9 +67,7 @@ class ClientsQueue {
 		if (clientFactory) {
 			const newPromise = clientFactory(clientId);
 			this.promises.set(clientId, newPromise);
-			newPromise.then(v => this.promises.set(clientId, v));
 			newPromise.catch(v => this.promises.delete(clientId));
-			newPromise.finally();
 			return newPromise;
 		}
 
@@ -116,7 +114,7 @@ export class MqttOverWSProvider extends AbstractPubSubProvider {
 
 				// Trigger reconnection when the connection is disrupted
 				if (connectionStateChange === ConnectionState.ConnectionDisrupted) {
-					this.reconnectionMonitor.record(ReconnectEvent.RECONNECT);
+					this.reconnectionMonitor.record(ReconnectEvent.START_RECONNECT);
 				} else if (connectionStateChange !== ConnectionState.Connecting) {
 					// Trigger connected to halt reconnection attempts
 					this.reconnectionMonitor.record(ReconnectEvent.HALT_RECONNECT);
@@ -338,7 +336,7 @@ export class MqttOverWSProvider extends AbstractPubSubProvider {
 							});
 						}
 					} catch (e) {
-						logger.error('Error forming connection', e);
+						logger.debug('Error forming connection', e);
 					}
 				};
 

diff --git a/packages/pubsub/src/utils/ReconnectionMonitor.ts b/packages/pubsub/src/utils/ReconnectionMonitor.ts
@@ -2,28 +2,35 @@ import { Observer } from 'zen-observable-ts';
 import { RECONNECT_DELAY, RECONNECT_INTERVAL } from '../Providers/constants';
 
 export enum ReconnectEvent {
-	RECONNECT = 'RECONNECT',
+	START_RECONNECT = 'START_RECONNECT',
 	HALT_RECONNECT = 'HALT_RECONNECT',
 }
 
 export class ReconnectionMonitor {
 	private reconnectObservers: Observer<void>[] = [];
 	private reconnectIntervalId?: ReturnType<typeof setInterval>;
+	private reconnectSetTimeoutId?: ReturnType<typeof setTimeout>;
 
+	/**
+	 * Add reconnect observer to the list of observers to alert on reconnect
+	 */
 	addObserver(reconnectObserver: Observer<void>) {
 		this.reconnectObservers.push(reconnectObserver);
 	}
 
+	/**
+	 * Given a reconnect event, start the appropriate behavior
+	 */
 	record(event: ReconnectEvent) {
-		if (event === ReconnectEvent.RECONNECT) {
+		if (event === ReconnectEvent.START_RECONNECT) {
 			const triggerReconnect = () => {
 				this.reconnectObservers.forEach(reconnectObserver => {
 					reconnectObserver.next?.();
 				});
 			};
 			// If the reconnect interval isn't set
 			if (this.reconnectIntervalId === undefined) {
-				setTimeout(() => {
+				this.reconnectSetTimeoutId = setTimeout(() => {
 					// Reconnect now
 					triggerReconnect();
 					// Retry reconnect every periodically until it works
@@ -39,9 +46,16 @@ export class ReconnectionMonitor {
 				clearInterval(this.reconnectIntervalId);
 				this.reconnectIntervalId = undefined;
 			}
+			if (this.reconnectSetTimeoutId) {
+				clearTimeout(this.reconnectSetTimeoutId);
+				this.reconnectSetTimeoutId = undefined;
+			}
 		}
 	}
 
+	/**
+	 * Complete all reconnect observers
+	 */
 	close() {
 		this.reconnectObservers.forEach(reconnectObserver => {
 			reconnectObserver.complete?.();