将 Rx 与类型化消息代理一起使用

Question

我有一个类似于 Caliburn 提供的类型化消息代理：

public interface IMessageBroker
{
    void Publish<T>(T message);
    IDisposable Subscribe<T>(Action<T> subscriber);
}

How can I to conversion subscriptions to IObservable?

我想要一个扩展方法，如下所示：

public static IObservable<T> Subscribe<T>(this IMessageBroker messageBroker)
{
    var subject = new Subject<T>();
    messageBroker.Subscribe<T>(subject.OnNext);
    return subject;
}

此实现中的问题是我无法取消订阅，因此会泄漏。

也欢迎为 Subscribe 方法提供更好的名称。

Answer 1

试试这个（未经测试）：

public static IObservable<T> ToObservable<T>(this IMessageBroker messageBroker)
{
    IObservable<T> observable = Observable.CreateWithDisposable<T>(o =>
        {
            return messageBroker.Subscribe<T>(o.OnNext);
        });
    return observable;
}

您应该能够像这样使用：

var observableBroker = messageBroker.ToObservable<int>();
var subject = new Subject<int>();
observableBroker.Subscribe(subject.OnNext);

//alternatively, there are overloads of Observerable.Subscribe which take lambdas:
observableBroker.Subscribe(t => DoSomethingWith(t));

Answer 2

试试这个（已测试）

如何将订阅转换为 IObservable？

您可以通过创建以下扩展方法来使用 Observable.Create：

public static IObservable<T> AsObservable<T>(this IMessageBroker messageBroker)
{
    return Observable.Create<T>(observer => messageBroker.Subscribe<T>(observer.OnNext));
}

注意：System.Reactive nuget 包没有 Observable.CreateWithDisposable

或没有 Rx （为什么？可能不需要依赖项）：

public static IObservable<T> AsObservable<T>(this IMessageBroker messageBroker)
{
    return new DelegateObservable(observer => messageBroker.Subscribe<T>(observer.OnNext));
}

public class DelegateObservable<T> : IObservable<T>
{
    private Func<IObserver<T>, IDisposable> subscriber;

    public DelegateObservable(Func<IObserver<T>, IDisposable> subscriber)
    {
        this.subscriber = subscriber;
    }

    public IDisposable Subscribe(IObserver<T> observer)
    {
        return this.subscriber(observer);
    }
}