我可以使用 Reactive Extensions 进行独立订阅/发布吗？

Question

假设我有一些 MyClass 类。在我的代码的一部分中，我想要这样的东西：

Observable.Subscribe<MyClass>(myClass => DoSomething(myClass));

然后在另一个地方（文件/项目/时间）我有这样的东西：

Observable.Publish(instanceOfMyClass);

第二行会触发所有使用该确切类类型订阅的方法。这是响应式扩展（v1 或 v2）支持的东西吗？

将 SynchronizationContext 指定为订阅调用的一部分会很有用。在那里指定方法是否应该使用 Wea​​kReference 也是很好的。并且 Publish 方法应该能够同步完成这一切，或者给我一些可以等待的东西。

Answer 1

这并不难创建。

您只需要一个内部的Dictionary<Type, Object> 并使用它来存储Type 的每个Subject<T>（作为对象）。

然后您可以编写两个 Subscribe 和 Publish 方法来处理内部字典。

其实应该很简单。

---

与其说简单，我还想试试看。

这是我的 Rx Pub/Sub 类：

public static class RxPS
{
    private static Dictionary<Type, object> _subjects
        = new Dictionary<Type, object>();

    public static IDisposable Subscribe<T>(Action<T> observer)
    {
        lock(_subjects)
        {
            if (!_subjects.ContainsKey(typeof(T)))
            {
                _subjects.Add(typeof(T), new Subject<T>());
            }
            return (_subjects[typeof(T)] as Subject<T>)
                .Subscribe(observer);
        }
    }

    public static void Publish<T>(T item)
    {
        lock(_subjects)
        {
            if (_subjects.ContainsKey(typeof(T)))
            {
                (_subjects[typeof(T)] as Subject<T>)
                    .OnNext(item);
            }
        }
    }
}

这就是它的使用方式：

RxPS.Publish(1);
var d = RxPS.Subscribe<int>(x => Console.WriteLine(x)); 
RxPS.Publish(2);
d.Dispose();
RxPS.Publish(3);

结果是这段代码只会将2写入控制台。

享受吧！

Answer 2

我相信您正在寻找像 ReactiveUI 的 MessageBus class 这样的东西。这个类使用 Rx 来实现发布/订阅模型，基本上只需要一个 Dictionary of Type => IObservables。