接收如何从发布/订阅模式创建序列 [英] Rx how to create a sequence from a pub/sub pattern

问题描述

我试图用接收来创建一个发布/订阅模式的序列（其中一个元素是由生产商（S）公布，即经典的观察者模式）来评价。这基本上是一样的。NET的事件，但我们要概括它，这样有一个事件是不是必需的，所以我不能够采取Observable.FromEvent的优势。我已经与Observable.Create和Observable.Generate发挥四周，发现自己最终不得不写code采取发布/订阅的护理（即我有写生产者/消费者code藏匿发布项目，然后使用它通过调用IObserver.OnNext（）与它），这样好像我真的不利用接收的...

I'm trying to evaluate using Rx to create a sequence from a pub/sub pattern (i.e. classic observer pattern where next element is published by the producer(s)). This is basically the same as .net events, except we need to generalize it such that having an event is not a requirement, so I'm not able to take advantage of Observable.FromEvent. I've played around with Observable.Create and Observable.Generate and find myself end up having to write code to take care of the pub/sub (i.e. I have to write producer/consumer code to stash the published item, then consume it by calling IObserver.OnNext() with it), so it seems like I'm not really taking advantage of Rx...

上午我低头正确的路径，或者这是一个非常适合接收？

Am I looking down the correct path or is this a good fit for Rx?

感谢

推荐答案

您的出版商只是暴露了一些 IObservables 的属性。而你的消费者只需订阅他们（或认购前，做任何接收复他们想要的）。

Your publisher just exposes some IObservables as properties. And your consumers just Subscribe to them (or do any Rx-fu they want before subscribing).

有时候，这很简单，只要在你的发布者使用主题。有时它是更复杂，因为你的发行实际上是观察一些其他可观察的过程。

Sometimes this is as simple as using Subjects in your publisher. And sometimes it is more complex because your publisher is actually observing some other observable process.

下面是一个愚蠢的例子：

Here is a dumb example:

public class Publisher
{
    private readonly Subject<Foo> _topic1;

    /// <summary>Observe Foo values on this topic</summary>
    public IObservable<Foo> FooTopic
    {
       get { return _topic1.AsObservable(); }
    }

    private readonly IObservable<long> _topic2;

    /// <summary>Observe the current time whenever our clock ticks</summary>
    public IObservable<DateTime> ClockTickTopic
    {
        get { return _topic2.Select(t => DateTime.Now); }
    }

    public Publisher()
    {
         _topic1 = new Subject<Foo>();
         // tick once each second
         _topic2 = Observable.Interval(TimeSpan.FromSeconds(1));
    }

    /// <summary>Let everyone know about the new Foo</summary>
    public NewFoo(Foo foo) { _topic1.OnNext(foo); }
}


// interested code...
Publisher p = ...;
p.FooTopic.Subscribe(foo => ...);

p.ClickTickTopic.Subscribe(currentTime => ...);

// count how many foos occur during each clock tick
p.FooTopic.Buffer(p.ClockTickTopic)
    .Subscribe(foos => Console.WriteLine("{0} foos during this interval", foos.Count));

这篇关于接收如何从发布/订阅模式创建序列的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！