// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. // SPDX-License-Identifier: Apache-2.0 import { Amplify, browserOrNode, Category, ConsoleLogger as Logger, CustomUserAgentDetails, INTERNAL_AWS_APPSYNC_REALTIME_PUBSUB_PROVIDER, PubSubAction, } from '@aws-amplify/core'; import { PubSubProvider, PubSubOptions, ProviderOptions } from '../types'; import { AWSAppSyncRealTimeProvider } from '../Providers'; import { PubSubContent } from '../types/PubSub'; import Observable from 'zen-observable-ts'; const { isNode } = browserOrNode(); const logger = new Logger('PubSub'); type PubSubObservable = { provider: PubSubProvider; value: string | Record; }; export class InternalPubSubClass { private _options: PubSubOptions; private _pluggables: PubSubProvider[]; /** * Internal instance of AWSAppSyncRealTimeProvider used by the API category to subscribe to AppSync */ private _awsAppSyncRealTimeProvider?: AWSAppSyncRealTimeProvider; /** * Lazy instantiate AWSAppSyncRealTimeProvider when it is required by the API category */ private get awsAppSyncRealTimeProvider() { if (!this._awsAppSyncRealTimeProvider) { this._awsAppSyncRealTimeProvider = new AWSAppSyncRealTimeProvider( this._options ); } return this._awsAppSyncRealTimeProvider; } /** * Initialize PubSub with AWS configurations * * @param {PubSubOptions} options - Configuration object for PubSub */ constructor(options?: PubSubOptions) { this._options = options ?? {}; logger.debug('PubSub Options', this._options); this._pluggables = []; this.subscribe = this.subscribe.bind(this); } public getModuleName() { return 'InternalPubSub'; } /** * Configure PubSub part with configurations * * @param {PubSubOptions} config - Configuration for PubSub * @return {Object} - The current configuration */ configure(options: PubSubOptions) { const opt: Record = options ? options.PubSub || options : {}; logger.debug('configure PubSub', { opt }); this._options = Object.assign({}, this._options, opt); this._pluggables.map(pluggable => pluggable.configure(this._options)); return this._options; } /** * add plugin into Analytics category * @param {Object} pluggable - an instance of the plugin */ public async addPluggable(pluggable: PubSubProvider) { if (pluggable && pluggable.getCategory() === 'PubSub') { this._pluggables.push(pluggable); const config = pluggable.configure(this._options); return config; } } /** * remove plugin from PubSub category * @param providerName - the name of the plugin */ removePluggable(providerName: string): void { this._pluggables = this._pluggables.filter( pluggable => pluggable.getProviderName() !== providerName ); } private getProviderByName(providerName: string | symbol) { if (providerName === INTERNAL_AWS_APPSYNC_REALTIME_PUBSUB_PROVIDER) { return this.awsAppSyncRealTimeProvider; } return this._pluggables.find( pluggable => pluggable.getProviderName() === providerName ); } private getProviders(options: ProviderOptions = {}) { const providerName = options.provider; if (!providerName) { return this._pluggables; } const provider = this.getProviderByName(providerName); if (!provider) { throw new Error(`Could not find provider named ${String(providerName)}`); } return [provider]; } async publish( topics: string[] | string, msg: PubSubContent, options?: ProviderOptions ) { return Promise.all( this.getProviders(options).map(provider => provider.publish(topics, msg, options) ) ); } subscribe( topics: string[] | string, options?: ProviderOptions, customUserAgentDetails?: CustomUserAgentDetails ): Observable { if (isNode && this._options && this._options.ssr) { throw new Error( 'Subscriptions are not supported for Server-Side Rendering (SSR)' ); } logger.debug('subscribe options', options); const providers = this.getProviders(options); const pubSubUserAgentDetails: CustomUserAgentDetails = { category: Category.PubSub, action: PubSubAction.Subscribe, ...customUserAgentDetails, }; return new Observable(observer => { const observables = providers.map(provider => ({ provider, observable: provider.subscribe(topics, options, pubSubUserAgentDetails), })); const subscriptions = observables.map(({ provider, observable }) => observable.subscribe({ start: console.error, next: (value: PubSubContent) => observer.next({ provider, value }), error: (error: unknown) => observer.error({ provider, error }), // complete: observer.complete, // TODO: when all completed, complete the outer one }) ); return () => subscriptions.forEach(subscription => subscription.unsubscribe()); }); } } export const InternalPubSub = new InternalPubSubClass(); Amplify.register(InternalPubSub);