Observing both synchronous and asynchronous results

Using Rx, I want to observe a legacy object that exposes both the method GetItems and the event NewItem.

When GetItems is called, it'l开发者_StackOverflow社区l synchronously return a list of any items it has in cache. It'll also spawn an async fetch of items that will be published via the NewItem event as they are received.

How can I observe both of these sources (sync + async) in a consistent way by formulating a LINQ query such that both result-sets are captured? The production order is of no importance.

Let me see if I've understood your legacy object. I'm assuming it is a generic type that looks like this:

public class LegacyObject<T>
{
    public IEnumerable<T> GetItems();
    public event EventHandler<NewItemEventArgs<T>> NewItem;
}

With the new item event args like this:

public class NewItemEventArgs<T> : System.EventArgs
{
    public T NewItem { get; private set; }
    public NewItemEventArgs(T newItem)
    {
        this.NewItem = newItem;
    }
}

Now, I've created a .ToObservable() extension method for LegacyObject<T>:

public static IObservable<T> ToObservable<T>(
    this LegacyObject<T> @this)
{
    return Observable.Create<T>(o =>
    {
        var gate = new object();
        lock (gate)
        {
            var list = new List<T>();
            var subject = new Subject<T>();
            var newItems = Observable
                .FromEventPattern<NewItemEventArgs<T>>(
                    h => @this.NewItem += h,
                    h => @this.NewItem -= h)
                .Select(ep => ep.EventArgs.NewItem);
            var inner = newItems.Subscribe(ni =>
            {
                lock (gate)
                {
                    if (!list.Contains(ni))
                    {
                        list.Add(ni);
                        subject.OnNext(ni);
                    }
                }
            });
            list.AddRange(@this.GetItems());
            var outer = list.ToArray().ToObservable()
                .Concat(subject).Subscribe(o);
            return new CompositeDisposable(inner, outer);
        }
    });
}

This method creates a new observable for every subscriber - which is the correct thing to do when writing extension methods like this.

It creates a gate object to lock access to the internal list.

Because you said that the act of calling GetItems spawns the async function to get new items I've made sure that the NewItem subscription is created before the call to GetItems.

The inner subscription checks if the new item is in the list or not and only calls OnNext on the subject if it's not in the list.

The call to GetItems is made and the values are added to the internal list via AddRange.

There is an unlikely, but possible, chance that the items won't be added to the list before the NewItem event begins firing on another thread. This is why there is a lock around the access to the list. The inner subscription will wait until it can obtain the lock before attempting to add items to the list and this will happen after the initial items are added to the list.

Finally the internal list is turned into an observable, concatenated with the subject, and the o observer subscribes to this observable.

The two subscriptions are returned as a single IDisposable using CompositeDisposable.

And that is it for the ToObservable method.

Now, I tested this by creating a constructor on the legacy object that let me pass in both an enumerable and an observable of values. The enumerable is returned when the GetItems is called and the observable drives the NewItem event.

So, my test code looked like this:

var tester = new Subject<int>();
var legacy = new LegacyObject<int>(new [] { 1, 2, 3, }, tester);

var values = legacy.ToObservable();

values.Subscribe(v => Console.WriteLine(v));

tester.OnNext(3);
tester.OnNext(4);
tester.OnNext(4);
tester.OnNext(5);

And the values written to the console were:

1
2
3
4
5

Let me know if this meets your needs.

If I understood your question correctly, it could be done using something like this:

legacyObject.GetItems().ToObservable()
    .Merge(
        Observable.FromEventPattern<IntEventArgs>(legacyObject, "NewItem")
            .Select(e => e.EventArgs.Value));

Edit: Enigmativity spotted a race condition on the solution above (see comments below). This one hopefully solves that:

Observable.FromEventPattern<IntEventArgs>(legacyObject, "NewItem")
   .Select(e => e.EventArgs.Value)
   .Merge(Observable.Defer(() => legacyObject.GetItems().ToObservable()));

Here is the rest of my test code if it helps. I don't know if I accurately modeled your legacy class though:

class IntEventArgs : EventArgs
{
    public IntEventArgs(int value)
    {
        Value = value;
    }

    public int Value { get; private set; }
}

class Legacy
{
    public event EventHandler<IntEventArgs> NewItem;

    public IList<int> GetItems()
    {
        GenerateNewItemAsync();
        return new List<int> { 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15 };
    }

    private void GenerateNewItemAsync()
    {
        Task.Factory.StartNew(() =>
        {
            for (var i = 0; i < 100000; ++i)
            {
                var handler = NewItem;
                if (handler != null) handler(this, new IntEventArgs(i));
                Thread.Sleep(500);
            }
        });
    }
}

class Example
{
    static void Main()
    {
        var legacyObject = new Legacy();

        legacyObject.GetItems().ToObservable()
            .Merge(
                Observable.FromEventPattern<IntEventArgs>(legacyObject, "NewItem")
                    .Select(e => e.EventArgs.Value))
            .Subscribe(Console.WriteLine);

        Console.ReadLine();
    }
}