add states.activated

lib/src/observables/observables.dart

@@ -1,6 +1,7 @@
 
 export 'package:disposal/disposal.dart' show Disposable;
 export 'states/states.dart';
+export 'states/states_activated.dart';
 export 'states/observable_as_states_x.dart';
 export 'errors/latest_state_not_replay_error.dart';
 export 'observables/observable.dart';

lib/src/observables/states/states.dart

@@ -5,6 +5,7 @@ import 'package:typedef_equals/typedef_equals.dart';
 import '../observers/observer.dart';
 import '../observables/observable.dart';
 import 'observable_as_states_x.dart';
+import 'states_activated.dart';
 import 'states_first.dart';
 
 class States<T> {
@@ -105,4 +106,10 @@ extension StatesX<T> on States<T> {
   T get first {
     return statesFirst(this);
   }
+
+  StatesActivated<T> activated() {
+    return StatesActivated<T>(
+      child: this
+    );
+  }
 }

lib/src/observables/states/states_activated.dart

@@ -0,0 +1,49 @@
+
+import 'package:meta/meta.dart';
+import 'package:disposal/disposal.dart';
+
+import '../subjects/replay_subject.dart';
+import 'observable_as_states_x.dart';
+import 'states.dart';
+
+abstract class HasStates<T> {
+  States<T> get states;
+}
+
+abstract class StatesActivated<T> implements HasStates, Disposable {
+  @internal
+  factory StatesActivated({
+    required States<T> child
+  }) = _StatesActivated;
+}
+
+class _StatesActivated<T> implements StatesActivated<T> {
+
+  _StatesActivated({
+    required States<T> child
+  }): _child = child {
+    _startObserve();
+  }
+
+  final States<T> _child;
+  final _subject = ReplaySubject<T>(bufferSize: 1);
+
+  @override
+  late final states = _subject.asStates();
+
+  late final _observation;
+
+  void _startObserve() {
+    _observation = _child.observe(_subject.onData);
+  }
+
+  void _stopObserve() {
+    _observation.dispose();
+  }
+
+  @override
+  void dispose() {
+    _stopObserve();
+    _subject.dispose();
+  }
+}

test/observables/observables_test.dart

@@ -1,4 +1,5 @@
 
+import 'states/states_activated_test.dart' as states_activated_test;
 import 'states/states_cache_test.dart' as states_cache_test;
 import 'states/states_cast_test.dart' as states_cast_test;
 import 'states/states_combine_test.dart' as states_combine_test;
@@ -27,6 +28,7 @@ import 'subjects/value_subject_test.dart' as value_subject_test;
 
 
 void main() {
+  states_activated_test.main();
   states_cache_test.main();
   states_cast_test.main();
   states_combine_test.main();

test/observables/states/states_activated_test.dart

@@ -0,0 +1,143 @@
+
+import 'package:test/test.dart';
+import 'package:scopes/scopes.dart';
+
+import '../shared/states_tester.dart';
+
+void main() {
+
+  test('`states.activated` connect immediately', () {
+
+    int invokes = 0;
+
+    final states = States<String>((setState) {
+      setState('a');
+      invokes += 1;
+      return Disposable.empty;
+    });
+
+    expect(invokes, 0);
+    final activated = states.activated();
+    expect(invokes, 1);
+
+    activated.dispose();
+
+  });
+
+  test('`states.activated` disconnect when dispose(deactivate) called', () {
+
+    int invokes = 0;
+
+    final states = States<String>((setState) {
+      setState('a');
+      return Disposable(() {
+        invokes += 1;
+      });
+    });
+
+    final activated = states.activated();
+
+    expect(invokes, 0);
+    activated.dispose();
+    expect(invokes, 1);
+
+  });
+
+  test('`states.activated` forward data to observers', () async {
+
+    final states = States<String>((setState) {
+      setState('a');
+      Future(() => setState('b'));
+      return Disposable.empty;
+    });
+
+    final activated = states.activated();
+
+    final tester1 = StatesTester(
+      activated.states,
+    );
+
+    final tester2 = StatesTester(
+      activated.states,
+    );
+
+    tester1.startObserve();
+    tester2.startObserve();
+
+    expect(tester1.recorded, [
+      'a',
+    ]);
+    expect(tester2.recorded, [
+      'a',
+    ]);
+    await Future(() {});
+    expect(tester1.recorded, [
+      'a',
+      'b',
+    ]);
+    expect(tester2.recorded, [
+      'a',
+      'b',
+    ]);
+
+    tester1.stopObserve();
+    tester2.stopObserve();
+
+    activated.dispose();
+
+  });
+
+  test('`states.activated` replay data to observers', () {
+
+    final states = States<String>((setState) {
+      setState('a');
+      setState('b');
+      setState('c');
+      return Disposable.empty;
+    });
+
+    final activated = states.activated();
+
+    final tester = StatesTester(
+      activated.states,
+    );
+
+    expect(tester.recorded, []);
+    tester.startObserve();
+    expect(tester.recorded, [
+      'c',
+    ]);
+
+    tester.stopObserve();
+    activated.dispose();
+  });
+
+  test("`states.activated` won't forward data after disposed(deactivated)", () async {
+
+    final states = States<String>((setState) {
+      setState('a');
+      Future(() => setState('b'));
+      return Disposable.empty;
+    });
+
+    final activated = states.activated();
+
+    final tester = StatesTester(
+      activated.states,
+    );
+
+    tester.startObserve();
+    activated.dispose();
+
+    expect(tester.recorded, [
+      'a',
+    ]);
+    await Future(() {});
+    expect(tester.recorded, [
+      'a',
+    ]);
+
+    tester.stopObserve();
+
+  });
+}