Amplify has re-imagined the way frontend developers build fullstack applications. Develop and deploy without the hassle.

Page updated Apr 29, 2024

Subscribe to real-time events

Amplify Flutter v1 is now in Maintenance Mode until April 30th, 2025. This means that we will continue to include updates to ensure compatibility with backend services and security. No new features will be introduced in v1.

Please use the latest version (v2) of Amplify Flutter to get started.

If you are currently using v1, follow these instructions to upgrade to v2.

Subscribe to mutations for creating real-time clients.

Setup subscription with callbacks

When creating subscriptions, a Stream object will be returned to you. This Stream will continue producing events until either the subscription encounters an error or you cancel the subscription. In the case of need for limiting the amount of data that is omitted, you can take advantage of the Stream's helper functions such as take. The cancellation occurs when the defined amount of event has occurred:

Stream<GraphQLResponse<Todo>> subscribe() {
final subscriptionRequest = ModelSubscriptions.onCreate(Todo.classType);
final Stream<GraphQLResponse<Todo>> operation = Amplify.API
.subscribe(
subscriptionRequest,
onEstablished: () => safePrint('Subscription established'),
)
// Listens to only 5 elements
.take(5)
.handleError(
(Object error) {
safePrint('Error in subscription stream: $error');
},
);
return operation;
}

Alternatively, you can call Stream.listen to create a StreamSubscription object which can be programmatically canceled.

// Be sure to import this
import 'dart:async';
...
StreamSubscription<GraphQLResponse<Todo>>? subscription;
void subscribe() {
final subscriptionRequest = ModelSubscriptions.onCreate(Todo.classType);
final Stream<GraphQLResponse<Todo>> operation = Amplify.API.subscribe(
subscriptionRequest,
onEstablished: () => safePrint('Subscription established'),
);
subscription = operation.listen(
(event) {
safePrint('Subscription event data received: ${event.data}');
},
onError: (Object e) => safePrint('Error in subscription stream: $e'),
);
}
void unsubscribe() {
subscription?.cancel();
subscription = null;
}

In addition to an onCreate subscription, you can also call .onUpdate() or .onDelete().

final onUpdateSubscriptionRequest = ModelSubscriptions.onUpdate(Todo.classType);
// or
final onDeleteSubscriptionRequest = ModelSubscriptions.onDelete(Todo.classType);

Subscription connection status

Now that you set up the application and are using subscriptions, you may want to know when the subscription is closed, or reflect to your users when the subscription isn’t healthy. You can monitor the subscription status for changes via Amplify.Hub

Amplify.Hub.listen(
HubChannel.Api,
(ApiHubEvent event) {
if (event is SubscriptionHubEvent) {
safePrint(event.status);
}
},
);

SubscriptionStatus

  • connected - Connected and working with no issues
  • connecting - Attempting to connect (both initial connection and reconnection)
  • pendingDisconnect - Connection has no active subscriptions and is shutting down
  • disconnected - Connection has no active subscriptions and is disconnected
  • failed - Connection had a failure and has been disconnected

Automated Reconnection

Under the hood, we will attempt to maintain a healthy web socket connection through network changes. For example, if a device’s connection changes from Wi-Fi to 5g network, the plugin will attempt to reconnect using the new network.

Likewise, when disconnected from the internet unexpectedly, the subscription will attempt to reconnect using an exponential retry/back off strategy. By default, we will make 8 recovery attempts over about 50 seconds. If we cannot make a successful connection, then the web socket will be closed. You can customize this strategy when configuring the API plugin through RetryOptions.

Future<void> _configureAmplify() async {
final apiPlugin = AmplifyAPI(
modelProvider: ModelProvider.instance,
// Optional config
subscriptionOptions: const GraphQLSubscriptionOptions(
retryOptions: RetryOptions(maxAttempts: 10),
),
);
await Amplify.addPlugin(apiPlugin);
try {
await Amplify.configure(amplifyconfig);
} on AmplifyAlreadyConfiguredException {
safePrint(
"Tried to reconfigure Amplify; this can occur when your app restarts on Android.");
}
}

Important: While offline, your application will miss messages and will not automatically catch up when reconnection happens. Depending on your use case, you may want to take action to catch up when your app comes back online. The following example solves this problem by retrieving all data on reconnection.

import 'package:amplify_flutter/amplify_flutter.dart';
import 'package:amplify_api/amplify_api.dart';
import './models/ModelProvider.dart'; // <--- Update import to reflect your project
import 'dart:async';
// ...
List<Todo?> allTodos = [];
SubscriptionStatus prevSubscriptionStatus = SubscriptionStatus.disconnected;
StreamSubscription<GraphQLResponse<Todo>>? subscription;
/// ...
// Init listeners
Amplify.Hub.listen(
HubChannel.Api,
(ApiHubEvent event) {
if (event is SubscriptionHubEvent) {
if (prevSubscriptionStatus == SubscriptionStatus.connecting &&
event.status == SubscriptionStatus.connected) {
getTodos(); // refetch todos
}
prevSubscriptionStatus = event.status;
}
},
);
subscribe();
/// ...
Future<void> getTodos() async {
try {
final request = ModelQueries.list(Todo.classType);
final response = await Amplify.API.query(request: request).response;
final todos = response.data?.items ?? [];
if (response.errors.isNotEmpty) {
safePrint('errors: ${response.errors}');
}
setState(() {
allTodos = todos;
});
} on ApiException catch (e) {
safePrint('Query failed: $e');
return;
}
}
void subscribe() {
final subscriptionRequest = ModelSubscriptions.onCreate(Todo.classType);
final Stream<GraphQLResponse<Todo>> operation = Amplify.API.subscribe(
subscriptionRequest,
onEstablished: () => safePrint('Subscription established'),
);
subscription = operation.listen(
(event) {
setState(() {
allTodos.add(event.data);
});
},
onError: (Object e) => safePrint('Error in subscription stream: $e'),
);
}