AWS AppSync subscriptions disconnected

AlanChauchet commented 4 years ago

Describe the bug Hello, I started an Amplify project within my Flutter app and I am struggling with GraphQL subscriptions. I set up my environment with AppSync and I am able to perform queries and mutations but when trying to subscribe, the socket is instantly connected then disconnected. Here is the output of the console (where {appId} is replaced by my AppSync app id) :

Connecting to websocket: wss://{app_id}.appsync-realtime-api.eu-west-1.amazonaws.com/graphql...
flutter: Connected to websocket.
flutter: Disconnected from websocket.
flutter: Scheduling to connect in 5 seconds...
flutter: Connecting to websocket: wss://{app_id}.appsync-realtime-api.eu-west-1.amazonaws.com/graphql...
flutter: Connected to websocket.
flutter: Disconnected from websocket.
flutter: Scheduling to connect in 5 seconds...

(It doesn't wait for inactivityTimeout but instantly gets disconnected)

Here is the code that I use to perform the subscription:

import 'package:graphql/client.dart';

class GraphQLService {
  static HttpLink _httpLink;
  static WebSocketLink _webSocketLink;
  static InMemoryCache cache = InMemoryCache();
  static GraphQLClient _httpClient;
  static GraphQLClient _websocketClient;

  static void setupClient(String token) {
    final AuthLink authLink = AuthLink(
      getToken: () => token
    );

    _httpLink = HttpLink(
      uri: GRAPHQL_URL,
    );

    _webSocketLink = WebSocketLink(
      url: GRAPHQL_URL_REALTIME,
      config: SocketClientConfig(
        autoReconnect: true,
        inactivityTimeout: Duration(seconds: 30),
      )
    );

    Link httpLink = authLink.concat(_httpLink);
    Link webSocketlink = authLink.concat(_webSocketLink);

    _httpClient = GraphQLClient(link: httpLink, cache: cache);
    _websocketClient = GraphQLClient(link: webSocketlink, cache: cache);
  }

  static Future<QueryResult> query(String query, { Map<String, dynamic> variables }) {
    return _httpClient.query(QueryOptions(
      documentNode: gql(query),
      variables: variables
    ));
  }

  static Future<QueryResult> mutate(String mutation, { Map<String, dynamic> variables }) {
    return _httpClient.mutate(MutationOptions(
      documentNode: gql(mutation),
      variables: variables
    ));
  }

  static Stream<FetchResult> subscribe(String subscription, String operationName, { Map<String, dynamic> variables }) {
    return _websocketClient.subscribe(Operation(
      documentNode: gql(subscription),
      variables: variables,
      operationName: operationName
    ));
  }
}

// Start subscription in another service
final token = AuthService.getJwtToken();
GraphQLService.setupClient(token);

GraphQLService.subscribe("""
  subscription OnTodoAdded {
    onCreateTodo {
      id
      name
      description
      createdAt
      updatedAt
    }
  }
  """, 'OnTodoAdded').listen((event) {
    print('Todo added sub');
    print(event.data);
    print(event.errors);
  }).onError((err) => print(err));

Expected behavior If an error is preventing me from connecting to the socket, I should be warned of it. And I am not sure of what could be wrong in what I am doing there but subscription should be working (the subscription has been tested on the AppSync console and is working as expected)

fadulalla commented 4 years ago

I don't think this is a bug with this library. Unless recently changed, connecting to AppSync subscriptions through third-party gql libraries is not straightforward.

As far as I know, AppSync doesn't use wss://. I have tried this before, and my http connection was never "upgraded to a wss" connection. I had to POST a request to the http endpoint requesting a subscription connection, then AWS returned connection details containing a new endpoint (a wss url), a topic and a client id. Only then could I connect to the socket using using a third party library (using the URL AWS returned). I had to dig into their official sdk to find this out, and then through trial and error to get it to work.

And now I realise that maybe this library doesn't even work with AppSync (for subscriptions) at all. AppSync seems to have its own weird implementation. Using the url, the client and the array of topics which means you might have to use the websocket library this library is using under the hood, and pass in the response you get from AWS to connect.

If you want to have a go, I've written a class that does this. I'm not sure if it would help your particular case, but it's worth a shot. It's in C#, but converting it to Dart should be relatively straightforward.

Create an http request to your endpoint, where the content is your query, your operation name and your variables:

private StringContent PrepareSubscriptionRequest(string query, string operationName, Dictionary<string, object> variables) {
var contentDictionary = new Dictionary<string, object> {
    { nameof(query), query },
    { nameof(operationName), operationName },
    { nameof(variables), variables }
};
var content = JsonConvert.SerializeObject(contentDictionary);
return new StringContent(content);
}

POST that to your endpoint (make sure your request headers have the api key under header name x-api-key, and content type of application/json) :

private async Task<Subscription> SubscribeAsync(StringContent subscriptionRequest) {
SubscriptionIntermediateResponse intermediateResponse = null;
try {
    var response = await _httpClient.PostAsync(Consts.CONFIG_APPSYNC_END_POINT, subscriptionRequest);
    if (response.IsSuccessStatusCode) {
        var responseContentString = await response.Content.ReadAsStringAsync();
        intermediateResponse = JsonConvert.DeserializeObject<SubscriptionIntermediateResponse>(responseContentString);
    }
    else {
        Logger.LogError(response.ReasonPhrase);
    }
}
catch (Exception e) {
    Logger.LogError(e);
}

if (intermediateResponse?.data is null) {
    return new Subscription() { Success = false };
}

string mqttConnection = intermediateResponse.extensions.subscription.mqttConnections[0].url;
string[] topics = intermediateResponse.extensions.subscription.mqttConnections[0].topics;
string client = intermediateResponse.extensions.subscription.mqttConnections[0].client;

return new Subscription() {
    MqttWssConnectionString = mqttConnection,
    Client = client,
    Topics = topics,
    Success = true
};
}

If successful, AWS will return a JSON object that contains a wss URL, an array of topics, and a client id, for you to connect with using a ws client:

private IMqttClientOptions MakeOptions(Subscription subscription) {
return new MqttClientOptionsBuilder()
            .WithWebSocketServer(subscription.MqttWssConnectionString)
            .WithClientId(subscription.Client)
            .WithTls()
            .WithProtocolVersion(MQTTnet.Serializer.MqttProtocolVersion.V311)
            .WithCleanSession()
            .Build();
}

I also started out with AppSync, but then a few months later ditched it and set up my own apollo server on an ec2 instance. AppSync had too many restrictions, for my needs anyway. Also often times things would just not work, without any explanation:

Subscription fields in a mutation had to be lowercase (camel case, I was trying to use them with Pascal case). Uppercase them and you won't get any subscription messages 🤷.
You can't do any complex filtering. My case was: users would subscribe to one or many chat rooms. My goal was to only make one subscription, using the user's id. Then ideally all messages in a chat room a user was part of would go to that user. That is not possible without having the user literally subscribe to all chat rooms (using something like chatRoomId as a filter).
When connecting, I had to trial with "QoS". Setting it to "At most once" or "Exactly once" didn't seem to work, and only "At least once" would, with no explanation, and no mention in the documentation. ... Just to name a few.

This was my experience back in 2018, they might have updated it. I'd suggest you check and make sure all the features you need are supported, before you invest time into making AppSync work.

micimize commented 4 years ago

Here's another thread on AppSync subscriptions: https://github.com/zino-app/graphql-flutter/issues/209

idk anything about AppSync but it seems like they use a different protocol from websockets for subscriptions

jtn-devecto commented 3 years ago

The protocol is described here Building a Real-time WebSocket Client.

I am stuck in how I can create Operation for subscription registration. It should be like this:

{
    "id": "ee849ef0-cf23-4cb8-9fcb-152ae4fd1e69",
    "payload": {
        "data": "{\"query\":\"subscription onCreateMessage {\\n onCreateMessage {\\n __typename\\n message\\n }\\n }\",\"variables\":{}}",
        "extensions": {
            "authorization": {
                "Authorization": "xxx",
                "host": "example1234567890000.appsync-api.us-east-1.amazonaws.com"
            }
        }
    },
    "type": "start"
}

This operation:

Operation(
      documentNode: gql("subscription onCreateMessage { onCreateMessage { __typename message } }"),
      variables: <String,dynamic> {
        'var': 'value'
      },
      extensions: <String,dynamic> { 
        "authorization": {
          "Authorization": "xxx",
          "host": "example1234567890000.appsync-api.us-east-1.amazonaws.com"
        }
      }
    )

produces the message

{
    "type": "start",
    "id": "8e83b832-16f5-4c43-af47-7ff24b058414",
    "payload": {
        "operationName": "onCreateMessage",
        "query": "subscription onCreateMessage {\n  onCreateMessage {\n    __typename\n    message\n  }\n}",
        "variables": {
            "var": "value"
        }
    }
}

The payload is different what AppSync expects. There should be fields "data" and "extensions". "data" is clearly combination of "query" and "variables", "extensions" is missing and "operationName" is extra.

I didn't find a way to accomplish this, but I am not familiar with the library, so I would appreciate if someone could point me to right direction?

micimize commented 3 years ago

@jtn-devecto on the v4 alpha you can provide a custom RequestSerializer to the SocketClientConfig you pass to WebSocketLink. I don't remember how you would do so with v3 so def upgrade to 4.0.0-alpha.7 (it is very stable)

jtn-devecto commented 3 years ago

@micimize thanks for the tip. For v3 I end it up writing own link.

micimize commented 3 years ago

@jtn-devecto that's excellent! If you could put it in a gist I might have time to generalize and port it to v4

vytautas-pranskunas- commented 3 years ago

I am having same problem just i am using Apollo GraphQL subscriptions package with Redis hosted on heroku. Any ideas how to configure server to send keepAlive messages? I am thinking about this:

this.pubsub = new RedisPubSub({
                publisher: new Redis(process.env.REDIS_URL, { keepAlive: 10000 }),
                subscriber: new Redis(process.env.REDIS_URL, { keepAlive: 10000 }),
            });

Kifah commented 3 years ago

we are having a similar issue here. Would be great to have a working solution using this package for subscriptions.

jluisrojas commented 3 years ago

After some reading I got it to work. I implemented a custom RequestSerializer as @micimize commented and added a query string to the AWS AppSync endpoint containing the authorization parameters as described here Building a Real-time WebSocket Client.

This is the custom RequestSerializer:

class AppSyncRequest extends RequestSerializer {
  final Map<String, dynamic> authHeader;

  const AppSyncRequest({
    this.authHeader,
  });

  @override
  Map<String, dynamic> serializeRequest(Request request) => {
    "data": jsonEncode({
      "query": printNode(request.operation.document),
      "variables": request.variables,
    }),
    "extensions": {
      "authorization": this.authHeader,
    }
  };
}

And this is the client code:

const apiId = "example1234567890000";
const zone = "us-west-1";

final token = session.getAccessToken().getJwtToken();

String toBase64(Map data) => base64.encode(utf8.encode(jsonEncode(data)));

final authHeader = {
  "Authorization": token,
  "host": "$apiId.appsync-api.$zone.amazonaws.com",
};

final encodedHeader = toBase64(authHeader);

final WebSocketLink wsLink = WebSocketLink(
  'wss://$apiId.appsync-realtime-api.$zone.amazonaws.com/graphql?header=$encodedHeader&payload=e30=',
  config: SocketClientConfig(
    serializer: AppSyncRequest(authHeader: authHeader),
    inactivityTimeout: Duration(seconds: 60),
  )
);

final AuthLink authLink = AuthLink(
  getToken: () => token,
);

final Link link = authLink.concat(wsLink);

final client = GraphQLClient(
  link: link,
  cache: GraphQLCache(),
  alwaysRebroadcast: true
);

client.subscribe(SubscriptionOptions(
  document: gql("""
  subscription SubPost {
    addedPost {
      id
      title
    }
  }
  """
))).listen((result) {
  print("Subscription data:");
  print(result.data);
});

varund29 commented 3 years ago

Hi, I'm also facing appsync subscription issue in flutter, which is working perfectly in angular application. and my issue explained clearly here , please have a look.

brianschardt commented 3 years ago

I am having this issue I use GCP cloud run, and am running into the same issue.

micimize commented 3 years ago

I'm going to close this as between https://github.com/zino-app/graphql-flutter/issues/682#issuecomment-759078492 and the s/o answer on amplify there seem to be solutions. If anyone who understands this better wants I would encourage a PR on updating the docs on AppSync Support.

capraqua commented 3 years ago

After some reading I got it to work. I implemented a custom RequestSerializer as @micimize commented and added a query string to the AWS AppSync endpoint containing the authorization parameters as described here Building a Real-time WebSocket Client.

This is the custom RequestSerializer:

class AppSyncRequest extends RequestSerializer {
  final Map<String, dynamic> authHeader;

  const AppSyncRequest({
    this.authHeader,
  });

  @override
  Map<String, dynamic> serializeRequest(Request request) => {
    "data": jsonEncode({
      "query": printNode(request.operation.document),
      "variables": request.variables,
    }),
    "extensions": {
      "authorization": this.authHeader,
    }
  };
}

And this is the client code:

const apiId = "example1234567890000";
const zone = "us-west-1";

final token = session.getAccessToken().getJwtToken();

String toBase64(Map data) => base64.encode(utf8.encode(jsonEncode(data)));

final authHeader = {
  "Authorization": token,
  "host": "$apiId.appsync-api.$zone.amazonaws.com",
};

final encodedHeader = toBase64(authHeader);

final WebSocketLink wsLink = WebSocketLink(
  'wss://$apiId.appsync-realtime-api.$zone.amazonaws.com/graphql?header=$encodedHeader&payload=e30=',
  config: SocketClientConfig(
    serializer: AppSyncRequest(authHeader: authHeader),
    inactivityTimeout: Duration(seconds: 60),
  )
);

final AuthLink authLink = AuthLink(
  getToken: () => token,
);

final Link link = authLink.concat(wsLink);

final client = GraphQLClient(
  link: link,
  cache: GraphQLCache(),
  alwaysRebroadcast: true
);

client.subscribe(SubscriptionOptions(
  document: gql("""
  subscription SubPost {
    addedPost {
      id
      title
    }
  }
  """
))).listen((result) {
  print("Subscription data:");
  print(result.data);
});

Thank you, your solution is working for me. Saved me hours.

flutterdev-2021 commented 3 years ago

After some reading I got it to work. I implemented a custom RequestSerializer as @micimize commented and added a query string to the AWS AppSync endpoint containing the authorization parameters as described here Building a Real-time WebSocket Client. This is the custom RequestSerializer:

class AppSyncRequest extends RequestSerializer {
  final Map<String, dynamic> authHeader;

  const AppSyncRequest({
    this.authHeader,
  });

  @override
  Map<String, dynamic> serializeRequest(Request request) => {
    "data": jsonEncode({
      "query": printNode(request.operation.document),
      "variables": request.variables,
    }),
    "extensions": {
      "authorization": this.authHeader,
    }
  };
}

And this is the client code:

const apiId = "example1234567890000";
const zone = "us-west-1";

final token = session.getAccessToken().getJwtToken();

String toBase64(Map data) => base64.encode(utf8.encode(jsonEncode(data)));

final authHeader = {
  "Authorization": token,
  "host": "$apiId.appsync-api.$zone.amazonaws.com",
};

final encodedHeader = toBase64(authHeader);

final WebSocketLink wsLink = WebSocketLink(
  'wss://$apiId.appsync-realtime-api.$zone.amazonaws.com/graphql?header=$encodedHeader&payload=e30=',
  config: SocketClientConfig(
    serializer: AppSyncRequest(authHeader: authHeader),
    inactivityTimeout: Duration(seconds: 60),
  )
);

final AuthLink authLink = AuthLink(
  getToken: () => token,
);

final Link link = authLink.concat(wsLink);

final client = GraphQLClient(
  link: link,
  cache: GraphQLCache(),
  alwaysRebroadcast: true
);

client.subscribe(SubscriptionOptions(
  document: gql("""
  subscription SubPost {
    addedPost {
      id
      title
    }
  }
  """
))).listen((result) {
  print("Subscription data:");
  print(result.data);
});

Thank you, your solution is working for me. Saved me hours.

Can you please ellaborate , from where this session comes ? final token = session.getAccessToken().getJwtToken();

jluisrojas commented 3 years ago

@flutterdev-2021 The session comes from the current authenticated user. For this you can use the amazon_cognito_identity_dart package.

Here is a more detailed example: https://github.com/furaiev/amazon-cognito-identity-dart-2/#for-appsyncs-graphql

zino-hofmann / graphql-flutter

AWS AppSync subscriptions disconnected #682