当我需要 SelectMany observables 时，我是否以正确的方式思考？

Question

所以我可能会对以下代码提出几个问题，可能是因为我使用的是Subject，我不确定它是否可以使用。

这对我来说很难解释，我不知道为什么，也许我缺少术语。

我想要做的是拥有一种可以注册多个可观察对象的类型，因此我可以传递这种类型并将所有可观察对象分组并从中公开一个可观察对象。

所以我的第一个问题我觉得我没有以正确的方式思考事情，我想知道这是正确的还是有更“被动”的方式来做这件事？

我的意思是我有这种类型，您可以使用它注册 observables，而且这种类型会公开我也可以订阅的单个 Observable。

我将尝试用下面的代码示例来解释一下。 因此，SomeTypeWithObservable 可能是暴露IObservable<SomeEvents> 的众多类型之一 ReactiveTesting 类型是尝试将所有可观察对象组合在一起并公开单个 IObservable<SomEvents> 的类型。有一个RegisterObservable 方法会将其发送到内部Subject<IObservable<SomeEvents>>。构造函数将我想要公开的 Observable 设置为该主题的 SelectMany。

使用下面的实现，在ReactiveTesting 构造函数中，我执行SelectMany.Publish.RefCount，然后使用虚拟订阅来启动可观察对象，我注意到如果我不使用虚拟订阅，则不会使用可观察对象的注册。

所以我的第二个问题是该代码可以有一个虚拟订阅来启动 observable，或者我应该做我在评论中所做的事情，我刚刚从 Publish 获得了可连接，然后之后立即连接，或者他们都错了，在这种情况下，有人能指出我正确的方向吗？

我的第三个问题我应该使用一个主题吗？

如果我在订阅之前调用RegisterObservable，并且如果我不将虚拟Subscribe 或Connect 放入，那么我将不会观察任何触发的事件。

我的第四个问题 有人可以解释一下后者吗？ 我有点想，因为它是 Publish 和 RefCount，所以在订阅从 Observable 开始之前什么都做。

代码

--已编辑 - 为了表明我想在ReactiveTesting 注册多个可观察对象

enum SomeEvents
{
    event1,
    event2,
    event3,
    event4
}

interface ISomeTypeWithObservable
{
    IObservable<SomeEvents> SomeObservableEvents { get; }
}

class SomeTypeWithObservable2 : ISomeTypeWithObservable
{
    private event EventHandler SpecialEvent;
    public SomeTypeWithObservable2()
    {
        var observableFromSpecialEvent = Observable.FromEventPattern(h => SpecialEvent += h, h => SpecialEvent -= h).Select(x => SomeEvents.event2);
        SomeObservableEvents = Observable.Create<SomeEvents>(observer =>
            {
                return observableFromSpecialEvent.Subscribe(observer);
            })
            .Publish()
            .RefCount();
    }

    public IObservable<SomeEvents> SomeObservableEvents { get; }
    public void TriggerEvent()
    {
        SpecialEvent.Invoke(this, new EventArgs());
    }
}

class SomeTypeWithObservable : ISomeTypeWithObservable
{
    private event EventHandler SpecialEvent;
    public SomeTypeWithObservable()
    {
        var observableFromSpecialEvent = Observable.FromEventPattern(h => SpecialEvent += h, h => SpecialEvent -= h).Select(x => SomeEvents.event1);
        SomeObservableEvents = Observable.Create<SomeEvents>(observer =>
            {
                return observableFromSpecialEvent.Subscribe(observer);
            })
        .Publish()
        .RefCount();
    }

    //Some code in here that will produce things to observe, maybe Observable.FromEventPattern...
    public IObservable<SomeEvents> SomeObservableEvents { get; }

    public void TriggerEvent()
    {
        SpecialEvent.Invoke(this, new EventArgs());
    }
}

class ReactiveTesting
{
    private Subject<IObservable<SomeEvents>> _innerEvents = new Subject<IObservable<SomeEvents>>();

    public IObservable<SomeEvents> AllEvents;
    public ReactiveTesting()
    {
        AllEvents = _innerEvents.SelectMany(x => x).Publish().RefCount();
        AllEvents.Subscribe(next => { }, exception => { }, () => { });

        //This instead of the above??
        //var connectableObservable = _innerEvents.SelectMany(x => x).Publish();
        //AllEvents = connectableObservable;
        //connectableObservable.Connect();
    }

    public void RegisterObservable(ISomeTypeWithObservable someTypeWithObservable)
    {
        _innerEvents.OnNext(someTypeWithObservable.SomeObservableEvents);
    }
}

class Program
{
    static void Main(string[] args)
    {
        var reactiveTesting = new ReactiveTesting();
        var someTypeWithObservable = new SomeTypeWithObservable();
        var someTypeWithObservable2 = new SomeTypeWithObservable2();
        reactiveTesting.AllEvents.Subscribe(next => Console.WriteLine(string.Format("Subscriber 1 - {0}", next.ToString("G"))));
        reactiveTesting.AllEvents.Subscribe(next => Console.WriteLine(string.Format("Subscriber 2 - {0}", next.ToString("G"))));
        reactiveTesting.RegisterObservable(someTypeWithObservable);
        reactiveTesting.RegisterObservable(someTypeWithObservable2);
        someTypeWithObservable.TriggerEvent();
        someTypeWithObservable.TriggerEvent();
        someTypeWithObservable.TriggerEvent();
        someTypeWithObservable2.TriggerEvent();
        someTypeWithObservable2.TriggerEvent();
        someTypeWithObservable2.TriggerEvent();

        Console.WriteLine("Press key...");
        Console.ReadLine();
    }
}

Answer 1

所以我的第一个问题是我觉得我没有考虑 正确的方法，我想知道这是正确的还是有更多 “反应式”的做法呢？

在ReactiveTesting 中使用Subject 通常表明您有一些命令式代码可以被消除或推得更远。它可能需要重新编写一些周围的代码。在这种情况下，你会得到这样的结果：

class ReactiveTesting
{
    public IObservable<SomeEvents> AllEvents { get; }
    public ReactiveTesting(IObservable<IObservable<SomeEvents>> eventSource)
    {
        AllEvents = eventSource.Merge().Publish().RefCount();
    }
}
class Program
{
    public static void Main(string[] args)
    {
        var someTypeWithObservable = new SomeTypeWithObservable();
        var reactiveTesting = new ReactiveTesting(Observable.Return(someTypeWithObservable.SomeObservableEvents));
        reactiveTesting.AllEvents.Subscribe(next => Console.WriteLine(string.Format("Subscriber 1 - {0}", next.ToString("G"))));
        reactiveTesting.AllEvents.Subscribe(next => Console.WriteLine(string.Format("Subscriber 2 - {0}", next.ToString("G"))));
        someTypeWithObservable.TriggerEvent();
        someTypeWithObservable.TriggerEvent();
        someTypeWithObservable.TriggerEvent();

        Console.WriteLine("Press key...");
        Console.ReadLine();

    }
}

所以我的第二个问题是该代码可以虚拟订阅 从 observable 开始，或者我应该做我在下面做的事情 评论我刚刚从 Publish 获得可连接然后连接 紧随其后，或者他们都错了，在这种情况下，可以 有人指出我正确的方向吗？

应该没有必要。它对我来说和删除它一样有效。虚拟订阅有助于.Replay().Refcount()。我不明白.Publish() 的意义所在。

我的第三个问题我应该使用主题吗？

它们基本上是一种代码味道。如果您可以消除它们，或者将它们从您的业务逻辑中推开，那么您的情况会更好。