Add since to state

lib/src/client/centrifuge.dart

@@ -199,10 +199,10 @@ base mixin CentrifugeEventReceiverMixin
             'and user ${leave.info.user}');
         _presenceController.add(leave);
         _leaveController.add(leave);
-      case CentrifugeSubscribe _:
-        break; // For server side subscriptions.
-      case CentrifugeUnsubscribe _:
-        break; // For server side subscriptions.
+      case CentrifugeSubscribe subscribe:
+        _serverSubscriptionManager.subscribe(subscribe);
+      case CentrifugeUnsubscribe unsubscribe:
+        _serverSubscriptionManager.unsubscribe(unsubscribe);
       case CentrifugeConnect _:
         break;
       case CentrifugeDisconnect event:
@@ -383,7 +383,9 @@ base mixin CentrifugeConnectionMixin
     try {
       _refreshTimer?.cancel();
       _refreshTimer = null;
-      await _transport.connect(url);
+      final subs = _serverSubscriptionManager.subscriptions.values
+          .toList(growable: false);
+      await _transport.connect(url, subs);
     } on CentrifugeException catch (error, stackTrace) {
       _emitError(error, stackTrace);
       rethrow;
@@ -544,18 +546,30 @@ base mixin CentrifugeClientSubscriptionMixin
   @override
   Future<void> close() async {
     await super.close();
-    _clientSubscriptionManager.removeAll();
+    _clientSubscriptionManager.close();
   }
 }
 
 /// Mixin responsible for server-side subscriptions.
 /// {@nodoc}
 @internal
 base mixin CentrifugeServerSubscriptionMixin on CentrifugeBase {
+  @override
+  void _onConnected(CentrifugeState$Connected state) {
+    super._onConnected(state);
+    _serverSubscriptionManager.setSubscribedAll();
+  }
+
+  @override
+  void _onDisconnected(CentrifugeState$Disconnected state) {
+    super._onDisconnected(state);
+    _serverSubscriptionManager.setSubscribingAll();
+  }
+
   @override
   Future<void> close() async {
     await super.close();
-    _serverSubscriptionManager.setUnsubscribed();
+    _serverSubscriptionManager.close();
   }
 }
 

lib/src/subscription/client_subscription_impl.dart

@@ -65,6 +65,12 @@ abstract base class CentrifugeClientSubscriptionBase
   /// Offset of last received publication.
   late fixnum.Int64 _offset;
 
+  @override
+  CentrifugeStreamPosition? get since => switch (state.since?.epoch) {
+        String epoch => (epoch: epoch, offset: _offset),
+        _ => null,
+      };
+
   /// Weak reference to transport.
   /// {@nodoc}
   @nonVirtual
@@ -287,7 +293,10 @@ base mixin CentrifugeClientSubscriptionSubscribeMixin
       }
       _refreshTimer?.cancel();
       _refreshTimer = null;
-      _setState(CentrifugeSubscriptionState$Subscribing(since: state.since));
+      _setState(CentrifugeSubscriptionState$Subscribing(
+        since: since ?? state.since,
+        recoverable: state.recoverable,
+      ));
       final subscribed = await _transport.subscribe(
         channel,
         _config,
@@ -302,7 +311,7 @@ base mixin CentrifugeClientSubscriptionSubscribeMixin
       final offset = subscribed.since?.offset;
       if (offset != null && offset > _offset) _offset = offset;
       _setState(CentrifugeSubscriptionState$Subscribed(
-        since: subscribed.since,
+        since: subscribed.since ?? since ?? state.since,
         recoverable: subscribed.recoverable,
         ttl: subscribed.ttl,
       ));
@@ -380,7 +389,8 @@ base mixin CentrifugeClientSubscriptionSubscribeMixin
     _setState(CentrifugeSubscriptionState.unsubscribed(
       code: code,
       reason: reason,
-      since: state.since,
+      since: since ?? state.since,
+      recoverable: state.recoverable,
     ));
     try {
       await _transport.unsubscribe(channel, _config);

lib/src/subscription/client_subscription_manager.dart

@@ -112,7 +112,7 @@ final class ClientSubscriptionManager {
 
   /// Remove all subscriptions for the specific client from internal registry.
   /// {@nodoc}
-  void removeAll([
+  void close([
     int code = 0,
     String reason = 'client closed',
   ]) {

lib/src/subscription/server_subscription_impl.dart

@@ -61,6 +61,12 @@ abstract base class CentrifugeServerSubscriptionBase
   @override
   final String channel;
 
+  @override
+  CentrifugeStreamPosition? get since => switch (state.since?.epoch) {
+        String epoch => (epoch: epoch, offset: _offset),
+        _ => null,
+      };
+
   /// Offset of last received publication.
   fixnum.Int64 _offset = fixnum.Int64.ZERO;
 
@@ -198,15 +204,18 @@ base mixin CentrifugeServerSubscriptionEventReceiverMixin
         _presenceController.add(leave);
         _leaveController.add(leave);
       case CentrifugeSubscribe sub:
+        final offset = sub.streamPosition?.offset;
+        if (offset != null && offset > _offset) _offset = offset;
         _setState(CentrifugeSubscriptionState.subscribed(
-          since: sub.streamPosition,
+          since: sub.streamPosition ?? since ?? state.since,
           recoverable: sub.recoverable,
         ));
       case CentrifugeUnsubscribe unsub:
         _setState(CentrifugeSubscriptionState.unsubscribed(
           code: unsub.code,
           reason: unsub.reason,
-          since: state.since,
+          recoverable: state.recoverable,
+          since: since ?? state.since,
         ));
       case CentrifugeConnect _:
         break;

lib/src/subscription/server_subscription_manager.dart

@@ -1,6 +1,8 @@
 import 'dart:collection';
 
 import 'package:centrifuge_dart/src/model/channel_push.dart';
+import 'package:centrifuge_dart/src/model/subscribe.dart';
+import 'package:centrifuge_dart/src/model/unsubscribe.dart';
 import 'package:centrifuge_dart/src/subscription/server_subscription_impl.dart';
 import 'package:centrifuge_dart/src/subscription/subscription.dart';
 import 'package:centrifuge_dart/src/transport/transport_interface.dart';
@@ -35,31 +37,37 @@ final class ServerSubscriptionManager {
           entry.key: entry.value,
       });
 
+  /// Called on [CentrifugeSubscribe] push from server.
+  void subscribe(CentrifugeSubscribe subscribe) {}
+
+  /// Called on [CentrifugeUnsubscribe] push from server.
+  void unsubscribe(CentrifugeUnsubscribe subscribe) {}
+
   /// Called when subscribed to a server-side channel upon Client moving to
   /// connected state or during connection lifetime if server sends Subscribe
   /// push message.
   /// {@nodoc}
-  void setSubscribed() {
+  void setSubscribedAll() {
     for (final entry in _channelSubscriptions.values) {}
   }
 
   /// Called when existing connection lost (Client reconnects) or Client
   /// explicitly disconnected. Client continue keeping server-side subscription
   /// registry with stream position information where applicable.
   /// {@nodoc}
-  void setSubscribing() {
+  void setSubscribingAll() {
     for (final entry in _channelSubscriptions.values) {}
   }
 
   /// Called when server sent unsubscribe push or server-side subscription
   /// previously existed in SDK registry disappeared upon Client reconnect.
   /// {@nodoc}
-  void setUnsubscribed() {
+  void setUnsubscribedAll() {
     for (final entry in _channelSubscriptions.values) {}
   }
 
   void close() {
-    setUnsubscribed();
+    setUnsubscribedAll();
     _channelSubscriptions.clear();
   }
 

lib/src/subscription/subscription.dart

@@ -21,6 +21,9 @@ abstract interface class ICentrifugeSubscription {
   /// Current subscription state.
   abstract final CentrifugeSubscriptionState state;
 
+  /// Offset of last successfully received message.
+  abstract final CentrifugeStreamPosition? since;
+
   /// Stream of subscription states.
   abstract final CentrifugeSubscriptionStateStream states;
 

lib/src/subscription/subscription_state.dart

@@ -16,7 +16,10 @@ import 'package:meta/meta.dart';
 sealed class CentrifugeSubscriptionState
     extends _$CentrifugeSubscriptionStateBase {
   /// {@macro subscription_state}
-  const CentrifugeSubscriptionState(super.timestamp, super.since);
+  const CentrifugeSubscriptionState(
+      {required super.timestamp,
+      required super.since,
+      required super.recoverable});
 
   /// Unsubscribed
   /// {@macro subscription_state}
@@ -25,13 +28,15 @@ sealed class CentrifugeSubscriptionState
     required String reason,
     DateTime? timestamp,
     ({fixnum.Int64 offset, String epoch})? since,
+    bool recoverable,
   }) = CentrifugeSubscriptionState$Unsubscribed;
 
   /// Subscribing
   /// {@macro subscription_state}
   factory CentrifugeSubscriptionState.subscribing({
     DateTime? timestamp,
     ({fixnum.Int64 offset, String epoch})? since,
+    bool recoverable,
   }) = CentrifugeSubscriptionState$Subscribing;
 
   /// Subscribed
@@ -57,7 +62,11 @@ final class CentrifugeSubscriptionState$Unsubscribed
     required this.reason,
     DateTime? timestamp,
     ({fixnum.Int64 offset, String epoch})? since,
-  }) : super(timestamp ?? DateTime.now(), since);
+    bool recoverable = false,
+  }) : super(
+            timestamp: timestamp ?? DateTime.now(),
+            since: since,
+            recoverable: recoverable);
 
   /// Unsubscribe code.
   final int code;
@@ -108,7 +117,11 @@ final class CentrifugeSubscriptionState$Subscribing
   CentrifugeSubscriptionState$Subscribing({
     DateTime? timestamp,
     ({fixnum.Int64 offset, String epoch})? since,
-  }) : super(timestamp ?? DateTime.now(), since);
+    bool recoverable = false,
+  }) : super(
+            timestamp: timestamp ?? DateTime.now(),
+            since: since,
+            recoverable: recoverable);
 
   @override
   bool get isUnsubscribed => false;
@@ -153,12 +166,12 @@ final class CentrifugeSubscriptionState$Subscribed
   CentrifugeSubscriptionState$Subscribed({
     DateTime? timestamp,
     ({fixnum.Int64 offset, String epoch})? since,
-    this.recoverable = false,
+    bool recoverable = false,
     this.ttl,
-  }) : super(timestamp ?? DateTime.now(), since);
-
-  /// Whether channel is recoverable.
-  final bool recoverable;
+  }) : super(
+            timestamp: timestamp ?? DateTime.now(),
+            since: since,
+            recoverable: recoverable);
 
   /// Time to live in seconds.
   final DateTime? ttl;
@@ -208,14 +221,21 @@ typedef CentrifugeSubscriptionStateMatch<R,
 @immutable
 abstract base class _$CentrifugeSubscriptionStateBase {
   /// {@nodoc}
-  const _$CentrifugeSubscriptionStateBase(this.timestamp, this.since);
+  const _$CentrifugeSubscriptionStateBase({
+    required this.timestamp,
+    required this.since,
+    required this.recoverable,
+  });
 
   /// Timestamp of state change.
   final DateTime timestamp;
 
   /// Stream Position
   final ({fixnum.Int64 offset, String epoch})? since;
 
+  /// Whether channel is recoverable.
+  final bool recoverable;
+
   /// Whether channel is unsubscribed.
   abstract final bool isUnsubscribed;
 

lib/src/transport/transport_interface.dart

@@ -29,8 +29,9 @@ abstract interface class ICentrifugeTransport {
 
   /// Connect to the server.
   /// [url] is a URL of endpoint.
+  /// [subs] is a list of server-side subscriptions to subscribe on connect.
   /// {@nodoc}
-  Future<void> connect(String url);
+  Future<void> connect(String url, List<CentrifugeServerSubscription> subs);
 
   /// Send asynchronous message to a server. This method makes sense
   /// only when using Centrifuge library for Go on a server side. In Centrifuge