可观察有状态生命周期

Question

一般设计问题可以描述为：

我有一个严格的生命周期的websocket连接 - 它需要适当地调用connect和disconnect，并且因为它与系统对话，所以它使用。在这个websocket连接中，我们有多个不同的Subscription对象，每个对象都有一个它想要被尊重的严格生命周期（subscribe和unsubscribe），并且它依赖于其父websocket的状态来使这些操作成功。

这是三个嵌套生命周期可观察量的理想行为的时间线，其中C依赖于B，它取决于A：

A = someInput.switchMap((i) => LifecycleObservable())
B = A.switchMap((a) => LifecycleObservable())
C = B.switchMap((b) => LifecycleObservable())

C.listen(print);

// <-- listen to c
// <-- produce [someInput]
setup A
setup B
setup C
// <-- c is produced

// <-- c is unsubscribed
teardown C
teardown B
teardown A

// <-- C is re-subscribed-to
setup A
setup B
setup C

// <-- produce [someInput]
teardown C
teardown B
teardown A
setup A
setup B
setup C
// <-- c is produced

第一个问题：这是反模式吗？我无法在网上找到很多关于这种模式的东西，但它看起来像是一个非常标准的东西，你会遇到observables：一些对象只有一个生命周期，一些对象可能想要依赖它。

我可以使用以下内容非常接近这种理想的行为：

class LifecycleObservable {
  static Observable<T> fromObservable<T>({
    @required Observable<T> input,
    @required Future<void> Function(T) setup,
    @required Future<void> Function(T) teardown,
  }) {
    return input.asyncMap((T _input) async {
      await setup(_input);
      return _input;
    }).switchMap((T _input) {
      return Observable<T>(Observable.never()) //
          .startWith(_input)
          .doOnCancel(() async {
        await teardown(_input);
      });
    });
  }
}

这段代码接受一个有状态对象流，在它们生成时对它们运行setup，并且当teardown中的子可观察对象被取消时对它们进行switchMap。

当在原始理想化时间轴中产生第二个[someInput]时出现问题：使用上面的代码我得到一个类似的调用图

// <-- listen to c
// <-- produce [someInput]
setup A
setup B
setup C
// <-- c is produced

// <-- produce [someInput]
teardown A
setup A
teardown B
setup B
teardown C
setup C
// <-- c is produced

问题在于，如果B依赖于A（就像从依赖于打开的websocket传输的订阅调用unsubscribe），这个拆卸命令会打破每个对象的预期生命周期（订阅尝试通过封闭的websocket传输发送unsubscribe。

Answer 1

在我看来，很简单，可观察的模式不能表达这些语义。具体来说，可观察模式不是为级联依赖项而设计的 - 父级可观察对于其子可观察对象的状态一无所知。

我用以下的dart代码解决了这个问题。我确定这很糟糕，但它似乎对我有用。

class WithLifecycle<T> {
  final FutureOr<void> Function() setup;
  final FutureOr<void> Function() teardown;
  final T value;
  final WithLifecycle parent;

  List<WithLifecycle> _children = [];
  bool _disposed = false;
  WithLifecycle({
    @required this.value,
    this.setup,
    this.teardown,
    this.parent,
  });

  void addDependency(WithLifecycle child) => _children.add(child);
  void removeDependency(WithLifecycle child) => _children.remove(child);
  Future<void> init() async {
    parent?.addDependency(this);
    await setup();
  }

  Future<void> dispose() async {
    if (_disposed) {
      return;
    }

    _disposed = true;
    for (var _child in _children) {
      await _child.dispose();
    }
    _children.clear();
    await teardown();
  }
}

然后在使用observable时用于创建必要的依赖链：

class LifecycleObservable {
  static Observable<WithLifecycle<T>> fromObservable<T>({
    @required Observable<T> value,
    WithLifecycle parent,
    @required Future<void> Function(T) setup,
    @required Future<void> Function(T) teardown,
  }) {
    return value.concatMap((T _value) {
      final withLifecycle = WithLifecycle<T>(
        value: _value,
        parent: parent,
        setup: () => setup(_value),
        teardown: () => teardown(_value),
      );
      return Observable<WithLifecycle<T>>(Observable.never())
          .startWith(withLifecycle)
          .doOnListen(() async {
        await withLifecycle.init();
      }).doOnCancel(() async {
        await withLifecycle.dispose();
      });
    });
  }
}

用的像

token$ = PublishSubject();
    channel$ = token$.switchMap((token) {
      return LifecycleObservable.fromObservable<IOWebSocketChannel>(
          value: Observable.just(IOWebSocketChannel.connect(Constants.connectionString)),
          setup: (channel) async {
            print("setup A ${channel.hashCode}");
          },
          teardown: (channel) async {
            print("teardown A ${channel.hashCode}");
            await channel.sink.close(status.goingAway);
          });
    });

    streams$ = channel$.switchMap((channel) {
      return LifecycleObservable.fromObservable<Stream<String>>(
        parent: channel,
        value: Observable.just(channel.value.stream.cast<String>()),
        setup: (thing) async {
          print("setup B ${thing.hashCode}");
        },
        teardown: (thing) async {
          print("teardown B ${thing.hashCode}");
        },
      );
    });

    messages = streams$.flatMap((i) => i.value).share();

并最终得到如下所示的调用图

// <- push [token]
flutter: setup A 253354366
flutter: setup B 422603720
// <- push [token]
flutter: teardown B 422603720
flutter: teardown A 253354366
flutter: setup A 260164938
flutter: setup B 161253018