Merge branch 'master' into patch-2

zegates · Aug 23, 2019 · 54e6e1f · 54e6e1f
2 parents e4f0dba + 5526728
commit 54e6e1f
Show file tree

Hide file tree

Showing 5 changed files with 40 additions and 5 deletions.
diff --git a/lgtm.yml b/lgtm.yml
@@ -0,0 +1,5 @@
+extraction:
+  javascript:
+    index:
+      filters:
+        - exclude: "docs"
diff --git a/packages/pubsub/__tests__/PubSub-unit-test.ts b/packages/pubsub/__tests__/PubSub-unit-test.ts
@@ -213,6 +213,22 @@ describe('PubSub', () => {
                     provider: 'AWSIoTProvider',
             })).rejects.toMatchObject({error: new Error('Failed to connect to the network')});
         });
+
+        test('trigger observer error when disconnected', (done) => {
+            const pubsub = new PubSub();
+
+            const awsIotProvider = new AWSIoTProvider({
+                aws_pubsub_region: 'region',
+                aws_pubsub_endpoint: 'wss://iot.mymockendpoint.org:443/notrealmqtt'
+            });
+            pubsub.addPluggable(awsIotProvider);
+
+            pubsub.subscribe('topic').subscribe({
+                error: () => done()
+            });
+
+            awsIotProvider.onDisconnect({ errorCode: 1, clientId: '123' });
+        });
     });
 
     describe('MqttOverWSProvider local testing config', () => {

diff --git a/packages/pubsub/src/Providers/AWSAppSyncProvider.ts b/packages/pubsub/src/Providers/AWSAppSyncProvider.ts
@@ -129,10 +129,17 @@ export class AWSAppSyncProvider extends MqttOverWSProvider {
                 // reconnect everything we have in the map
                 await Promise.all(map.map(async ([clientId, { url, topics }]) => {
                     // connect to new client
-                    const client = await this.connect(clientId, {
-                        clientId,
-                        url,
-                    });
+                    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

diff --git a/packages/pubsub/src/Providers/MqttOverWSProvider.ts b/packages/pubsub/src/Providers/MqttOverWSProvider.ts
@@ -89,7 +89,14 @@ export class MqttOverWSProvider extends AbstractPubSubProvider {
     public onDisconnect({ clientId, errorCode, ...args }) {
         if (errorCode !== 0) {
             logger.warn(clientId, JSON.stringify({ errorCode, ...args }, null, 2));
+            this._topicObservers.forEach((observerForTopic, _observerTopic) => {
+                observerForTopic.forEach(observer => {
+                    observer.error('Disconnected, error code: ' + errorCode);
+                    observer.complete();
+                });
+            });
         }
+        this._topicObservers = new Map();
     }
 
     public async newClient({ url, clientId }: MqttProvidertOptions): Promise<any> {

diff --git a/packages/pushnotification/src/PushNotification.ts b/packages/pushnotification/src/PushNotification.ts
@@ -328,7 +328,7 @@ export default class PushNotification {
         }
 
         let ret = null;
-        const dataPayload = dataObj.data;
+        const dataPayload = dataObj.data || {};
         if (dataPayload['pinpoint.campaign.campaign_id']) {
             ret = {
                 title: dataPayload['pinpoint.notification.title'],