CODEWITHSUNDEEP
c#.netsystem.reactive
Daniel Little
Daniel Little

Reputation: 17263

Reactive Extensions converting from IObservable to IEnumerable

Given an Observable that will start pushing events once Subscribe is called. How can the sequence be converted into an IEnumerable.

In 2010 Jon Skeet wrote about his first impressions of RX and wrote this, which would be equivalent to what I need.

public static List<T> ToList<T>(this IObservable<T> source) 
{ 
    List<T> ret = new List<T>(); 
    source.Subscribe(x => ret.Add(x)); 
    return ret; 
}

https://codeblog.jonskeet.uk/2010/01/16/first-encounters-with-reactive-extensions/

While this solution would work it seems like there should be a better way.

What we're really interested in is the ability to await the observable in a way that behaves similarly to Subscribe. 

await observable.ToList()

However, this doesn't seem to work for me.

I've also looked into these two methods

observable.ToEnumerable
observable.Next

A little more information

I plan to use this to write some unit tests.

  1. I'm using a ReplaySubject and performing a sequence of actions with a test substitute.
  2. Calls to the substitute are recorded via OnNext.
  3. I can get the list of calls and Assert they are correct.

Upvotes: 2

Views: 2673

Answers (2)

Lee Campbell
Lee Campbell

Reputation: 10783

Interestingly Mr Skeet's implementation is pretty woeful. It doesn't block so will only pump values into the list if the source happens to be a replay subject (or similar). So for most Observable Sequences, this would just return an empty list. I am sure Jon knows that, but maybe got lost in translation.

The ToList() is what you want, so I assume that perhaps you are not completing? If you only want a known number of elements, then Take(x).ToList() will suffice instead.

Upvotes: 3

Shlomo
Shlomo

Reputation: 14350

EDIT:

If you're unable to complete the stream as described below, then using await will wait interminably. I would test in some other manner. You can look at Nuget Package Microsoft.Reactive.Testing, which contains some powerful testing tools. Or just using Subscribe and asserting in there.

await awaits for the observable to finish or error out. You probably just need to add an OnCompleted call to your ReactiveSubject and you should be fine. The following code works for me:

var rs = new ReplaySubject<int>();

rs.OnNext(1);
rs.OnNext(2);
rs.OnNext(3);
rs.OnCompleted();
var l = await rs.ToList();

//l is equivalent to  new List<int>() {1, 2, 3};

Upvotes: 2

Related Questions