Update placement

analysis_options.yaml

@@ -58,6 +58,7 @@ linter:
     prefer_relative_imports: false
     prefer_final_locals: false
     avoid_escaping_inner_quotes: false
+    curly_braces_in_flow_control_structures: false
 
     # Enabled
     use_named_constants: true
@@ -149,7 +150,6 @@ linter:
     avoid_shadowing_type_parameters: true
     avoid_types_as_parameter_names: true
     camel_case_extensions: true
-    curly_braces_in_flow_control_structures: true
     empty_catches: true
     empty_constructor_bodies: true
     library_names: true

lib/centrifuge.dart

@@ -1,14 +1,14 @@
 library centrifuge;
 
 export 'package:centrifuge_dart/src/client/centrifuge.dart' show Centrifuge;
+export 'package:centrifuge_dart/src/client/config.dart';
+export 'package:centrifuge_dart/src/client/state.dart';
+export 'package:centrifuge_dart/src/client/states_stream.dart';
 export 'package:centrifuge_dart/src/model/client_info.dart';
-export 'package:centrifuge_dart/src/model/config.dart';
 export 'package:centrifuge_dart/src/model/exception.dart';
 export 'package:centrifuge_dart/src/model/jwt.dart';
 export 'package:centrifuge_dart/src/model/publication.dart';
-export 'package:centrifuge_dart/src/model/state.dart';
-export 'package:centrifuge_dart/src/model/states_stream.dart';
-export 'package:centrifuge_dart/src/model/subscription.dart'
+export 'package:centrifuge_dart/src/subscription/subscription.dart'
     show CentrifugeClientSubscription, CentrifugeServerSubscription;
-export 'package:centrifuge_dart/src/model/subscription_config.dart';
-export 'package:centrifuge_dart/src/model/subscription_state.dart';
+export 'package:centrifuge_dart/src/subscription/subscription_config.dart';
+export 'package:centrifuge_dart/src/subscription/subscription_state.dart';

lib/interface.dart

@@ -1,17 +1,17 @@
 library interface;
 
 export 'package:centrifuge_dart/src/client/centrifuge_interface.dart';
+export 'package:centrifuge_dart/src/client/config.dart';
+export 'package:centrifuge_dart/src/client/state.dart';
+export 'package:centrifuge_dart/src/client/states_stream.dart';
 export 'package:centrifuge_dart/src/model/client_info.dart';
-export 'package:centrifuge_dart/src/model/config.dart';
 export 'package:centrifuge_dart/src/model/exception.dart';
 export 'package:centrifuge_dart/src/model/jwt.dart';
 export 'package:centrifuge_dart/src/model/publication.dart';
-export 'package:centrifuge_dart/src/model/state.dart';
-export 'package:centrifuge_dart/src/model/states_stream.dart';
-export 'package:centrifuge_dart/src/model/subscription.dart'
+export 'package:centrifuge_dart/src/subscription/subscription.dart'
     show
         ICentrifugeSubscription,
         CentrifugeClientSubscription,
         CentrifugeServerSubscription;
-export 'package:centrifuge_dart/src/model/subscription_config.dart';
-export 'package:centrifuge_dart/src/model/subscription_state.dart';
+export 'package:centrifuge_dart/src/subscription/subscription_config.dart';
+export 'package:centrifuge_dart/src/subscription/subscription_state.dart';

lib/src/client/centrifuge.dart

@@ -1,13 +1,13 @@
 import 'dart:async';
 
 import 'package:centrifuge_dart/src/client/centrifuge_interface.dart';
-import 'package:centrifuge_dart/src/model/config.dart';
+import 'package:centrifuge_dart/src/client/config.dart';
+import 'package:centrifuge_dart/src/client/state.dart';
+import 'package:centrifuge_dart/src/client/states_stream.dart';
 import 'package:centrifuge_dart/src/model/exception.dart';
-import 'package:centrifuge_dart/src/model/state.dart';
-import 'package:centrifuge_dart/src/model/states_stream.dart';
-import 'package:centrifuge_dart/src/model/subscription.dart';
-import 'package:centrifuge_dart/src/model/subscription_config.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';
 import 'package:centrifuge_dart/src/transport/transport_interface.dart';
 import 'package:centrifuge_dart/src/transport/ws_protobuf_transport.dart';
 import 'package:centrifuge_dart/src/util/event_queue.dart';

lib/src/model/states_stream.dart → lib/src/client/states_stream.dart

@@ -1,6 +1,6 @@
 import 'dart:async';
 
-import 'package:centrifuge_dart/src/model/state.dart';
+import 'package:centrifuge_dart/src/client/state.dart';
 
 /// Stream of Centrifuge's [CentrifugeState] changes.
 /// {@category Client}

lib/src/subscription/client_subscription_impl.dart

@@ -1,7 +1,10 @@
 import 'dart:async';
 
 import 'package:centrifuge_dart/centrifuge.dart';
-import 'package:centrifuge_dart/src/model/subscription_states_stream.dart';
+import 'package:centrifuge_dart/src/subscription/subscription.dart';
+import 'package:centrifuge_dart/src/subscription/subscription_config.dart';
+import 'package:centrifuge_dart/src/subscription/subscription_state.dart';
+import 'package:centrifuge_dart/src/subscription/subscription_states_stream.dart';
 import 'package:centrifuge_dart/src/transport/transport_interface.dart';
 import 'package:centrifuge_dart/src/util/event_queue.dart';
 import 'package:centrifuge_dart/src/util/logger.dart' as logger;
@@ -80,7 +83,8 @@ abstract base class CentrifugeClientSubscriptionBase
   @override
   CentrifugeSubscriptionState get state => _state;
   late CentrifugeSubscriptionState _state =
-      CentrifugeSubscriptionState.unsubscribed(since: _config.since);
+      CentrifugeSubscriptionState.unsubscribed(
+          since: _config.since, code: 0, reason: 'initial state');
 
   /// Stream of subscription states.
   /// {@nodoc}
@@ -175,8 +179,17 @@ base mixin CentrifugeClientSubscriptionSubscribeMixin
       if (state is CentrifugeSubscriptionState$Subscribing) {
         return await ready();
       }
-      _setState(CentrifugeSubscriptionState$Subscribing());
-      // TODO(plugfox): implement
+      _setState(CentrifugeSubscriptionState$Subscribing(since: state.since));
+      final subscribed = await _transport.subscribe(
+        channel,
+        _config,
+        state.since,
+      );
+      _setState(CentrifugeSubscriptionState$Subscribed(
+        since: subscribed.since,
+        recoverable: subscribed.recoverable,
+        ttl: subscribed.ttl,
+      ));
     } on CentrifugeException catch (error, stackTrace) {
       _emitError(error, stackTrace);
       rethrow;

lib/src/subscription/client_subscription_manager.dart

@@ -1,9 +1,9 @@
 import 'dart:collection';
 
 import 'package:centrifuge_dart/src/model/exception.dart';
-import 'package:centrifuge_dart/src/model/subscription.dart';
-import 'package:centrifuge_dart/src/model/subscription_config.dart';
 import 'package:centrifuge_dart/src/subscription/client_subscription_impl.dart';
+import 'package:centrifuge_dart/src/subscription/subscription.dart';
+import 'package:centrifuge_dart/src/subscription/subscription_config.dart';
 import 'package:centrifuge_dart/src/transport/transport_interface.dart';
 import 'package:meta/meta.dart';
 

lib/src/model/subscription.dart → lib/src/subscription/subscription.dart

@@ -1,7 +1,9 @@
 import 'dart:async';
 
-import 'package:centrifuge_dart/interface.dart';
-import 'package:centrifuge_dart/src/model/subscription_states_stream.dart';
+import 'package:centrifuge_dart/src/model/exception.dart';
+import 'package:centrifuge_dart/src/model/publication.dart';
+import 'package:centrifuge_dart/src/subscription/subscription_state.dart';
+import 'package:centrifuge_dart/src/subscription/subscription_states_stream.dart';
 import 'package:fixnum/fixnum.dart' as fixnum;
 
 /// {@template subscription}

lib/src/model/subscription_state.dart → lib/src/subscription/subscription_state.dart

@@ -21,6 +21,8 @@ sealed class CentrifugeSubscriptionState
   /// Unsubscribed
   /// {@macro subscription_state}
   factory CentrifugeSubscriptionState.unsubscribed({
+    required int code,
+    required String reason,
     DateTime? timestamp,
     ({fixnum.Int64 offset, String epoch})? since,
   }) = CentrifugeSubscriptionState$Unsubscribed;
@@ -37,6 +39,8 @@ sealed class CentrifugeSubscriptionState
   factory CentrifugeSubscriptionState.subscribed({
     DateTime? timestamp,
     ({fixnum.Int64 offset, String epoch})? since,
+    bool recoverable,
+    DateTime? ttl,
   }) = CentrifugeSubscriptionState$Subscribed;
 }
 
@@ -49,10 +53,18 @@ final class CentrifugeSubscriptionState$Unsubscribed
     extends CentrifugeSubscriptionState with _$CentrifugeSubscriptionState {
   /// {@nodoc}
   CentrifugeSubscriptionState$Unsubscribed({
+    required this.code,
+    required this.reason,
     DateTime? timestamp,
     ({fixnum.Int64 offset, String epoch})? since,
   }) : super(timestamp ?? DateTime.now(), since);
 
+  /// Unsubscribe code.
+  final int code;
+
+  /// Unsubscribe reason.
+  final String reason;
+
   @override
   R map<R>({
     required CentrifugeSubscriptionStateMatch<R,
@@ -123,8 +135,16 @@ final class CentrifugeSubscriptionState$Subscribed
   CentrifugeSubscriptionState$Subscribed({
     DateTime? timestamp,
     ({fixnum.Int64 offset, String epoch})? since,
+    this.recoverable = false,
+    this.ttl,
   }) : super(timestamp ?? DateTime.now(), since);
 
+  /// Whether channel is recoverable.
+  final bool recoverable;
+
+  /// Time to live in seconds.
+  final DateTime? ttl;
+
   @override
   R map<R>({
     required CentrifugeSubscriptionStateMatch<R,
@@ -140,7 +160,7 @@ final class CentrifugeSubscriptionState$Subscribed
       subscribed(this);
 
   @override
-  int get hashCode => Object.hash(2, timestamp, since);
+  int get hashCode => Object.hash(2, timestamp, since, recoverable, ttl);
 
   @override
   bool operator ==(Object other) => identical(this, other);

lib/src/model/subscription_states_stream.dart → lib/src/subscription/subscription_states_stream.dart

@@ -1,6 +1,6 @@
 import 'dart:async';
 
-import 'package:centrifuge_dart/src/model/subscription_state.dart';
+import 'package:centrifuge_dart/src/subscription/subscription_state.dart';
 
 /// Stream of Centrifuge's [CentrifugeSubscriptionState] changes.
 /// {@category Subscription}

lib/src/transport/transport_interface.dart

@@ -1,9 +1,9 @@
 import 'dart:async';
 
-import 'package:centrifuge_dart/src/model/state.dart';
+import 'package:centrifuge_dart/src/client/state.dart';
 import 'package:centrifuge_dart/src/model/stream_position.dart';
-import 'package:centrifuge_dart/src/model/subcibed_on_channel.dart';
-import 'package:centrifuge_dart/src/model/subscription_config.dart';
+import 'package:centrifuge_dart/src/subscription/subcibed_on_channel.dart';
+import 'package:centrifuge_dart/src/subscription/subscription_config.dart';
 import 'package:meta/meta.dart';
 
 /// Class responsible for sending and receiving data from the server.

lib/src/transport/ws_protobuf_transport.dart

@@ -5,7 +5,7 @@ import 'package:centrifuge_dart/centrifuge.dart';
 import 'package:centrifuge_dart/src/model/disconnect_code.dart';
 import 'package:centrifuge_dart/src/model/protobuf/client.pb.dart' as pb;
 import 'package:centrifuge_dart/src/model/stream_position.dart';
-import 'package:centrifuge_dart/src/model/subcibed_on_channel.dart';
+import 'package:centrifuge_dart/src/subscription/subcibed_on_channel.dart';
 import 'package:centrifuge_dart/src/transport/transport_interface.dart';
 import 'package:centrifuge_dart/src/transport/transport_protobuf_codec.dart';
 import 'package:centrifuge_dart/src/util/logger.dart' as logger;

lib/src/util/notifier.dart

@@ -0,0 +1,36 @@
+/// Notify about value changes.
+/// {@nodoc}
+typedef ValueChanged<T> = void Function(T value);
+
+/// Notify about value changes.
+/// {@nodoc}
+final class CentrifugeValueNotifier<T> {
+  /// Notify about value changes.
+  /// {@nodoc}
+  CentrifugeValueNotifier(this._value);
+
+  /// Current value.
+  /// {@nodoc}
+  T get value => _value;
+  T _value;
+
+  /// Notify about value changes.
+  /// {@nodoc}
+  void notify(T value) {
+    if (_value == value) return;
+    _value = value;
+    for (var i = 0; i < _listeners.length; i++) _listeners[i](value);
+  }
+
+  /// Listeners.
+  /// {@nodoc}
+  final List<ValueChanged<T>> _listeners = <ValueChanged<T>>[];
+
+  /// Add listener.
+  /// {@nodoc}
+  void addListener(ValueChanged<T> listener) => _listeners.add(listener);
+
+  /// Remove listener.
+  /// {@nodoc}
+  void removeListener(ValueChanged<T> listener) => _listeners.remove(listener);
+}