refactor(llc): introduce event controller, resolver (#2301)

Author: xsahil03x

packages/stream_chat/lib/src/client/client.dart

@@ -5,6 +5,7 @@ import 'package:logging/logging.dart';
 import 'package:meta/meta.dart';
 import 'package:rxdart/rxdart.dart';
 import 'package:stream_chat/src/client/channel.dart';
+import 'package:stream_chat/src/client/event_resolvers.dart' as event_resolvers;
 import 'package:stream_chat/src/client/retry_policy.dart';
 import 'package:stream_chat/src/core/api/attachment_file_uploader.dart';
 import 'package:stream_chat/src/core/api/requests.dart';
@@ -33,6 +34,7 @@ import 'package:stream_chat/src/core/models/poll_option.dart';
 import 'package:stream_chat/src/core/models/poll_vote.dart';
 import 'package:stream_chat/src/core/models/thread.dart';
 import 'package:stream_chat/src/core/models/user.dart';
+import 'package:stream_chat/src/core/util/event_controller.dart';
 import 'package:stream_chat/src/core/util/utils.dart';
 import 'package:stream_chat/src/db/chat_persistence_client.dart';
 import 'package:stream_chat/src/event_type.dart';
@@ -222,21 +224,15 @@ class StreamChatClient {
 
   StreamSubscription<ConnectionStatus>? _connectionStatusSubscription;
 
-  final _eventController = PublishSubject<Event>();
-
   /// Stream of [Event] coming from [_ws] connection
   /// Listen to this or use the [on] method to filter specific event types
-  Stream<Event> get eventStream => _eventController.stream.map(
-        // If the poll vote is an answer, we should emit a different event
-        // to make it easier to handle in the state.
-        (event) => switch ((event.type, event.pollVote?.isAnswer == true)) {
-          (EventType.pollVoteCasted || EventType.pollVoteChanged, true) =>
-            event.copyWith(type: EventType.pollAnswerCasted),
-          (EventType.pollVoteRemoved, true) =>
-            event.copyWith(type: EventType.pollAnswerRemoved),
-          _ => event,
-        },
-      );
+  Stream<Event> get eventStream => _eventController.stream;
+  late final _eventController = EventController<Event>(
+    resolvers: [
+      event_resolvers.pollAnswerCastedResolver,
+      event_resolvers.pollAnswerRemovedResolver,
+    ],
+  );
 
   final _wsConnectionStatusController =
       BehaviorSubject.seeded(ConnectionStatus.disconnected);

packages/stream_chat/lib/src/client/event_resolvers.dart

@@ -0,0 +1,37 @@
+import 'package:stream_chat/src/core/models/event.dart';
+import 'package:stream_chat/src/event_type.dart';
+
+/// Resolves casted or changed poll vote events into more specific
+/// `pollAnswerCasted` events for easier downstream state handling.
+///
+/// Applies when:
+/// - `event.type` is `pollVoteCasted` or `pollVoteChanged`, and
+/// - `event.pollVote?.isAnswer == true`
+///
+/// Returns a modified event with type `pollAnswerCasted`,
+/// or `null` if not applicable.
+Event? pollAnswerCastedResolver(Event event) {
+  return switch (event.type) {
+    EventType.pollVoteCasted ||
+    EventType.pollVoteChanged when event.pollVote?.isAnswer == true =>
+      event.copyWith(type: EventType.pollAnswerCasted),
+    _ => null,
+  };
+}
+
+/// Resolves removed poll vote events into more specific
+/// `pollAnswerRemoved` events for easier downstream state handling.
+///
+/// Applies when:
+/// - `event.type` is `pollVoteRemoved`, and
+/// - `event.pollVote?.isAnswer == true`
+///
+/// Returns a modified event with type `pollAnswerRemoved`,
+/// or `null` if not applicable.
+Event? pollAnswerRemovedResolver(Event event) {
+  return switch (event.type) {
+    EventType.pollVoteRemoved when event.pollVote?.isAnswer == true =>
+      event.copyWith(type: EventType.pollAnswerRemoved),
+    _ => null,
+  };
+}

packages/stream_chat/lib/src/core/util/event_controller.dart

@@ -0,0 +1,77 @@
+import 'dart:async';
+import 'package:rxdart/rxdart.dart';
+import 'package:stream_chat/src/core/models/event.dart';
+
+/// A function that inspects an event and optionally resolves it into a
+/// more specific or refined version of the same type.
+///
+/// If the resolver does not recognize or handle the event,
+/// it returns `null`, allowing other resolvers to attempt resolution.
+typedef EventResolver<T extends Event> = T? Function(T event);
+
+/// {@template eventController}
+/// A reactive event stream controller for [Event]s that supports conditional
+/// resolution before emitting events to subscribers.
+///
+/// When an event is added:
+/// - Each resolver is evaluated in order.
+/// - The first resolver that returns a non-null result is used to produce
+///   the resolved event that gets emitted.
+/// - If no resolver returns a result, the original event is emitted unchanged.
+///
+/// This is useful for normalizing or refining generic events into more
+/// specific ones (e.g. rewriting `pollVoteCasted` into `pollAnswerCasted`)
+/// before they reach business logic or state layers.
+/// {@endtemplate}
+class EventController<T extends Event> extends Subject<T> {
+  /// {@macro eventController}
+  factory EventController({
+    bool sync = false,
+    void Function()? onListen,
+    void Function()? onCancel,
+    List<EventResolver<T>> resolvers = const [],
+  }) {
+    // ignore: close_sinks
+    final controller = StreamController<T>.broadcast(
+      sync: sync,
+      onListen: onListen,
+      onCancel: onCancel,
+    );
+
+    return EventController<T>._(
+      controller,
+      controller.stream,
+      resolvers,
+    );
+  }
+
+  EventController._(
+    super.controller,
+    super.stream,
+    this._resolvers,
+  );
+
+  /// The list of resolvers used to inspect and optionally resolve events
+  /// before they are emitted.
+  ///
+  /// Resolvers are evaluated in order. The first to return a non-null result
+  /// determines the event that will be emitted. If none apply, the original
+  /// event is emitted as-is.
+  final List<EventResolver<T>> _resolvers;
+
+  /// Adds an [event] to the stream.
+  ///
+  /// Each [EventResolver] is applied in order until one returns a non-null
+  /// result. That resolved event is emitted, and no further resolvers are
+  /// evaluated. If all resolvers return `null`, the original event is emitted.
+  @override
+  void add(T event) {
+    for (final resolver in _resolvers) {
+      final result = resolver(event);
+      if (result != null) return super.add(result);
+    }
+
+    // No resolver matched — emit the event as-is.
+    return super.add(event);
+  }
+}

packages/stream_chat/test/src/mocks.dart

@@ -1,7 +1,6 @@
 import 'package:dio/dio.dart';
 import 'package:logging/logging.dart';
 import 'package:mocktail/mocktail.dart';
-import 'package:rxdart/rxdart.dart';
 import 'package:stream_chat/src/client/channel.dart';
 import 'package:stream_chat/src/client/client.dart';
 import 'package:stream_chat/src/core/api/attachment_file_uploader.dart';
@@ -18,6 +17,7 @@ import 'package:stream_chat/src/core/http/stream_http_client.dart';
 import 'package:stream_chat/src/core/http/token_manager.dart';
 import 'package:stream_chat/src/core/models/channel_config.dart';
 import 'package:stream_chat/src/core/models/event.dart';
+import 'package:stream_chat/src/core/util/event_controller.dart';
 import 'package:stream_chat/src/db/chat_persistence_client.dart';
 import 'package:stream_chat/src/event_type.dart';
 import 'package:stream_chat/src/ws/websocket.dart';
@@ -98,7 +98,7 @@ class MockStreamChatClient extends Mock implements StreamChatClient {
 
   @override
   Stream<Event> get eventStream => _eventController.stream;
-  final _eventController = PublishSubject<Event>();
+  final _eventController = EventController<Event>();
   void addEvent(Event event) => _eventController.add(event);
 
   @override