Amplify-js: PubSub reconnect

[leantorres73]

Related to https://github.com/aws-amplify/amplify-js/issues/1844, if I close the websocket connections and try to resubscribe does not reconnect the connection.

To Reproduce
Steps to reproduce the behavior:

1. Subscribe to a topic, then close the connection and then try to resubscribe to the same topic again.

Describe the solution you'd like
There should be a way to reconnect to a closed websocket connection.

A method like removePlugabble must be included to fix and have the ability to unsubscribe and start over again.

[kirkryan]

We also need this. If you lose signal on a portable device you're left without a valid clean method of re-establishing a connection.

[timoteialbu]

Why are none of the maintainers acknowledging this and offering some support? We had to do some very complex reconnect functionality which is a little frail and always needs fine tuning. I believe the PubSub module should handle these scenarios that are very common (signal drops)

[kirkryan]

Worst thing is that MQTT.js DOES have this functionality and that the AWS IoT client that is built on top of it simply ignores it for some reason. I’ve heard that the AWS team are working on a new release that will improve this but would like to see a comment here from the team.

As it stands - the smallest blip in connectivity for longer than a few seconds will cause the AWS IoT provider to disconnect and there is no way to reconnect it or tests its current state.

[elorzafe]

@leantorres73 I tried this code on a react-native app

import React, { useEffect, useState } from 'react';
import { StyleSheet, Text, View, Button } from 'react-native';
import Amplify, { Analytics, API, PubSub } from 'aws-amplify';
import { AWSIoTProvider } from "@aws-amplify/pubsub/lib/Providers";

Analytics.disable();

Amplify.configure({
  Auth: {
    identityPoolId: "us-west-2:xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxx",
    region: 'us-west-2'
  }
});
Amplify.addPluggable(new AWSIoTProvider({
  aws_pubsub_region: 'us-west-2',
  aws_pubsub_endpoint: 'wss://xxxxxxxxxxxx-ats.iot.us-west-2.amazonaws.com/mqtt',
}));

let subscription;

function App() {
  const [message, setMessage] = useState('Open up App.js to start working on your app!');
  useEffect(() => {
    connect();
  }, []);


  function connect() {
    subscription = PubSub.subscribe('myTopic').subscribe({
      next: data => setMessage(JSON.stringify(data, null, 2)),
      error: error => setMessage(JSON.stringify(error, null, 2)),
      close: () => setMessage('Done'),
    });
  };

  function disconnect() {
    subscription.unsubscribe();
  };

  return (
    <View style={styles.container}>
      <Button title="connect" onPress={connect}></Button>
      <Button title="disconnect" onPress={disconnect}></Button>
      <Text>{message}</Text>
    </View>
  );
}

export default App;

const styles = StyleSheet.create({
  container: {
    flex: 1,
    backgroundColor: '#fff',
    alignItems: 'center',
    justifyContent: 'center',
  },
});

I was able to subscribe to the same topic without issues.

[elorzafe]

@timoteialbu @kirkryan sorry for the late response, for errors on connection on observer I merge a pr this week that solves that problem (pr #3890) so you can reconnect from there, can you try latest unstable version.

[kirkryan]

@timoteialbu @kirkryan sorry for the late response, for errors on connection on observer I merge a pr this week that solves that problem (pr #3890) so you can reconnect from there, can you try latest unstable version.

Hi @elorzafe - we’ll try it out this week thank you. I believe this will fix part 1 of the overall issue (detection of a disconnect), however there is still no clean way of re-establishing a connection. Simply running addPluggable restablishes the existing connection AND a new connection resulting in two client IDs and duplicate messages, what we really need is a method of either:

1: IOTpluggabke reconnect
Or

1. IOTremovepluggable so that we can clear the old stale connection then run an addPluggable again

Does that clarify the issue?

[elorzafe]

@kirkryan if you try to subscribe to the same topic (without adding the plugin) for me some reasion this not working for you? Let me know how it goes.

[kirkryan]

@elorzafe - this doesn't work as the WebSocket itself is disconnected at this point and therefore there is no way to re-activate it, other than running an addPluggable - which then adds a new WebSocket but you then have 2 client ID's subscribed and start to receive duplicate messages!

What we really need is:

1. A method to check the state of the WebSocket connection (pluggableState?)
2. A method to reconnect (or auto-reconnect?) the WebSocket connection (reconnectPluggable?)
3. A method to disconnect or remove the existing connection added by addPluggable, so that we can run it again and not have duplicate messages

Does that make sense?

[kirkryan]

Hi @elorzafe - I upgraded our amplify to the latest version as of today. I can confirm that when the client kills the WebSocket (after approx 2 mins if you lock your phone), the app shows a code 7: Socket error undefined when you bring it back to the foreground (expected as the app is suspended therefore the client will hit timeout and the connection becomes invalid.

If you resubscribe to a topic it looks like the WebSocket is re-connected (without having to run an addPluggable command).

Therefore this leaves the final piece of the puzzle, how do we check the state of the client so that we can cleanly handle having to resubscribe to all relevant topics?

[mevert]

I think that we have similar issue (using aws-amplify v. 1.1.40
). For example, if I subscribe to two topics:

subscriptions = PubSub.subscribe([‘topic1’, ‘topic2’]).subscribe({
      next: data => setMessage(JSON.stringify(data, null, 2)),
      error: error => setMessage(JSON.stringify(error, null, 2)),
      close: () => setMessage('Done'),
    })

And then

subscriptions.unsubscribe()

When I try to subscribe again to ‘topic1’, I can no longer receive its messages. What is the recommended way to subscribe topics again with aws-amplify? @elorzafe ?

Edit: fixed typing error in my example.

[mevert]

I think that we have similar issue (using aws-amplify v. 1.1.40
). For example, if I subscribe to two topics:

subscriptions = PubSub.subscribe(‘topic1’, ‘topic2’).subscribe({
      next: data => setMessage(JSON.stringify(data, null, 2)),
      error: error => setMessage(JSON.stringify(error, null, 2)),
      close: () => setMessage('Done'),
    })

And then

subscriptions.unsubscribe()

When I try to subscribe again to ‘topic1’, I can no longer receive its messages. What is the recommended way to subscribe topics again with aws-amplify? @elorzafe ?

I just noticed that subscribing again to old topics after unsubscribing used to work with aws-amplify v. 1.1.30 but does not work anymore with the latest v. 1.1.40. Is it possible that there is some breaking change between these version? @elorzafe So I will have to use the old version. Edit: Did not get it working again even with 1.1.30 when I deleted my package-lock.json, node_modules and reinstalled everything. Perhaps I had some old version hanging or something... Would be really nice to get this fixed soon.

[elorzafe]

subscriptions = PubSub.subscribe(‘topic1’, ‘topic2’).subscribe({
     next: data => setMessage(JSON.stringify(data, null, 2)),
     error: error => setMessage(JSON.stringify(error, null, 2)),
     close: () => setMessage('Done'),
   })

@mevert topic2 on that case is providerOptions not a second topic. The first parameter of subscribe is a string or string[] it should be like this.

subscriptions = PubSub.subscribe([‘topic1’, ‘topic2’]).subscribe({
      next: data => setMessage(JSON.stringify(data, null, 2)),
      error: error => setMessage(JSON.stringify(error, null, 2)),
      close: () => setMessage('Done'),
    })

[elorzafe]

@kirkryan why do you need the state if the observable is closing the subscription and sending the error? I think is cleaner to handle reconnection logic than having a process that is checking state. Another option could be sending events on Hub that you can handle the resubcription logic.

How did you plan to check the state of the client?

[mevert]

subscriptions = PubSub.subscribe(‘topic1’, ‘topic2’).subscribe({
     next: data => setMessage(JSON.stringify(data, null, 2)),
     error: error => setMessage(JSON.stringify(error, null, 2)),
     close: () => setMessage('Done'),
   })

@mevert topic2 on that case is providerOptions not a second topic. The first parameter of subscribe is a string or string[] it should be like this.

subscriptions = PubSub.subscribe([‘topic1’, ‘topic2’]).subscribe({
      next: data => setMessage(JSON.stringify(data, null, 2)),
      error: error => setMessage(JSON.stringify(error, null, 2)),
      close: () => setMessage('Done'),
    })

Sorry, I had typo when writing it here. However, were are using string[] in our real case. Any idea that did your previous changes to PubSub affect to this subscription problem since I don't have the same problem with "@aws-amplify/pubsub": "1.1.0" ?

[elorzafe]

@mevert I will look into this

[stale[bot]]

This issue has been automatically marked as stale because it has not had recent activity. It will be closed if no further activity occurs. Thank you for your contributions.

[nayan27]

Hi,

Is there any progress on this ? I am having a similar issue:

1 - My applicationA subscribe to a few topics
2- My applicationA opens my other applicationB
3- My applicationB does not use Amplify and only perform specific tasks.
4- After about 1-2 minutes (but could be more), applicationB reopens back applicationA
5- Upon foreground, applicationA's subscriptions call their error function with error: "Disconnected, error code 8"

react-native: 0.57.8
aws-amplify: ^1.2.2

How can i handle this ? is there a way to reconnect automatically ?

Thank you

[stale[bot]]

This issue has been automatically marked as stale because it has not had recent activity. It will be closed if no further activity occurs. Thank you for your contributions.

[peterecofit]

I would love to see a clear example of how to reconnect on disconnect.

[camin-mccluskey]

Also struggling with this using graphql subscriptions within react native.

graphqlOperation(onCreateMessage, {messageHouseId: this.props.house.id})
  ).subscribe({
    next: (message) => {
      const messageObj = message.value.data.onCreateMessage;
      const { user_id : sender_id } = messageObj.user;
      if(sender_id !== this.props.user.user_id) {
        this.props.sendMessage({message: [messageObj]});
      }
    },
    error: err => console.log(err),
    close: () => console.log('Done')
  });

When the app goes to the background the the subscription errors out after a couple of minutes. Should I resubscribe inside error: err => {...}??

[stale[bot]]

This issue has been automatically marked as stale because it has not had recent activity. It will be closed if no further activity occurs. Thank you for your contributions.

[houmark]

Still an issue, should not be closed!

[peterecofit]

Very much agree. This needs a clear resolution documented!

[peterecofit]

Bumping again. We need a clear solution!

[justinjoyn]

Keeping this from going stale. Need a fix soon

[Dinesh5799]

Hi All,
I came across this issue in Mid-January and implemented my own logic for re-connecting using PubSub.subscribe api of Amplify but recently I saw it not working as there was an underlying issue in the internal implementation of library(I dug deep into and noticed that if the client is disconnected from Network Connection and if the client re-connects before it runs into error part of PubSub.subscribe it works, but if it runs into error part then it never re-connects). So, I after researching for a quite long time I used another approach where I can now re-connect without any issue. But this approach does not use PubSub from amplify. I also tried contacting AWS support but none of their said things worked.

Work Around Explanation :-

• I tried to use mqtt which not only has re-connect mechanism but also sends all the messages if the connection is lost for short time(I mean before the socket disconnects and tries to establish a new socket connection). But if entirely disconnects and re-connects you can have a mechanism where you can call all the required api to stay updated.

• But to use the above library I am generating and signing my own Socket url which is quite easy with SigV4Utils.

• Please find attached AWS_MQTT_IOT_Reconnect repo where you can find the code for all that is said above. Please read inline comments for better understanding.

[Thembelani]

Hi everyone. Has this issue been resolved ?

[nubpro]

This issue should be a priority, it impacts DataStore #6162 and it's giving us a tough time of using it. We are also in the midst of dropping it.
Imagine you are building a food delivery system for a franchise and none of the branch can receive any new orders because of this, amend to that.

This issue is raised on Oct 2018 #1844 and has not been solved ever since. 3 more months before it reaches the 2 years anniversary.

Seriously? What is taking so long and why isnt this a priority?
This just shows how unproven Amplify is and how no one should sink their feet into this blackhole if they'd really care for a stable and reliable app.

Come on, please do better than this. Please put in more resources into fixing bugs and getting Amplify to be as stable as possible. Hell, make apps with it and battle test this piece of thing. Does Amazon even use Amplify for their projects?
You know what, don't use Discord. Build a chat communicaton app with Amplify. Let that be the testing ground.

If you guys aren't doing any of these, Amplify just looks like another pet project to me.
Amplify team cranks out Amplify -> Only performs regular unit testing -> Developers do the integration -> Developers complaints -> Amplify team doesn't respond or takes months to respond
How could we rely on such a fragile framework to take on any actual production projects when the Amplify team can't be bothered enough to ensure the essentials are working as advertised?

Either the Amplify team does not seem to be motivated enough to get this fixed because they don't neccessarily have to or that they don't understand how big of the problem this is causing.

[cuongvo]

This issue is a significant blocker to customers. A library that's mobile and web centric with its bi-directional communication component not handling reconnects and for this long, after this many issues? Decide as a team / company if you want to support builders who will make actual products or if you want to keep pushing for flashy demos that go nowhere.

---

A workaround I've had to implement is to reach into the internals of PubSub and reset the pluggables myself here. You can take advantage of the code not caring about mutability and returning a reference to the array of pluggables itself by calling getProviders without a provider name (logic here).

To handle reconnects yourself, you'll need to remove the existing pluggable and create a new one, with a new client ID. The problem is on intermittent connection drops, the client id (either generated automatically if not passed in or the one you provided via options) is still considered connected, resulting in the mentioned Disconnected, error code 8 on socket close when you attempt to reconnect.

Even after cleaning up on client disconnect (here), on new subscribes, in the PubSub wrapper, the existing client id continues to be used here.

tl;dr:

1. On intermittent disconnects (observable emits an error with Disconnected, error code 8) handle clearing existing pluggables with established client ids by reaching into the singleton PubSub implementation and clearing the _pluggables private value array.
2. Create a new pluggable to generate / pass in a new client id and register it again.
3. Re-attempt to subscribe to the observable with the topic and the underlying Paho client will reconnect with the new client id.

I'm happy to send in a pull request if the team wants to provide direction on how they think this should be solved knowing the above workaround and underlying issue.

[kirkryan]

I can confirm this has been a long time problem for our field based units and we have had to implement workarounds.

AWS : your IoT implementation never reconnects if an app gets suspended (screen switched off) making it pretty much unusable.

[mevert]

Hey, @iartemiev could you also check the related issue https://github.com/aws-amplify/amplify-js/issues/2692 and please let us know when we can expect the solutions coming?

[stale[bot]]

This issue has been automatically marked as stale because it has not had recent activity. It will be closed if no further activity occurs. Thank you for your contributions.

[stale[bot]]

This issue has been automatically closed because of inactivity. Please open a new issue if are still encountering problems.

[rajatraj733]

any updates on this? this issue should not be closed, if not resolved yet.

[mevert]

@iartemiev and other AWS people, what is the status of this issue? This should not be closed.

[kirkryan]

We used the repo from @Dinesh5799 and it works brilliantly - all devices out in the field now reconnect when losing signal intermittently- I highly suggest you check out his awesome work.

[valerusg]

I was having the same problem, and like @cuongvo said, the "clientId" is always the same.
I overrided the PubSub class by implementing it again and adding a new function in order to restart the client. The final code is:

`
import Observable from "zen-observable-ts";

import {
Amplify,
browserOrNode,
INTERNAL_AWS_APPSYNC_PUBSUB_PROVIDER,
INTERNAL_AWS_APPSYNC_REALTIME_PUBSUB_PROVIDER,
} from "@aws-amplify/core";
import {PubSubProvider, PubSubOptions, ProvidertOptions} from "@aws-amplify/pubsub/src/types";
import {AWSAppSyncProvider, AWSAppSyncRealTimeProvider} from "@aws-amplify/pubsub/src/Providers";
// @ts-ignore
import {AWSIoTProvider} from "@aws-amplify/pubsub/lib";
import Config from "react-native-config";

const {isNode} = browserOrNode();

export class PubSubClass {
private _options: PubSubOptions;

private _pluggables: PubSubProvider[];

/**
 * Internal instance of AWSAppSyncProvider used by the API category to subscribe to AppSync
 */
private _awsAppSyncProvider: AWSAppSyncProvider | undefined;

/**
 * Internal instance of AWSAppSyncRealTimeProvider used by the API category to subscribe to AppSync
 */
private _awsAppSyncRealTimeProvider: AWSAppSyncRealTimeProvider | undefined;

/**
 * Lazy instantiate AWSAppSyncProvider when it is required by the API category
 */
private get awsAppSyncProvider() {
    if (!this._awsAppSyncProvider) {
        this._awsAppSyncProvider = new AWSAppSyncProvider(this._options);
    }
    return this._awsAppSyncProvider;
}

/**
 * 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 | any) {
    this._options = options;
    this.log("PubSub Options", this._options);
    this._pluggables = [];
    this.subscribe = this.subscribe.bind(this);
}

public getModuleName() {
    return "PubSub";
}

/**
 * Configure PubSub part with configurations
 *
 * @param {PubSubOptions} config - Configuration for PubSub
 * @return {Object} - The current configuration
 */
configure(options: PubSubOptions) {
    const opt = options ? options.PubSub || options : {};
    this.log("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;
    }
}

private getProviderByName(providerName: any) {
    if (providerName === INTERNAL_AWS_APPSYNC_PUBSUB_PROVIDER) {
        return this.awsAppSyncProvider;
    }
    if (providerName === INTERNAL_AWS_APPSYNC_REALTIME_PUBSUB_PROVIDER) {
        return this.awsAppSyncRealTimeProvider;
    }

    return this._pluggables.find(
        pluggable => pluggable.getProviderName() === providerName
    );
}

private getProviders(options: ProvidertOptions = {}) {
    const {provider: providerName} = options;
    if (!providerName) {
        return this._pluggables;
    }

    const provider = this.getProviderByName(providerName);
    if (!provider) {
        throw new Error(`Could not find provider named ${providerName}`);
    }

    return [provider];
}

public restartAWSIoTProvider() {
    this.log("Restarting AWSIoTProvider...");
    this._pluggables[0] = new AWSIoTProvider({
        aws_pubsub_region: Config.AMAZON_REGION,
        aws_pubsub_endpoint: `wss://${Config.AMAZON_PUBSUB_ENDPOINT}/mqtt`,
    });
}

async publish(
    topics: string[] | string,
    msg: any,
    options?: ProvidertOptions
) {
    return Promise.all(
        this.getProviders(options).map(provider =>
            provider.publish(topics, msg, options)
        )
    );
}

subscribe(
    topics: string[] | string,
    options?: ProvidertOptions
): Observable<any> {
    if (isNode) {
        throw new Error("Subscriptions are not supported in Node");
    }

    const providers = this.getProviders(options);

    return new Observable(observer => {
        const observables = providers.map(provider => ({
            provider,
            observable: provider.subscribe(topics, options),
        }));

        const subscriptions = observables.map(({provider, observable}) =>
            observable.subscribe({
                start: console.error,
                next: value => observer.next({provider, value}),
                error: error => observer.error({provider, error}),
                // complete: observer.complete, // TODO: when all completed, complete the outer one
            })
        );

        return () =>
            subscriptions.forEach(subscription => subscription.unsubscribe());
    });
}

private log(message: string, data?: any) {
    console.log("CustomPubSub: " + message, data);
}

}
export const PubSub = new PubSubClass(null);
Amplify.register(PubSub);
`

Start using this "PubSub" class import instead of the "import from 'amplify'", the final line "Amplify.register()" will override the PubSub class using this one instead.

The "restartAWSIoTProvider" method should be called after the "unsubscribe" function returned from "subscribe".

In my case it's working fine:

1. Send a command to the "iot" device.
2. After 5s waiting for response and no answer, the connection is restarted.
3. Call "unsubscribe" from topics.
4. Call "PubSub.restartAWSIoTProvider()".
5. Call "subscribe" -> Know we have a new clientId, In my case, this fixes the problem.

[dingdongping]

@valerusg
Thank you very much. I used your PubSub to solve the problem of reconnecting when the link was lost

subscribeTopic(){

    if (!this._device){
        _Log_('call subsribeTopic before,please call setConfig method');
        return;
    }
    // Auth.configure();
    const topic = `wwwwww`;
     this.curSubscriber= PubSub.subscribe(topic).subscribe({
            next: data => {
                try {
                    this._isConnected=true;
                    let payload=data.value;

                    if (payload.cmd=='subscribeSuccess'){
                        this._subSuccess&&this._subSuccess();
                    }else {
                        this._handlerMsg&& this._handlerMsg({topic,payload});
                    }

                }catch (err) {
                    _Log_('Decode Data catch:', err);
                }


            },
            error: error =>{
                this._isConnected=false;
                setTimeout(()=>{
                    PubSub.restartAWSIoTProvider();
                    this.subscribeTopic();
                },1000*10)
            },
            close: (val) => {
                _Log_('cancel:'+val)
            },
        }
    );




   setTimeout(()=>{
       PubSub.publish(topic,{ cmd: 'subscribeSuccess', })
   },400)


}

[houmark]

@valerusg would you consider making a pull request for this fix to be included in the main library? I think a lot of people would appreciate that.

[enriqueavg]

What is the status of this issue?