Take the 2-minute tour ×

I have a lot of code in my project like Hit and mute by using Reactive extension like this way:

    IDisposable dsp = null;
    dsp = TargetObservable.Subscribe((incomingContent) =>
        if (incomingContent == "something")

First of all, I concerns about the thread safety since my Observable is quite busy and have bunch of content pushing all the way, but later, I was told I should combine with the ObserveOn(thread) to guarantee thread safe, I totally agree, so let's forget the thread safe thing.

Here I want to know:

  1. How or when I should call the Dispose for an observable.
  2. What's the correct way to satisfy Hit and mute, combine with some complete-able extension method like Take(count), 'TakeWhile(predict)'?
  3. If OnComplete() called, the Dispose() will be called internally, correct? Then the reference relationship between the Observer and Observable will break(because my observable is a long life static instance, the reference would cause memory leak).
share|improve this question

1 Answer 1

up vote 7 down vote accepted

I would avoid following the pattern you have here. It makes it difficult to understand the problem space if other devs have to mix global state with the inner function for the subscribe/OnNext handler.

You are much better off creating the TakeWhile/TakeUntilIncluding extension method which encapsulates the sequence termination. Then you can separate your 'adding to the list' concern.

An alternatitve thing to do is the super-simple:

var subscription = source.Where(x => x=="something")
share|improve this answer

Your Answer


