gt gt Rx is sweet

Niner since 2004

I am the Architect. I created the Matrix. I've been waiting for you. You have many questions, and although the process has altered your consciousness, you remain irrevocably human. Ergo, some of my answers you will understand, and some of them you will not. Concordantly, while your first question may be the most pertinent, you may or may not realize it is also the most irrelevant.


  • Rx Workshop: Programming the Cloud

    @dmarsh: good point - compiler could figure out serializable & mbr types

    Wonder why the remote queries weren't using the QObservable "provider" where expressions & values were remoted rather than type references...

  • Kim Hamilton and Wes Dyer: Inside .NET Rx and ​IObservable/​IObserver in the BCL (VS 2010)

    Yes, you can unsubscribe. You don't need to use another thread, but given the ToObservable() method you do.

    public static IObservable<T> ToObservable<T>(this IEnumerable<T> xs) {
     return new Observable<T> () { Enumerable = xs };

    Here's the internal class needed to handle subscription:

    internal sealed class Observable<T> : IObservable<T> {
    private readonly UnSubscribe _unsubscribe = new UnSubscribe (); 
    public IEnumerable<T> Enumerable { get; set; } 
    public IObserver<T> Observer { get; set; } 
    public IDisposable Subscribe(IObserver<T> o) { 
    Observer = o; 
    return _unsubscribe; 
    private void Worker (object state) { 
    foreach (T item in Enumerable) {
     if (_unsubscribe.Cancel)
    Observer.OnNext (item); 
     Observer.OnCompleted ();

    And here's the IDisposable internal class that flags when the subscriber is no longer interested:

    internal sealed class UnSubscribe : IDisposable {
     public bool Cancel { get; set; }
     public void Dispose() {
     Cancel = true;