fix(api): Reconnect WebSocket when resuming app from a paused state (#…

…5567)

* fix(api): Reconnect WebSocket when resuming app from a paused state

packages/api/amplify_api/lib/amplify_api.dart

@@ -5,4 +5,8 @@ library amplify_api;
 
 export 'package:amplify_api/src/api_plugin_impl.dart';
 export 'package:amplify_api_dart/amplify_api_dart.dart'
- hide AmplifyAPIDart, ConnectivityPlatform, ConnectivityStatus;
+ hide
+ AmplifyAPIDart,
+ ConnectivityPlatform,
+ ProcessLifeCycle,
+ ConnectivityStatus;

packages/api/amplify_api/lib/src/api_plugin_impl.dart

@@ -2,6 +2,7 @@
 // SPDX-License-Identifier: Apache-2.0
 
 import 'package:amplify_api/src/connectivity_plus_platform.dart';
+import 'package:amplify_api/src/flutter_life_cycle.dart';
 import 'package:amplify_api_dart/amplify_api_dart.dart';
 import 'package:amplify_core/amplify_core.dart';
 
@@ -14,6 +15,7 @@ class AmplifyAPI extends AmplifyAPIDart with AWSDebuggable {
  super.options,
  }) : super(
  connectivity: const ConnectivityPlusPlatform(),
+ processLifeCycle: FlutterLifeCycle(),
  );
 
  @override

packages/api/amplify_api/lib/src/flutter_life_cycle.dart

@@ -0,0 +1,42 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+import 'dart:async';
+
+import 'package:amplify_api_dart/amplify_api_dart.dart';
+import 'package:flutter/widgets.dart';
+import 'package:meta/meta.dart';
+
+/// {@template amplify_api.flutter_life_cycle}
+/// Creates a stream of [ProcessStatus] mapped from [AppLifecycleListener](https://api.flutter.dev/flutter/widgets/AppLifecycleListener-class.html).
+/// {@endtemplate}
+@internal
+class FlutterLifeCycle extends ProcessLifeCycle {
+ /// {@macro amplify_api.flutter_life_cycle}
+ FlutterLifeCycle() {
+ AppLifecycleListener(
+ onStateChange: _onStateChange,
+ );
+ }
+
+ final _stateController =
+ StreamController<ProcessStatus>.broadcast(sync: true);
+
+ @override
+ Stream<ProcessStatus> get onStateChanged => _stateController.stream;
+
+ void _onStateChange(AppLifecycleState state) {
+ switch (state) {
+ case AppLifecycleState.detached:
+ _stateController.add(ProcessStatus.detached);
+ case AppLifecycleState.paused:
+ _stateController.add(ProcessStatus.paused);
+ case AppLifecycleState.hidden:
+ _stateController.add(ProcessStatus.hidden);
+ case AppLifecycleState.inactive:
+ _stateController.add(ProcessStatus.inactive);
+ case AppLifecycleState.resumed:
+ _stateController.add(ProcessStatus.resumed);
+ }
+ }
+}

packages/api/amplify_api_dart/lib/amplify_api_dart.dart

@@ -19,3 +19,4 @@ export 'src/graphql/model_helpers/model_subscriptions.dart';
 
 /// Network connectivity util not needed by consumers of Flutter package amplify_api.
 export 'src/graphql/web_socket/types/connectivity_platform.dart';
+export 'src/graphql/web_socket/types/process_life_cycle.dart';

packages/api/amplify_api_dart/lib/src/api_plugin_impl.dart

@@ -11,6 +11,7 @@ import 'package:amplify_api_dart/src/graphql/web_socket/blocs/web_socket_bloc.da
 import 'package:amplify_api_dart/src/graphql/web_socket/services/web_socket_service.dart';
 import 'package:amplify_api_dart/src/graphql/web_socket/state/web_socket_state.dart';
 import 'package:amplify_api_dart/src/graphql/web_socket/types/connectivity_platform.dart';
+import 'package:amplify_api_dart/src/graphql/web_socket/types/process_life_cycle.dart';
 import 'package:amplify_api_dart/src/util/amplify_api_config.dart';
 import 'package:amplify_api_dart/src/util/amplify_authorization_rest_client.dart';
 import 'package:amplify_core/amplify_core.dart';
@@ -30,8 +31,10 @@ class AmplifyAPIDart extends APIPluginInterface with AWSDebuggable {
  AmplifyAPIDart({
  APIPluginOptions options = const APIPluginOptions(),
  ConnectivityPlatform connectivity = const ConnectivityPlatform(),
+ ProcessLifeCycle processLifeCycle = const ProcessLifeCycle(),
  }) : _options = options,
- _connectivity = connectivity {
+ _connectivity = connectivity,
+ _processLifeCycle = processLifeCycle {
  _options.authProviders.forEach(registerAuthProvider);
  }
 
@@ -43,6 +46,9 @@ class AmplifyAPIDart extends APIPluginInterface with AWSDebuggable {
  /// Creates a stream representing network connectivity at the hardware level.
  final ConnectivityPlatform _connectivity;
 
+ /// Creates a stream representing the process life cycle state.
+ final ProcessLifeCycle _processLifeCycle;
+
  /// A map of the keys from the Amplify API config with auth modes to HTTP clients
  /// to use for requests to that endpoint/auth mode. e.g. { "myEndpoint.AWS_IAM": AWSHttpClient}
  final Map<String, AWSHttpClient> _clientPool = {};
@@ -277,6 +283,7 @@ class AmplifyAPIDart extends APIPluginInterface with AWSDebuggable {
  wsService: AmplifyWebSocketService(),
  subscriptionOptions: _options.subscriptionOptions,
  connectivity: _connectivity,
+ processLifeCycle: _processLifeCycle,
  );
  }
 

packages/api/amplify_api_dart/lib/src/graphql/web_socket/blocs/web_socket_bloc.dart

@@ -9,6 +9,7 @@ import 'package:amplify_api_dart/src/graphql/web_socket/services/web_socket_serv
 import 'package:amplify_api_dart/src/graphql/web_socket/state/web_socket_state.dart';
 import 'package:amplify_api_dart/src/graphql/web_socket/state/ws_subscriptions_state.dart';
 import 'package:amplify_api_dart/src/graphql/web_socket/types/connectivity_platform.dart';
+import 'package:amplify_api_dart/src/graphql/web_socket/types/process_life_cycle.dart';
 import 'package:amplify_api_dart/src/graphql/web_socket/types/subscriptions_event.dart';
 import 'package:amplify_api_dart/src/graphql/web_socket/types/web_socket_types.dart';
 import 'package:amplify_core/amplify_core.dart' hide SubscriptionEvent;
@@ -33,8 +34,10 @@ class WebSocketBloc with AWSDebuggable, AmplifyLoggerMixin {
  required WebSocketService wsService,
  required GraphQLSubscriptionOptions subscriptionOptions,
  required ConnectivityPlatform connectivity,
+ required ProcessLifeCycle processLifeCycle,
  AWSHttpClient? pollClientOverride,
  }) : _connectivity = connectivity,
+ _processLifeCycle = processLifeCycle,
  _pollClient = pollClientOverride ?? AWSHttpClient() {
  final subBlocs = <String, SubscriptionBloc<Object?>>{};
 
@@ -49,6 +52,7 @@ class WebSocketBloc with AWSDebuggable, AmplifyLoggerMixin {
  );
  final blocStream = _wsEventStream.asyncExpand(_eventTransformer);
  _networkSubscription = _getConnectivityStream();
+ _processLifeCycleSubscription = _getProcessLifecycleStream();
  _stateSubscription = blocStream.listen(_emit);
  add(const InitEvent());
  }
@@ -81,10 +85,14 @@ class WebSocketBloc with AWSDebuggable, AmplifyLoggerMixin {
  late final Stream<WebSocketEvent> _wsEventStream = _wsEventController.stream;
  late final StreamSubscription<WebSocketState> _stateSubscription;
  late final StreamSubscription<ConnectivityStatus> _networkSubscription;
+ late final StreamSubscription<ProcessStatus> _processLifeCycleSubscription;
 
  /// Creates a stream representing network connectivity at the hardware level.
  final ConnectivityPlatform _connectivity;
 
+ /// Creates a stream representing the process life cycle state.
+ final ProcessLifeCycle _processLifeCycle;
+
  /// The underlying event stream, used only in testing.
  @visibleForTesting
  Stream<WebSocketEvent> get wsEventStream => _wsEventStream;
@@ -164,6 +172,8 @@ class WebSocketBloc with AWSDebuggable, AmplifyLoggerMixin {
  yield* _networkLoss();
  } else if (event is NetworkFoundEvent) {
  yield* _networkFound();
+ } else if (event is ProcessResumeEvent) {
+ yield* _processResumed();
  } else if (event is PollSuccessEvent) {
  yield* _pollSuccess();
  } else if (event is PollFailedEvent) {
@@ -328,6 +338,16 @@ class WebSocketBloc with AWSDebuggable, AmplifyLoggerMixin {
  yield* const Stream.empty();
  }
 
+ Stream<WebSocketState> _processResumed() async* {
+ final state = _currentState;
+ if (state is ConnectedState) {
+ yield state.reconnecting(networkState: NetworkState.disconnected);
+ add(const ReconnectEvent());
+ }
+ // TODO(dnys1): Yield broken on web debug build.
+ yield* const Stream.empty();
+ }
+
  /// Handle successful polls
  Stream<WebSocketState> _pollSuccess() async* {
  // TODO(dnys1): Yield broken on web debug build.
@@ -467,6 +487,7 @@ class WebSocketBloc with AWSDebuggable, AmplifyLoggerMixin {
  await Future.wait<void>([
  // TODO(equartey): https:/fluttercommunity/plus_plugins/issues/1382
  if (!isWindows()) _networkSubscription.cancel(),
+ _processLifeCycleSubscription.cancel(),
  Future.value(_pollClient.close()),
  _stateSubscription.cancel(),
  _wsEventController.close(),
@@ -507,6 +528,41 @@ class WebSocketBloc with AWSDebuggable, AmplifyLoggerMixin {
  );
  }
 
+ /// Process life cycle stream monitors when the process resumes from a paused state.
+ StreamSubscription<ProcessStatus> _getProcessLifecycleStream() {
+ var prev = ProcessStatus.detached;
+ return _processLifeCycle.onStateChanged.listen(
+ (state) {
+ if (_isResuming(state, prev)) {
+ // ignore: invalid_use_of_internal_member
+ if (!WebSocketOptions.autoReconnect) {
+ _shutdownWithException(
+ const NetworkException(
+ 'Unable to recover network connection, web socket will close.',
+ recoverySuggestion: 'Avoid pausing the process.',
+ ),
+ StackTrace.current,
+ );
+ } else {
+ add(const ProcessResumeEvent());
+ }
+ }
+
+ prev = state;
+ },
+ onError: (Object e, StackTrace st) =>
+ logger.error('Error in process life cycle stream $e, $st'),
+ );
+ }
+
+ bool _isResuming(ProcessStatus current, ProcessStatus previous) {
+ if (previous != ProcessStatus.paused) return false;
+
+ return current == ProcessStatus.hidden ||
+ current == ProcessStatus.inactive ||
+ current == ProcessStatus.resumed;
+ }
+
  Future<void> _poll() async {
  try {
  final res = await _sendPollRequest();

packages/api/amplify_api_dart/lib/src/graphql/web_socket/types/process_life_cycle.dart

@@ -0,0 +1,33 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+/// Possible process life cycle states
+enum ProcessStatus {
+ /// Engine is running without a view.
+ detached,
+
+ /// Application is not visible to the user or responding to user input.
+ paused,
+
+ /// All views of an application are hidden.
+ hidden,
+
+ /// A view of the application is visible, but none have input.
+ inactive,
+
+ /// Default running mode.
+ resumed,
+}
+
+/// {@template amplify_api_dart.process_life_cycle}
+/// Used to create a stream representing the process life cycle state.
+///
+/// The generated stream is empty.
+/// {@endtemplate}
+class ProcessLifeCycle {
+ /// {@macro amplify_api_dart.process_life_cycle}
+ const ProcessLifeCycle();
+
+ /// Generates a new stream of [ProcessStatus].
+ Stream<ProcessStatus> get onStateChanged => const Stream.empty();
+}

packages/api/amplify_api_dart/lib/src/graphql/web_socket/types/web_socket_event.dart

@@ -88,6 +88,19 @@ class NetworkLossEvent extends NetworkEvent {
  String get runtimeTypeName => 'NetworkLossEvent';
 }
 
+/// Discrete class for when the process is resumed
+/// Triggers when AppLifecycleListener detects the process has been resumed.
+class ProcessResumeEvent extends WebSocketEvent {
+ /// Create a process resumed event
+ const ProcessResumeEvent();
+
+ @override
+ String get runtimeTypeName => 'ProcessResumeEvent';
+
+ @override
+ Map<String, Object?> toJson() => const {};
+}
+
 /// Triggers when a successful ping to AppSync is made
 class PollSuccessEvent extends WebSocketEvent {
  /// Create a successful Poll event

packages/api/amplify_api_dart/test/graphql_test.dart

@@ -280,6 +280,7 @@ void main() {
  'payload': {'data': mockSubscriptionData},
  };
 
+ mockProcessLifeCycleController = StreamController<ProcessStatus>();
  mockWebSocketService = MockWebSocketService();
  const subscriptionOptions = GraphQLSubscriptionOptions(
  pollInterval: Duration(seconds: 1),
@@ -292,6 +293,7 @@ void main() {
  subscriptionOptions: subscriptionOptions,
  pollClientOverride: mockClient.client,
  connectivity: const ConnectivityPlatform(),
+ processLifeCycle: const MockProcessLifeCycle(),
  );
 
  sendMockConnectionAck(mockWebSocketBloc!, mockWebSocketService!);
@@ -599,6 +601,7 @@ void main() {
  });
 
  test('should have correct state flow during a failure', () async {
+ mockProcessLifeCycleController = StreamController<ProcessStatus>();
  mockWebSocketService = MockWebSocketService();
  const subscriptionOptions = GraphQLSubscriptionOptions(
  pollInterval: Duration(seconds: 1),
@@ -613,6 +616,7 @@ void main() {
  subscriptionOptions: subscriptionOptions,
  pollClientOverride: mockClient.client,
  connectivity: const ConnectivityPlatform(),
+ processLifeCycle: const MockProcessLifeCycle(),
  );
 
  final blocReady = Completer<void>();

packages/api/amplify_api_dart/test/util.dart

@@ -10,6 +10,7 @@ import 'package:amplify_api_dart/src/graphql/web_socket/blocs/web_socket_bloc.da
 import 'package:amplify_api_dart/src/graphql/web_socket/services/web_socket_service.dart';
 import 'package:amplify_api_dart/src/graphql/web_socket/state/web_socket_state.dart';
 import 'package:amplify_api_dart/src/graphql/web_socket/types/connectivity_platform.dart';
+import 'package:amplify_api_dart/src/graphql/web_socket/types/process_life_cycle.dart';
 import 'package:amplify_api_dart/src/graphql/web_socket/types/web_socket_types.dart';
 import 'package:amplify_core/amplify_core.dart';
 import 'package:amplify_core/src/config/amplify_outputs/data/data_outputs.dart';
@@ -329,6 +330,16 @@ class MockConnectivity extends ConnectivityPlatform {
  mockNetworkStreamController.stream;
 }
 
+late StreamController<ProcessStatus> mockProcessLifeCycleController;
+
+class MockProcessLifeCycle extends ProcessLifeCycle {
+ const MockProcessLifeCycle();
+
+ @override
+ Stream<ProcessStatus> get onStateChanged =>
+ mockProcessLifeCycleController.stream;
+}
+
 /// Ensures a query predicate converts to JSON correctly.
 void testQueryPredicateTranslation(
  QueryPredicate? queryPredicate,