how to convert IEnumerable to IObservable?

问题:

I'm looking to convert an IEnumerable collection to an IObservable one without using Rx ToObservable() and ToEnumerable() methods.

回答1:

Simple answer - use ToObservable. That's what it's for.

"Answering the actual question" answer - you can avoid using Subjects via Observable.Create:

void Main()
{
    var src = Enumerable.Range(0, 10);
    var observable = Observable.Create<int>(obs =>
    {
        foreach(var item in src)
        {
            obs.OnNext(item);
        }
        return Disposable.Create(()=>{});
    });

    using(observable.Subscribe(Console.WriteLine))
    {
        Console.ReadLine();
    }
}

Output:

0
1
2
3
4
5
6
7
8
9

回答2:

If one has an IEnumerable of unknown type, there are two ways one can "convert it to IObservable":

• Copy all of the data in the object to a new collection which implements IObservable. If this is done, only changes made to the new collection will be reported. Changes made to the original will not.

• Create a new object which will periodically take snapshots of the contents of the IEnumerable; after taking each snapshot, report all the changes that would have to be made to the previous snapshot to make it match the new one. Using this approach, changes made to the original object will be observed, eventually, but it's hard to provide timely update notifications without wasting a lot of time repeatedly reading the collection when nothing has changed.

There are some times when one needs to have the IObservable bound to the original IEnumerable object, rather than to a new object which is pre-filled with a copy of the data; in such cases, the second approach may be necessary. Often, however, it won't be possible to make the polling rate fast enough to provide timely updates without it causing unacceptable system loading unless the original collection supports features which are not available in an arbitrary IEnumerable. Further, if one doesn't impose requirements on the type of the IEnumerable, one will likely have to pose restrictions on the threading contexts where it might be updated.