Subscribe to mutations for creating real-time clients.

## Setup subscription with callbacks

When creating subscriptions, a [`Stream`](https://api.dart.dev/stable/dart-async/Stream-class.html) 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:

```dart
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`](https://api.dart.dev/stable/dart-async/Stream/listen.html) to create a [`StreamSubscription`](https://api.dart.dev/stable/dart-async/StreamSubscription-class.html) object which can be programmatically canceled.

```dart
// 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()`.

```dart
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`

```dart
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`.

```dart
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.");
  }
}
```

<Callout>

**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.

</Callout>

```dart
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'),
  );
}

```
