AWSAppSyncProvider.ts

Filename	Percent	Threshold	Total	Covered	Uncovered
src/Providers/AWSAppSyncProvider.ts	86.13%	87.63%	310	267	43

// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
import { Client } from 'paho-mqtt';
import Observable from 'zen-observable-ts';
import { ConsoleLogger as Logger } from '@aws-amplify/core';

import { MqttOverWSProvider } from './MqttOverWSProvider';

const logger = new Logger('AWSAppSyncProvider');

/**
 * @deprecated Unused, all usecases have migrated to AWSAppSyncRealtimeProvider
 */
export class AWSAppSyncProvider extends MqttOverWSProvider {
	protected get endpoint() {
		throw new Error('Not supported');
	}

	getProviderName() {
		return 'AWSAppSyncProvider';
	}

	public async publish(topics: string[] | string, msg: any, options?: any) {
		throw new Error('Operation not supported');
	}

	private _cleanUp(clientId: string) {
		const topicsForClient = Array.from(this._topicClient.entries())
			.filter(([, c]) => c.clientId === clientId)
			.map(([t]) => t);

		topicsForClient.forEach(t => this._cleanUpForTopic(t));
	}

	private _cleanUpForTopic(topic) {
		this._topicClient.delete(topic);
		this._topicAlias.delete(topic);
	}

	public onDisconnect({ clientId, errorCode, ...args }) {
		if (errorCode !== 0) {
			const topicsForClient = Array.from(this._topicClient.entries())
				.filter(([, c]) => c.clientId === clientId)
				.map(([t]) => t);

			topicsForClient.forEach(topic => {
				if (this._topicObservers.has(topic)) {
					this._topicObservers.get(topic).forEach(obs => {
						if (!obs.closed) {
							obs.error(args);
						}
					});

					this._topicObservers.delete(topic);
				}
			});

			this._cleanUp(clientId);
		}
	}

	private _topicClient: Map<string, Client> = new Map();

	private _topicAlias: Map<string, string> = new Map();

	protected async disconnect(clientId: string): Promise<void> {
		const client = await this.clientsQueue.get(clientId, () => null);

		await super.disconnect(clientId);

		this._cleanUp(clientId);
	}

	subscribe(topics: string[] | string, options: any = {}): Observable<any> {
		const result = new Observable<any>(observer => {
			const targetTopics = ([] as string[]).concat(topics);
			logger.debug('Subscribing to topic(s)', targetTopics.join(','));

			(async () => {
				// Add these topics to map
				targetTopics.forEach(t => {
					if (!this._topicObservers.has(t)) {
						this._topicObservers.set(t, new Set());
					}

					this._topicObservers.get(t).add(observer);
				});

				const { mqttConnections = [], newSubscriptions } = options;

				// creates a map of {"topic": "alias"}
				const newAliases = Object.entries(newSubscriptions).map(
					([alias, v]: [string, { topic: string }]) => [v.topic, alias]
				);

				// Merge new aliases with old ones
				this._topicAlias = new Map([
					...Array.from(this._topicAlias.entries()),
					...(newAliases as [string, string][]),
				]);

				// group by urls
				const map: [string, { url: string; topics: Set<string> }][] =
					Object.entries(
						targetTopics.reduce((acc, elem) => {
							const connectionInfoForTopic = mqttConnections.find(
								c => c.topics.indexOf(elem) > -1
							);

							if (connectionInfoForTopic) {
								const { client: clientId, url } = connectionInfoForTopic;

								if (!acc[clientId]) {
									acc[clientId] = {
										url,
										topics: new Set(),
									};
								}

								acc[clientId].topics.add(elem);
							}

							return acc;
						}, {})
					);

				// reconnect everything we have in the map
				await Promise.all(
					map.map(async ([clientId, { url, topics }]) => {
						// connect to new client
						let client = null;
						try {
							client = await this.connect(clientId, {
								clientId,
								url,
							});
						} catch (err) {
							observer.error({ message: 'Failed to connect', error: err });
							observer.complete();
							return undefined;
						}

						// subscribe to all topics for this client
						// store topic-client mapping
						topics.forEach(topic => {
							if (client.isConnected()) {
								client.subscribe(topic);

								this._topicClient.set(topic, client);
							}
						});

						return client;
					})
				);
			})();

			return () => {
				logger.debug('Unsubscribing from topic(s)', targetTopics.join(','));

				targetTopics.forEach(t => {
					const client = this._topicClient.get(t);

					if (client && client.isConnected()) {
						client.unsubscribe(t);
						this._topicClient.delete(t);

						if (
							!Array.from(this._topicClient.values()).some(c => c === client)
						) {
							this.disconnect(client.clientId);
						}
					}

					this._topicObservers.delete(t);
				});
			};
		});

		return Observable.from(result).map(value => {
			const topic = this.getTopicForValue(value);
			const alias = this._topicAlias.get(topic);

			value.data = Object.entries(value.data).reduce(
				(obj, [origKey, val]) => (
					(obj[(alias || origKey) as string] = val), obj
				),
				{}
			);

			return value;
		});
	}
}

[{"file":"src/Providers/AWSAppSyncProvider.ts","line":22,"character":49,"text":"msg","kind":1},{"file":"src/Providers/AWSAppSyncProvider.ts","line":22,"character":59,"text":"options","kind":1},{"file":"src/Providers/AWSAppSyncProvider.ts","line":34,"character":26,"text":"topic","kind":1},{"file":"src/Providers/AWSAppSyncProvider.ts","line":39,"character":23,"text":"clientId","kind":1},{"file":"src/Providers/AWSAppSyncProvider.ts","line":39,"character":33,"text":"errorCode","kind":1},{"file":"src/Providers/AWSAppSyncProvider.ts","line":40,"character":6,"text":"errorCode","kind":1},{"file":"src/Providers/AWSAppSyncProvider.ts","line":42,"character":38,"text":"clientId","kind":1},{"file":"src/Providers/AWSAppSyncProvider.ts","line":66,"character":8,"text":"client","kind":1},{"file":"src/Providers/AWSAppSyncProvider.ts","line":73,"character":38,"text":"options","kind":1},{"file":"src/Providers/AWSAppSyncProvider.ts","line":88,"character":12,"text":"mqttConnections","kind":1},{"file":"src/Providers/AWSAppSyncProvider.ts","line":88,"character":34,"text":"newSubscriptions","kind":1},{"file":"src/Providers/AWSAppSyncProvider.ts","line":105,"character":13,"text":"connectionInfoForTopic","kind":1},{"file":"src/Providers/AWSAppSyncProvider.ts","line":105,"character":38,"text":"mqttConnections","kind":1},{"file":"src/Providers/AWSAppSyncProvider.ts","line":105,"character":54,"text":"find","kind":1},{"file":"src/Providers/AWSAppSyncProvider.ts","line":106,"character":8,"text":"c","kind":1},{"file":"src/Providers/AWSAppSyncProvider.ts","line":106,"character":13,"text":"c","kind":1},{"file":"src/Providers/AWSAppSyncProvider.ts","line":106,"character":15,"text":"topics","kind":1},{"file":"src/Providers/AWSAppSyncProvider.ts","line":106,"character":22,"text":"indexOf","kind":1},{"file":"src/Providers/AWSAppSyncProvider.ts","line":109,"character":11,"text":"connectionInfoForTopic","kind":1},{"file":"src/Providers/AWSAppSyncProvider.ts","line":110,"character":24,"text":"clientId","kind":1},{"file":"src/Providers/AWSAppSyncProvider.ts","line":110,"character":34,"text":"url","kind":1},{"file":"src/Providers/AWSAppSyncProvider.ts","line":112,"character":17,"text":"clientId","kind":1},{"file":"src/Providers/AWSAppSyncProvider.ts","line":113,"character":13,"text":"clientId","kind":1},{"file":"src/Providers/AWSAppSyncProvider.ts","line":114,"character":10,"text":"url","kind":1},{"file":"src/Providers/AWSAppSyncProvider.ts","line":119,"character":12,"text":"clientId","kind":1},{"file":"src/Providers/AWSAppSyncProvider.ts","line":130,"character":10,"text":"client","kind":1},{"file":"src/Providers/AWSAppSyncProvider.ts","line":132,"character":7,"text":"client","kind":1},{"file":"src/Providers/AWSAppSyncProvider.ts","line":137,"character":54,"text":"error","kind":1},{"file":"src/Providers/AWSAppSyncProvider.ts","line":137,"character":61,"text":"err","kind":1},{"file":"src/Providers/AWSAppSyncProvider.ts","line":136,"character":15,"text":"err","kind":1},{"file":"src/Providers/AWSAppSyncProvider.ts","line":145,"character":11,"text":"client","kind":1},{"file":"src/Providers/AWSAppSyncProvider.ts","line":145,"character":18,"text":"isConnected","kind":1},{"file":"src/Providers/AWSAppSyncProvider.ts","line":146,"character":8,"text":"client","kind":1},{"file":"src/Providers/AWSAppSyncProvider.ts","line":146,"character":15,"text":"subscribe","kind":1},{"file":"src/Providers/AWSAppSyncProvider.ts","line":152,"character":13,"text":"client","kind":1},{"file":"src/Providers/AWSAppSyncProvider.ts","line":179,"character":37,"text":"value","kind":1},{"file":"src/Providers/AWSAppSyncProvider.ts","line":180,"character":9,"text":"topic","kind":1},{"file":"src/Providers/AWSAppSyncProvider.ts","line":180,"character":39,"text":"value","kind":1},{"file":"src/Providers/AWSAppSyncProvider.ts","line":183,"character":3,"text":"value","kind":1},{"file":"src/Providers/AWSAppSyncProvider.ts","line":183,"character":9,"text":"data","kind":1},{"file":"src/Providers/AWSAppSyncProvider.ts","line":183,"character":31,"text":"value","kind":1},{"file":"src/Providers/AWSAppSyncProvider.ts","line":183,"character":37,"text":"data","kind":1},{"file":"src/Providers/AWSAppSyncProvider.ts","line":190,"character":10,"text":"value","kind":1}]

TypeScript coverage report