订阅/重新订阅 Observable Stream [重复]

Question

我遇到了一个问题，我想在谓词为真时订阅可观察流，并在谓词为假时停止订阅。当谓词在未来某个时间再次为真时，它应该重新订阅可观察流。

用例：

如果我无法将日志实体插入数据库，它应该取消订阅流，并且当数据库重新运行时，它应该自动订阅流（基于属性IsSubscribed）并开始插入数据。

我自己的尝试：

我已经尝试了以下 没有工作：

var groups = from dataItem in items.SelectMany(o => o.GroupBy(i => i.EntityType))
    where dataItem.Any()
    select new {Type = dataItem.Key, List = dataItem.Select(o => o)};

groups
    .TakeWhile(o => IsSubscribed)
    .SubscribeOn(_scheduler)
    .Repeat()
    .Subscribe(o => Insert(o.Type, o.List));

基于属性IsSubscribed，我想流式订阅和取消订阅。当TakeWhile 为真时，OnCompleted 被调用，而当Subscribe 之后将无法工作。 旁注：这是一个冷的可观察流

问题：

如何创建一个可观察的流，我可以在其中订阅和取消订阅任意多次（有点像 C# 中的事件处理程序）

感谢您提前提供帮助

Answer 1

你想要的是添加 团体 .Delay(group.SelectMany(WaitForDatabaseUp))

public async Task WaitForDatabaseUp()
{
    //If IsSubscribed continue execution
    if(IsSubscribed) return;
    //Else wait until IsSubscribed == true
    await this.ObservableForProperty(x => x.IsSubscribed, skipInitial: false)
                       .Value()
                       .Where(isSubscribed => isSubscribed)
                       .Take(1);
}

使用您最喜欢的框架将 INPC 转换为您可以看到 ObserveProperty() 的 Observable

基本上我们内联了一个仅在IsSubscribed == true 时返回的任务。然后将该 Task 转换为 Observable，以便与 Rx 兼容。

Answer 2

看起来像一个重复的问题。

但是，从Pause and Resume Subscription on cold IObservable拉取代码，可以调整为

var subscription = Observable.Create<IObservable<YourType>>(o =>
{
    var current = groups.Replay();
    var connection = new SerialDisposable();
    connection.Disposable = current.Connect();

    return IsSubscribed
        .DistinctUntilChanged()
        .Select(isRunning =>
        {
            if (isRunning)
            {
                //Return the current replayed values.
                return current;
            }
            else
            {
                //Disconnect and replace current.
                current = source.Replay();
                connection.Disposable = current.Connect();
                //yield silence until the next time we resume.
                return Observable.Never<YourType>();
            }

        })
        .Subscribe(o);
})
.Switch()
.Subscribe(o => Insert(o.Type, o.List));

您可以看到 Matt Barrett（和我）谈论它here。我建议观看整个视频（可能是 2 倍速）以了解完整的上下文。