onPush

lib/src/client/centrifuge.dart

@@ -5,11 +5,14 @@ import 'package:centrifuge_dart/src/client/config.dart';
 import 'package:centrifuge_dart/src/client/disconnect_code.dart';
 import 'package:centrifuge_dart/src/client/state.dart';
 import 'package:centrifuge_dart/src/client/states_stream.dart';
+import 'package:centrifuge_dart/src/model/channel_presence.dart';
+import 'package:centrifuge_dart/src/model/channel_push.dart';
 import 'package:centrifuge_dart/src/model/event.dart';
-import 'package:centrifuge_dart/src/model/event_stream.dart';
 import 'package:centrifuge_dart/src/model/exception.dart';
 import 'package:centrifuge_dart/src/model/presence.dart';
 import 'package:centrifuge_dart/src/model/presence_stats.dart';
+import 'package:centrifuge_dart/src/model/publication.dart';
+import 'package:centrifuge_dart/src/model/pushes_stream.dart';
 import 'package:centrifuge_dart/src/subscription/client_subscription_manager.dart';
 import 'package:centrifuge_dart/src/subscription/subscription.dart';
 import 'package:centrifuge_dart/src/subscription/subscription_config.dart';
@@ -25,6 +28,7 @@ import 'package:stack_trace/stack_trace.dart' as st;
 /// {@endtemplate}
 final class Centrifuge extends CentrifugeBase
     with
+        CentrifugeEventReceiverMixin,
         CentrifugeErrorsMixin,
         CentrifugeStateMixin,
         CentrifugeConnectionMixin,
@@ -66,24 +70,17 @@ abstract base class CentrifugeBase implements ICentrifuge {
   @nonVirtual
   final CentrifugeConfig _config;
 
-  @protected
-  @nonVirtual
-  final StreamController<CentrifugeEvent> _eventsController =
-      StreamController<CentrifugeEvent>.broadcast();
-
-  @override
-  @nonVirtual
-  late final CentrifugeEventStream events =
-      CentrifugeEventStream(_eventsController.stream);
+  /// Manager responsible for client-side subscriptions.
+  /// {@nodoc}
+  late final ClientSubscriptionManager _clientSubscriptionManager =
+      ClientSubscriptionManager(_transport);
 
   /// Init centrifuge client, override this method to add custom logic.
   /// This method is called in constructor.
   /// {@nodoc}
   @protected
   @mustCallSuper
-  void _initCentrifuge() {
-    _transport.events.addListener(_onEvent);
-  }
+  void _initCentrifuge() {}
 
   /// Called when connection established.
   /// Right before [CentrifugeState$Connected] state.
@@ -103,25 +100,92 @@ abstract base class CentrifugeBase implements ICentrifuge {
     logger.fine('Connection lost');
   }
 
+  @override
+  @mustCallSuper
+  Future<void> close() async {}
+}
+
+/// Mixin responsible for event receiving and distribution by controllers
+/// and streams to subscribers.
+/// {@nodoc}
+base mixin CentrifugeEventReceiverMixin on CentrifugeBase {
+  @protected
+  @nonVirtual
+  final StreamController<CentrifugeChannelPush> _pushController =
+      StreamController<CentrifugeChannelPush>.broadcast();
+
+  @protected
+  @nonVirtual
+  final StreamController<CentrifugePublication> _publicationsController =
+      StreamController<CentrifugePublication>.broadcast();
+
+  @protected
+  @nonVirtual
+  final StreamController<CentrifugeJoin> _joinController =
+      StreamController<CentrifugeJoin>.broadcast();
+
+  @protected
+  @nonVirtual
+  final StreamController<CentrifugeLeave> _leaveController =
+      StreamController<CentrifugeLeave>.broadcast();
+
+  @protected
+  @nonVirtual
+  final StreamController<CentrifugeChannelPresence> _presenceController =
+      StreamController<CentrifugeChannelPresence>.broadcast();
+
+  @override
+  @nonVirtual
+  late final CentrifugePushesStream stream = CentrifugePushesStream(
+    pushes: _pushController.stream,
+    publications: _publicationsController.stream,
+    presenceEvents: _presenceController.stream,
+    joinEvents: _joinController.stream,
+    leaveEvents: _leaveController.stream,
+  );
+
+  @override
+  void _initCentrifuge() {
+    _transport.events.addListener(_onEvent);
+    super._initCentrifuge();
+  }
+
   /// Router for all events.
   /// {@nodoc}
   @protected
-  @mustCallSuper
+  @nonVirtual
+  @pragma('vm:prefer-inline')
+  @pragma('dart2js:tryInline')
   void _onEvent(CentrifugeEvent event) {
-    /* switch (event) {
+    if (event is! CentrifugeChannelPush) return;
+    // This is a push to a channel.
+    _clientSubscriptionManager.onPush(event);
+    _pushController.add(event);
+    switch (event) {
       case CentrifugePublication publication:
-        _onPublication(publication);
-      case CentrifugeChannelPresenceEvent event:
-        _onPresenceEvent(event);
-    } */
-    _eventsController.add(event);
+        _publicationsController.add(publication);
+      case CentrifugeJoin join:
+        _presenceController.add(join);
+        _joinController.add(join);
+      case CentrifugeLeave leave:
+        _presenceController.add(leave);
+        _leaveController.add(leave);
+    }
   }
 
   @override
-  @mustCallSuper
   Future<void> close() async {
+    await super.close();
     _transport.events.removeListener(_onEvent);
-    _eventsController.close().ignore();
+    for (final controller in <StreamSink<CentrifugeEvent>>[
+      _pushController,
+      _publicationsController,
+      _joinController,
+      _leaveController,
+      _presenceController,
+    ]) {
+      controller.close().ignore();
+    }
   }
 }
 
@@ -373,9 +437,6 @@ base mixin CentrifugeSendMixin on CentrifugeBase, CentrifugeErrorsMixin {
 @internal
 base mixin CentrifugeClientSubscriptionMixin
     on CentrifugeBase, CentrifugeErrorsMixin {
-  late final ClientSubscriptionManager _clientSubscriptionManager =
-      ClientSubscriptionManager(_transport);
-
   @override
   CentrifugeClientSubscription newSubscription(
     String channel, [

lib/src/client/centrifuge_interface.dart

@@ -3,9 +3,9 @@
 import 'dart:async';
 
 import 'package:centrifuge_dart/centrifuge.dart';
-import 'package:centrifuge_dart/src/model/event_stream.dart';
 import 'package:centrifuge_dart/src/model/presence.dart';
 import 'package:centrifuge_dart/src/model/presence_stats.dart';
+import 'package:centrifuge_dart/src/model/pushes_stream.dart';
 
 /// Centrifuge client interface.
 abstract interface class ICentrifuge
@@ -72,8 +72,8 @@ abstract interface class ICentrifugeAsyncMessageSender {
 
 /// Centrifuge event receiver interface.
 abstract interface class ICentrifugeEventReceiver {
-  /// Stream of centrifuge events.
-  abstract final CentrifugeEventStream events;
+  /// Stream of received pushes from Centrifugo server for a channel.
+  abstract final CentrifugePushesStream stream;
 }
 
 /// Centrifuge client subscriptions manager interface.

lib/src/model/channel_presence.dart

@@ -1,18 +1,18 @@
-import 'package:centrifuge_dart/src/model/channel_event.dart';
+import 'package:centrifuge_dart/src/model/channel_push.dart';
 import 'package:centrifuge_dart/src/model/client_info.dart';
 import 'package:meta/meta.dart';
 
-/// {@template channel_presence_event}
+/// {@template channel_presence}
 /// Channel presence.
 /// Join / Leave events.
 /// {@endtemplate}
 /// {@category Entity}
 /// {@subCategory Channel}
 /// {@subCategory Presence}
 @immutable
-sealed class CentrifugeChannelPresenceEvent extends CentrifugeChannelEvent {
-  /// {@macro channel_presence_event}
-  const CentrifugeChannelPresenceEvent({
+sealed class CentrifugeChannelPresence extends CentrifugeChannelPush {
+  /// {@macro channel_presence}
+  const CentrifugeChannelPresence({
     required super.channel,
     required this.info,
   });
@@ -27,29 +27,35 @@ sealed class CentrifugeChannelPresenceEvent extends CentrifugeChannelEvent {
   abstract final bool isLeave;
 }
 
-/// {@macro channel_presence_event}
-final class CentrifugeJoinEvent extends CentrifugeChannelPresenceEvent {
-  /// {@macro channel_presence_event}
-  const CentrifugeJoinEvent({
+/// {@macro channel_presence}
+final class CentrifugeJoin extends CentrifugeChannelPresence {
+  /// {@macro channel_presence}
+  const CentrifugeJoin({
     required super.channel,
     required super.info,
   });
 
+  @override
+  String get type => 'join';
+
   @override
   bool get isJoin => true;
 
   @override
   bool get isLeave => false;
 }
 
-/// {@macro channel_presence_event}
-final class CentrifugeLeaveEvent extends CentrifugeChannelPresenceEvent {
-  /// {@macro channel_presence_event}
-  const CentrifugeLeaveEvent({
+/// {@macro channel_presence}
+final class CentrifugeLeave extends CentrifugeChannelPresence {
+  /// {@macro channel_presence}
+  const CentrifugeLeave({
     required super.channel,
     required super.info,
   });
 
+  @override
+  String get type => 'leave';
+
   @override
   bool get isJoin => false;
 

lib/src/model/channel_event.dart → lib/src/model/channel_push.dart

@@ -3,9 +3,9 @@ import 'package:centrifuge_dart/src/model/event.dart';
 /// {@template centrifuge_channel_event}
 /// Base class for all channel events.
 /// {@endtemplate}
-abstract base class CentrifugeChannelEvent extends CentrifugeEvent {
+abstract base class CentrifugeChannelPush extends CentrifugeEvent {
   /// {@template centrifuge_channel_event}
-  const CentrifugeChannelEvent({
+  const CentrifugeChannelPush({
     required this.channel,
   });
 

lib/src/model/event.dart

@@ -7,4 +7,7 @@ import 'package:meta/meta.dart';
 abstract base class CentrifugeEvent {
   /// {@template centrifuge_event}
   const CentrifugeEvent();
+
+  /// Event type.
+  abstract final String type;
 }

lib/src/model/publication.dart

@@ -1,4 +1,4 @@
-import 'package:centrifuge_dart/src/model/channel_event.dart';
+import 'package:centrifuge_dart/src/model/channel_push.dart';
 import 'package:centrifuge_dart/src/model/client_info.dart';
 import 'package:fixnum/fixnum.dart' as fixnum;
 import 'package:meta/meta.dart';
@@ -8,7 +8,7 @@ import 'package:meta/meta.dart';
 /// {@endtemplate}
 /// {@category Entity}
 @immutable
-final class CentrifugePublication extends CentrifugeChannelEvent {
+final class CentrifugePublication extends CentrifugeChannelPush {
   /// {@macro publication}
   const CentrifugePublication({
     required super.channel,
@@ -18,6 +18,9 @@ final class CentrifugePublication extends CentrifugeChannelEvent {
     this.tags,
   });
 
+  @override
+  String get type => 'publication';
+
   /// Publication payload
   final List<int> data;
 

lib/src/model/pushes_stream.dart

@@ -0,0 +1,42 @@
+import 'dart:async';
+
+import 'package:centrifuge_dart/src/model/channel_presence.dart';
+import 'package:centrifuge_dart/src/model/channel_push.dart';
+import 'package:centrifuge_dart/src/model/event.dart';
+import 'package:centrifuge_dart/src/model/publication.dart';
+
+/// Stream of received pushes from Centrifugo server for a channel.
+/// {@category Entity}
+/// {@subCategory Event}
+/// {@subCategory Channel}
+final class CentrifugePushesStream extends StreamView<CentrifugeChannelPush> {
+  /// Stream of received events.
+  CentrifugePushesStream({
+    required Stream<CentrifugeChannelPush> pushes,
+    required this.publications,
+    required this.presenceEvents,
+    required this.joinEvents,
+    required this.leaveEvents,
+  }) : super(pushes);
+
+  /// Publications stream.
+  final Stream<CentrifugePublication> publications;
+
+  /// Stream of presence (join & leave) events.
+  final Stream<CentrifugeChannelPresence> presenceEvents;
+
+  /// Join events
+  final Stream<CentrifugeJoin> joinEvents;
+
+  /// Leave events
+  final Stream<CentrifugeLeave> leaveEvents;
+
+  /// Filtered stream of data of [CentrifugeEvent].
+  Stream<T> whereType<T extends CentrifugeChannelPush>() =>
+      transform<T>(StreamTransformer<CentrifugeChannelPush, T>.fromHandlers(
+        handleData: (data, sink) => switch (data) {
+          T valid => sink.add(valid),
+          _ => null,
+        },
+      )).asBroadcastStream();
+}

lib/src/model/subscribe_event.dart

@@ -0,0 +1,12 @@
+import 'package:centrifuge_dart/src/model/channel_push.dart';
+
+/// {@macro subscribe_event}
+final class CentrifugeSubscribeEvent extends CentrifugeChannelPush {
+  /// {@macro subscribe_event}
+  const CentrifugeSubscribeEvent({
+    required super.channel,
+  });
+
+  @override
+  String get type => 'subscribe';
+}