Should I expose IObservable<T> on my interfaces?

Question

My colleague and I have dispute. We are writing a .NET application that processes massive amounts of data. It receives data elements, groups subsets of them them into blocks according to some criterion and processes those blocks.

Let's say we have data items of type Foo arriving some source (from the network, for example) one by one. We wish to gather subsets of related objects of type Foo, construct an object of type Bar from each such subset and process objects of type Bar.

One of us suggested the following design. Its main theme is exposing IObservable<T> objects directly from the interfaces of our components.

// ********* Interfaces **********
interface IFooSource
{
    // this is the event-stream of objects of type Foo
    IObservable<Foo> FooArrivals { get; }
}

interface IBarSource
{
    // this is the event-stream of objects of type Bar
    IObservable<Bar> BarArrivals { get; }
}

/ ********* Implementations *********
class FooSource : IFooSource
{
    // Here we put logic that receives Foo objects from the network and publishes them to the FooArrivals event stream.
}

class FooSubsetsToBarConverter : IBarSource
{
    IFooSource fooSource;

    IObservable<Bar> BarArrivals
    {
        get
        {
            // Do some fancy Rx operators on fooSource.FooArrivals, like Buffer, Window, Join and others and return IObservable<Bar>
        }
    }
}

// this class will subscribe to the bar source and do processing
class BarsProcessor
{
    BarsProcessor(IBarSource barSource);
    void Subscribe(); 
}

// ******************* Main ************************
class Program
{
    public static void Main(string[] args)
    {
        var fooSource = FooSourceFactory.Create();
        var barsProcessor = BarsProcessorFactory.Create(fooSource) // this will create FooSubsetToBarConverter and BarsProcessor

        barsProcessor.Subscribe();
        fooSource.Run(); // this enters a loop of listening for Foo objects from the network and notifying about their arrival.
    }
}

The other suggested another design that its main theme is using our own publisher/subscriber interfaces and using Rx inside the implementations only when needed.

//********** interfaces *********

interface IPublisher<T>
{
    void Subscribe(ISubscriber<T> subscriber);
}

interface ISubscriber<T>
{
    Action<T> Callback { get; }
}


//********** implementations *********

class FooSource : IPublisher<Foo>
{
    public void Subscribe(ISubscriber<Foo> subscriber) { /* ...  */ }

    // here we put logic that receives Foo objects from some source (the network?) publishes them to the registered subscribers
}

class FooSubsetsToBarConverter  : ISubscriber<Foo>, IPublisher<Bar>
{
    void Callback(Foo foo)
    {
        // here we put logic that aggregates Foo objects and publishes Bars when we have received a subset of Foos that match our criteria
        // maybe we use Rx here internally.
    }

    public void Subscribe(ISubscriber<Bar> subscriber) { /* ...  */ }
}

class BarsProcessor : ISubscriber<Bar>
{
    void Callback(Bar bar)
    {
        // here we put code that processes Bar objects
    }
}

//********** program *********
class Program
{
    public static void Main(string[] args)
    {
        var fooSource = fooSourceFactory.Create();
        var barsProcessor = barsProcessorFactory.Create(fooSource) // this will create BarsProcessor and perform all the necessary subscriptions

        fooSource.Run();  // this enters a loop of listening for Foo objects from the network and notifying about their arrival.
    }
}

Which one do you think is better? Exposing IObservable<T> and making our components create new event streams from Rx operators, or defining our own publisher/subscriber interfaces and using Rx internally if needed?

Here are some things to consider about the designs:

• In the first design the consumer of our interfaces has the whole power of Rx at his/her fingertips and can perform any Rx operators. One of us claims this is an advantage and the other claims that this is a drawback.

• The second design allows us to use any publisher/subscriber architecture under the hood. The first design ties us to Rx.

• If we wish to use the power of Rx, it requires more work in the second design because we need to translate the custom publisher/subscriber implementation to Rx and back. It requires writing glue code for every class that wishes to do some event processing.

Answer 1

Exposing IObservable<T> does not pollute the design with Rx in any way. In fact the design decision is the exact same as pending between exposing an old school .NET event or rolling your own pub/sub mechanism. The only difference is that IObservable<T> is the newer concept.

Need a proof? Look at F# which is also a .NET language but younger than C#. In F# every event derives from IObservable<T>. Honestly, I see no sense in abstracting a perfectly suitable .NET pub/sub mechanism - that is IObservable<T> - away with your homegrown pub/sub abstraction. Just expose IObservable<T>.

Rolling your own pub/sub abstraction feels like applying Java patterns to .NET code to me. The difference is, in .NET there has always been great framework support for the Observer pattern and there is simply no need to roll your own.

Answer 2

First of all, it's worth noting that IObservable<T> is part of mscorlib.dll and the System namespace, and thus exposing it would be somewhat equivalent to exposing IComparable<T> or IDisposable. Which is equivalent to picking .NET as your platform, which you seem to have done already.

Now, instead of suggesting an answer, I want to suggest a different question, and then a different mindset, and I hope (and trust) that you'll manage from there.

You're basically asking: Do we want to promote scattered use of Rx operators all across our system?. Now obviously that's not very inviting, seeing as you probably conceptually treat Rx as a 3rd party library.

Either way, the answer doesn't lie in the basal designs you two proposed, but in the users of those designs. I recommend breaking your design down to abstraction levels, and making sure that the use of Rx operators is scoped in just one level. When I talk about abstraction levels, I mean something similar to the OSI Model, only in the same application's code.

The most important thing, in my book, is to not take the design standpoint of "Let's create something that's going to be used and scattered all across the system, and so we need to make sure we do it just once and just right, for all the years to come". I'm more of a "Let's make this abstraction layer produce the minimal API necessary for other layers to currently achieve their goals".

About the simplicity of both of your designs, it's actually hard to judge since Foo and Bar don't tell me much about use cases, and hence readability factors (which are, by the way, different from one use case to another).

Answer 3

In the first design the consumer of our interfaces has the whole power of Rx at his/her fingertips and can perform any Rx operators. One of us claims this is an advantage and the other claims that this is a drawback.

I would agree with the availability of Rx as an advantage. Listing some reasons why it is a drawback could help with determining how to address them. Some advantages I see are:

• As Yam and Christoph both brushed against, IObservable/IObserver is in mscorlib as of .NET 4.0, so it will (hopefully) become a standard concept that everyone will immediately understand, like events or IEnumerable.
• The operators of Rx. Once you need to compose, filter, or otherwise manipulate potentially multiple streams, these becomes very helpful. You will probably find yourself redoing this work in some form with your own interfaces.
• The contract of Rx. The Rx library enforces a well-defined contract and does as much of the enforcing of that contract as it can. Even when you need to make your own operators, Observable.Create will do the work to enforce the contract (which is why implementing IObservable directly is not recommended by the Rx team).
• The Rx library has good ways to ensure you end up on the right thread when needed.

I've written my share of operators where the library doesn't cover my case.

The second design allows us to use any publisher/subscriber architecture under the hood. The first design ties us to Rx.

I fail to see how the choice to expose Rx has much, if any, influence on how you implement the architecture under the hood any more than using your own interfaces would. I would assert that you should not be inventing new pub/sub architectures unless absolutely necessary.

Further, the Rx library may have operators that will simplify the "under the hood" parts.

If we wish to use the power of Rx, it requires more work in the second design because we need to translate the custom publisher/subscriber implementation to Rx and back. It requires writing glue code for every class that wishes to do some event processing.

Yes and no. The first thing I would think if I saw the second design is: "That's almost like IObservable; let's write some extension methods to convert the interfaces." The glue code is written once, used everywhere.

The glue code is straightforward, but if you think you will use Rx, just expose IObservable and save yourself the hassle.

Further Considerations

Basically, your alternate design differs in 3 key ways from IObservable/IObserver.

1. There is no way to unsubscribe. This may just be an oversight when copying to the question. If not, it's something to strongly consider adding if you go that route.
2. There is no defined path for errors to flow downstream (eg IObserver.OnError).
3. There is no way to indicate the completion of a stream (eg IObserver.OnCompleted). This is only relevant if your underlying data is intended to have a termination point.

Your alternate design also returns the callback as an action rather than having it as a method on the interface, but I don't think the distinction is important.

The Rx library encourages a functional approach. Your FooSubsetsToBarConverter class would be better suited as an extension method to IObservable<Foo> that returns IObservable<Bar>. This reduces clutter slightly (why make a class with one property when a function will do fine) and fits better with the chain-style composition of the rest of the Rx library. You could apply the same approach to the alternate interfaces, but without the operators to help, it may be more difficult.