代码之家 › 专栏 › 技术社区 › Konstantin Spirin

使用带类型化消息代理的RX

system.reactive .net

Konstantin Spirin · 技术社区 · 14 年前

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

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

如何将订阅转换为iobservable?

我想要一个扩展方法,类似于:

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

这个实现中的问题是我不能取消订阅,所以它泄漏了。

订阅方法的更好名称也受到欢迎。

2 回复 | 直到 14 年前

Benjol 14 年前

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));

Ahmed Alejo 8 年前

Observable.Create

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

System.Reactive Observable.CreateWithDisposable

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);
    }
}

推荐文章

Nicolas · c#/中多个Rx受试者的依赖注入。净核心

7 年前

user584018 · 如何获取文件的增量内容

7 年前

SuperJMN · 如何计算LINQ管道要完成的ETA?

7 年前

Seishin Okigaru · RXSwift,检测到重入异常

7 年前

user584018 · 这段代码是否适用于使用Rx核心的UdpClient

7 年前

Umair · IObservable收集当前流中的所有元素并执行批量操作

7 年前

marsop · 在typescript中观察集合

7 年前

thk · Rxjs init resource once with hot observable

7 年前

Wouter · Rx TestScheduler不能很好地处理返回和延迟

7 年前

gakera · 如何使用反应式用户界面触发按钮点击与按钮按下(按住)后的不同操作

7 年前