在rx.net中，如何使主题与TaskCompletionsource行为相似？

Question

在rx.net中，如何使主题类似于 taskcompletionsource.task 行为？

即使完成，它也需要缓存并回复第一个事件。 asyncSubject 也不是 replaySubject（BufferSize：1）都会做到这一点。

例如（让我们称其为 PromisesUbject ）：

//var subj = new ReplaySubject<int>(bufferSize: 1);
var subj = new PromiseSubject<int>();

subj.Subscribe(i => Console.WriteLine(i));

subj.OnNext(1);
subj.OnNext(2);
subj.OnNext(3);
subj.OnCompleted();

subj.Subscribe(i => Console.WriteLine(i));

Console.ReadLine();

预期输出：

1
1

我可以使用 taskcompletionsource ， taskobservableExtensions.toObservable >主题base 衍生的主题实现，但是使用RX运算符的组成，是否有一种优雅的方式？

更新，我通过 taskCompletionsource 的初步尝试：

public class PromiseSubject<T> : ISubject<T>
{
    private readonly TaskCompletionSource<(bool HasValue, T Value)> _tcs;
    private readonly IObservable<T> _observable;

    public PromiseSubject()
    {
        _tcs = new(TaskCreationOptions.RunContinuationsAsynchronously);
        _observable = _tcs.Task.ToObservable()
            .Where(r => r.HasValue).Select(r => r.Value!);
    }

    public void OnCompleted() =>
        _tcs.TrySetResult((false, default!));

    public void OnError(Exception error) =>
        _tcs.TrySetException(error);

    public void OnNext(T value) =>
        _tcs.TrySetResult((true, value));

    public IDisposable Subscribe(IObserver<T> observer) =>
        _observable.Subscribe(observer);
}

Answer 1

这是Jeff Mercado的答案的简化版本。我认为只需完成第一个 bufferSize：1）之后，就可以简单地实现理想的行为。

实际上， asyncSubject＆lt; t＆gt; ，如@noseratio指出的，in 评论更简单，而且更有效，因为它将其单个值存储在字段中而不是数组中。

public class WriteOnceSubject<T> : ISubject<T>
{
    private readonly AsyncSubject<T> subject = new();

    public void OnNext(T value) { subject.OnNext(value); subject.OnCompleted(); }
    public void OnError(Exception error) => subject.OnError(error);
    public void OnCompleted() => subject.OnCompleted();

    public IDisposable Subscribe(IObserver<T> observer) => subject.Subscribe(observer);
}

因此，在这一系列事件中：

writeOnceSubject.OnNext(1);
writeOnceSubject.OnNext(2);
writeOnceSubject.OnNext(3);
writeOnceSubject.OnCompleted();
writeOnceSubject.OnError(new Exception());

...除第一个以外的所有命令都将是无效的。稍后订阅 writeOncesubject 时，它将发射值 1 ，该值存储在其缓冲区中，然后是 on Completed Notification。

Answer 2

您可以用两个主题来编写它，一个重播以发射值，如果设置为了散发值，另一个则可以控制初始化。

public class PromiseSubject<T> : ISubject<T>
{
    private readonly Subject<T> initialize = new();
    private readonly ReplaySubject<T> subject = new(1);
    public PromiseSubject() => initialize.Subscribe(subject);
    
    public void OnCompleted() => initialize.OnCompleted();
    public void OnError(Exception error) => initialize.OnError(error);
    public void OnNext(T value)
    {
        initialize.OnNext(value);
        initialize.OnCompleted();
    }

    public IDisposable Subscribe(IObserver<T> observer) => subject.Subscribe(observer);
}

Answer 3

您所描述的听起来与 WriteOnceBlock＆lt; t＆gt; 来自TPL DataFlow库。数据流块具有方便的扩展方法 Asobervable ，因此，基于此想法，实现将看起来像这样：

public class WriteOnceSubject<T> : ISubject<T>
{
    private readonly WriteOnceBlock<T> _block = new WriteOnceBlock<T>(x => x);

    public void OnCompleted() => _block.Complete();
    public void OnError(Exception error) => ((ISourceBlock<T>)_block).Fault(error);
    public void OnNext(T value) => _block.Post(value);

    public IDisposable Subscribe(IObserver<T> observer)
        => _block.AsObservable().Subscribe(observer);
}

不幸的是，这个想法不起作用。 writeOncesubject＆lt; t＆gt; 的订阅者仅获得 oncompleted（） Notification。没有 OnNext（）发射。我刚刚发布了意外地表现出Asobservable”> bug Report bug Report >在Github上有关此问题。

---

更新：此处是 Microsoft的反馈史蒂芬·图布（Stephen Toub）的错误报告：

writeOnceBlock 只有一个值，该值可以多次消耗，因此块在给出值后立即完成。 ASOBSERVABLE 检查源是否已完成，并将其视为不再有数据的指示。因此，如果您在将数据传递给 writeNceBlock 之前订阅观察者，则 writeNceBlock 将尽职尽责地将这些数据传播到完成之前的链接目标，并且观察者将接收到它，但是如果观察者在 writeNceBlock 完成后订阅，则将假定没有数据来完成，并且它本身会发出信号完成。

有可能从 Asobervable 中删除这些检查，如果源已经完成，则可能会以某种费用，但现在 wrighonceblock 与 Asobservable 不是完美的。