feat: add ability to emit/watch incoming and outgoing payload (#328)

Author: sagar-deriv

lib/services/connection/call_manager/call_history.dart

@@ -1,7 +1,17 @@
+import 'dart:async';
+
 import 'package:flutter_deriv_api/services/connection/call_manager/call_history_entry.dart';
+import 'package:flutter_deriv_api/services/interfaces/call_history_provider.dart';
 
 /// Provides storage for messages sent/received via the web socket connection
-class CallHistory {
+class CallHistory implements CallHistoryProvider {
+  /// It initializes [CallHistory] instance.
+  CallHistory() {
+    _callHistoryBroadcaster = StreamController<NetworkPayload>.broadcast();
+  }
+
+  late final StreamController<NetworkPayload> _callHistoryBroadcaster;
+
   /// Messages that were sent to the remote endpoint
   final List<CallHistoryEntry> outgoing = <CallHistoryEntry>[];
 
@@ -29,7 +39,15 @@ class CallHistory {
     incoming.add(
       CallHistoryEntry(timeStamp: timestamp, method: method, message: message),
     );
-
+    if (!method.contains('ping')) {
+      _callHistoryBroadcaster.add(
+        NetworkPayload(
+            method: method,
+            body: message,
+            direction: NetworkDirections.received,
+            timeStamp: timestamp),
+      );
+    }
     _trimHistory(incoming);
   }
 
@@ -42,6 +60,15 @@ class CallHistory {
     outgoing.add(
       CallHistoryEntry(timeStamp: timestamp, method: method, message: message),
     );
+    if (!method.contains('ping')) {
+      _callHistoryBroadcaster.add(
+        NetworkPayload(
+            method: method,
+            body: message,
+            direction: NetworkDirections.sent,
+            timeStamp: timestamp),
+      );
+    }
 
     _trimHistory(outgoing);
   }
@@ -52,4 +79,7 @@ class CallHistory {
       callHistory.removeRange(0, callHistory.length - limit);
     }
   }
+
+  @override
+  Stream<NetworkPayload> get stream => _callHistoryBroadcaster.stream;
 }

lib/services/interfaces/call_history_provider.dart

@@ -0,0 +1,37 @@
+/// This interface provides stream of network payload that is going out(SENT)
+///  and coming in(RECEIVED) to the application.
+abstract class CallHistoryProvider {
+  /// Stream of network payload that is going out(SENT) and coming in(RECEIVED)
+  Stream<NetworkPayload> get stream;
+}
+
+/// Network payload that is going out and coming in from the web socket.
+class NetworkPayload {
+  /// Initializes [NetworkPayload] instance.
+  NetworkPayload({
+    required this.method,
+    required this.body,
+    required this.direction,
+    required this.timeStamp,
+  });
+
+  /// name of the api.
+  final String method;
+
+  /// content of the api.
+  final Object body;
+
+  /// direction of the api i.e SENT or RECEIVED.
+  final NetworkDirections direction;
+
+  /// time of the api.
+  final int timeStamp;
+}
+
+enum NetworkDirections {
+  /// Going out from the web socket.
+  sent,
+
+  /// Coming in from the web socket.
+  received
+}

lib/state/connection/connection_cubit.dart

@@ -46,7 +46,7 @@ class ConnectionCubit extends Cubit<ConnectionState> {
 
   final String _key = '${UniqueKey()}';
 
-  late final BaseAPI? _api;
+  late final BaseAPI _api;
 
   /// Enables debug mode.
   ///
@@ -86,6 +86,9 @@ class ConnectionCubit extends Cubit<ConnectionState> {
   /// Stream subscription for connectivity.
   StreamSubscription<ConnectivityResult>? connectivitySubscription;
 
+  /// Getter for [BaseAPI] implementation class. By default, it will be [BinaryAPI].
+  BaseAPI get api => _api;
+
   /// Reconnect to Websocket.
   Future<void> reconnect({
     ConnectionInformation? connectionInformation,
@@ -109,7 +112,7 @@ class ConnectionCubit extends Cubit<ConnectionState> {
     emit(const ConnectionConnectingState());
 
     try {
-      await _api!.disconnect().timeout(_pingTimeout);
+      await _api.disconnect().timeout(_pingTimeout);
     } on Exception catch (e) {
       dev.log('$runtimeType disconnect exception: $e', error: e);
 
@@ -118,7 +121,7 @@ class ConnectionCubit extends Cubit<ConnectionState> {
       return;
     }
 
-    await _api!.connect(
+    await _api.connect(
       _connectionInformation,
       printResponse: enableDebug && printResponse,
       onOpen: (String key) {