Connect the websocket

Author: renefloor

packages/stream_feeds/lib/src/feeds_client.dart

@@ -1,9 +1,13 @@
-import 'package:stream_core/src/user/user.dart';
+import 'dart:async';
+
+import 'package:stream_core/stream_core.dart';
 import 'package:uuid/uuid.dart';
 
 import '../stream_feeds.dart';
 import 'generated/api/api.g.dart' as api;
 import 'repositories.dart';
+import 'utils/endpoint_config.dart';
+import 'ws/feeds_ws_event.dart';
 
 class FeedsClient {
   FeedsClient({
@@ -12,17 +16,33 @@ class FeedsClient {
     required this.userToken,
     this.config = const FeedsConfig(),
     this.userTokenProvider,
+    this.networkMonitor,
   }) {
     apiClient = api.DefaultApi(
       api.ApiClient(
+        basePath: endpointConfig.baseFeedsUrl,
         authentication: _Authentication(
           apiKey: apiKey,
           user: user,
           getToken: () async => userToken,
-          getConnectionId: () => null,
+          getConnectionId: () => webSocketClient?.connectionId,
         ),
       ),
     );
+    final websocketUri = Uri.parse(endpointConfig.wsEndpoint).replace(
+      queryParameters: <String, String>{
+        'api_key': apiKey,
+        'stream-auth-type': 'jwt',
+        'X-Stream-Client': 'stream-feeds-dart',
+      },
+    );
+
+    webSocketClient = WebSocketClient(
+      url: websocketUri.toString(),
+      eventDecoder: FeedsWsEvent.fromEventObject,
+      onConnectionEstablished: _authenticate,
+    );
+
     feedsRepository = FeedsRepository(apiClient: apiClient);
   }
 
@@ -31,10 +51,74 @@ class FeedsClient {
   final String userToken;
   final FeedsConfig config;
   final UserTokenProvider? userTokenProvider;
+  final NetworkMonitor? networkMonitor;
 
   late final api.DefaultApi apiClient;
   late final FeedsRepository feedsRepository;
 
+  static final endpointConfig = EndpointConfig.production;
+  late final WebSocketClient webSocketClient;
+  ConnectionRecoveryHandler? connectionRecoveryHandler;
+
+  Completer<void>? _connectionCompleter;
+  StreamSubscription<WebSocketConnectionState>? _connectionSubscription;
+
+  /// Connects to the feeds websocket.
+  /// Future will complete when the connection is established and the user is authenticated.
+  /// If the authentication fails, the future will complete with an error.
+  Future<void> connect() async {
+    webSocketClient.connect();
+
+    _connectionSubscription = webSocketClient!.connectionStateStream
+        .listen(_onConnectionStateChanged);
+
+    connectionRecoveryHandler = DefaultConnectionRecoveryHandler(
+      client: webSocketClient,
+      networkMonitor: networkMonitor,
+    );
+
+    _connectionCompleter = Completer<void>();
+    return _connectionCompleter!.future;
+  }
+
+  /// Disconnects from the feeds websocket.
+  /// The FeedsClient should no longer be used after calling this method.
+  void disconnect() {
+    connectionRecoveryHandler?.dispose();
+    webSocketClient.disconnect();
+    _connectionSubscription?.cancel();
+    _connectionCompleter?.complete();
+    _connectionCompleter = null;
+  }
+
+  void _onConnectionStateChanged(WebSocketConnectionState state) {
+    if (_connectionCompleter != null) {
+      if (state is Connected) {
+        _connectionCompleter!.complete();
+        _connectionCompleter = null;
+      }
+      if (state is Disconnected) {
+        _connectionCompleter!.completeError(Exception('Connection failed'));
+        _connectionCompleter = null;
+      }
+    }
+  }
+
+  void _authenticate() {
+    final connectUserRequest = WsAuthMessageRequest(
+      products: ['feeds'],
+      token: userToken,
+      userDetails: ConnectUserDetailsRequest(
+        id: user.id,
+        name: user.originalName,
+        image: user.imageUrl,
+        customData: user.customData,
+      ),
+    );
+
+    webSocketClient.send(connectUserRequest);
+  }
+
   /// Creates a feed instance based on the provided query.
   ///
   /// This method creates a [Feed] object using a [FeedQuery] that can include additional

packages/stream_feeds/lib/src/state_layer/feed_query.dart

@@ -9,7 +9,7 @@ class FeedQuery {
   FeedQuery({
     required String group,
     required String id,
-    this.watch = false,
+    this.watch = true,
   }) : fid = FeedId(group: group, id: id);
 
   /// The unique identifier for the feed.

packages/stream_feeds/lib/src/utils/endpoint_config.dart

@@ -0,0 +1,25 @@
+enum EndpointConfig {
+  localhost(
+    hostname: 'http://localhost:3030',
+    wsEndpoint: 'ws://localhost:8800/api/v2/connect',
+  ),
+  staging(
+    hostname: 'https://chat-edge-frankfurt-ce1.stream-io-api.com',
+    wsEndpoint:
+        'wss://chat-edge-frankfurt-ce1.stream-io-api.com/api/v2/connect',
+  ),
+  production(
+    hostname: 'https://feeds.stream-io-api.com',
+    wsEndpoint: 'wss://feeds.stream-io-api.com/api/v2/connect',
+  );
+
+  final String hostname;
+  final String wsEndpoint;
+  final String baseFeedsUrl;
+
+  const EndpointConfig({
+    required this.hostname,
+    required this.wsEndpoint,
+    String? baseFeedsUrl,
+  }) : baseFeedsUrl = baseFeedsUrl ?? hostname;
+}

packages/stream_feeds/lib/src/ws/feeds_ws_event.dart

@@ -0,0 +1,44 @@
+// ignore_for_file: prefer_single_quotes
+
+import 'dart:convert';
+
+import 'package:stream_core/stream_core.dart' as core;
+
+import '../generated/api/api.g.dart' as api;
+
+class FeedsWsEvent extends core.WsEvent {
+  const FeedsWsEvent(this.event);
+
+  final api.WSClientEvent? event;
+
+  static core.WsEvent fromEventObject(Object message) {
+    try {
+      final json = jsonDecode(message.toString()) as Map<String, dynamic>;
+      final type = json['type'];
+      switch (type) {
+        case "connection.ok":
+          return core.HealthCheckPongEvent(
+            healthCheckInfo: core.HealthCheckInfo(
+              connectionId: json['connection_id'],
+            ),
+          );
+        case 'connection.error':
+          return core.WsErrorEvent(error: json['error'], message: message);
+        default:
+          final event = api.WSClientEvent.fromJson(json);
+
+          if (event is api.WSClientEventHealthCheckEvent) {
+            return core.HealthCheckPongEvent(
+              healthCheckInfo: core.HealthCheckInfo(
+                connectionId: event.healthCheckEvent.connectionId,
+              ),
+            );
+          }
+          print(type);
+          return FeedsWsEvent(event);
+      }
+    } catch (e) {
+      return core.WsErrorEvent(error: e, message: message);
+    }
+  }
+}

sample_app/lib/navigation/app_state.dart

@@ -24,25 +24,32 @@ class AppStateProvider extends ValueNotifier<AppState> {
     final userId = _prefs.getString('user_id');
     if (userId != null) {
       final credentials = UserCredentials.credentialsFor(userId);
-      setUser(credentials);
+      await setUser(credentials);
     } else {
       value = LoggedOutState();
     }
   }
 
-  void setUser(UserCredentials userCredentials) {
-    _prefs.setString('user_id', userCredentials.user.id);
+  Future<void> setUser(UserCredentials userCredentials) async {
+    await _prefs.setString('user_id', userCredentials.user.id);
+    value = LoadingState();
+    final client = FeedsClient(
+      apiKey: DemoAppConfig.current.apiKey,
+      user: userCredentials.user,
+      userToken: userCredentials.token,
+    );
+    await client.connect();
+
     value = LoggedInState(
-      feedsClient: FeedsClient(
-        apiKey: DemoAppConfig.current.apiKey,
-        user: userCredentials.user,
-        userToken: userCredentials.token,
-      ),
+      feedsClient: client,
     );
   }
 
   void clearUserId() {
     _prefs.remove('user_id');
+    if (value is LoggedInState) {
+      (value as LoggedInState).feedsClient.disconnect();
+    }
     value = LoggedOutState();
   }
 }
@@ -53,6 +60,8 @@ class InitialState extends AppState {}
 
 class LoggedOutState extends AppState {}
 
+class LoadingState extends AppState {}
+
 class LoggedInState extends AppState {
   LoggedInState({required this.feedsClient});