title: Sharing Observable between Activities with RxJava and unsubscribing
body: <p>I'm trying to implement a timer <code>Observable</code> that is Shared between the <code>Activities</code> of my application. I am making the implementation on a class that is a Dagger singleton which I inject in each <code>Presenter</code> of each different <code>Activity</code>. </p>
  
  <p>I create the Observable once, that way:</p>
  
  <pre><code>Observable.defer(() -&gt; Observable.timer(milliseconds, TimeUnit.MILLISECONDS).map(t -&gt; this::doSomethingCool()))
              .subscribeOn(Schedulers.newThread())
              .observeOn(AndroidSchedulers.mainThread())
              .share();
  </code></pre>
  
  <p>I make the subscribe from the Presenter with the function:</p>
  
  <pre><code>public Observable&lt;Status&gt; register(Callback callback) {
  
      PublishSubject&lt;Status&gt; subject = PublishSubject.create();
      subject.subscribe(status -&gt; {},
              throwable -&gt; L.LOGE(TAG, throwable.getMessage()),
              () -&gt; callback.onStatusChanged(mBasketStatus));
  
      mObservable.subscribe(subject);
      basketCounterCallback.onStatusChanged(status));
  
      subject.doOnUnsubscribe(() -&gt; L.LOGD(TAG, "Unsubcribed from subject!"));
      return subject.asObservable();
  }
  </code></pre>
  
  <p>I store that <code>Subject</code> as <code>Observable</code> in every presenter, and I call to:
      <code>obs.unsubscribeOn(AndroidSchedulers.mainThread())</code>
  to unsubscribe (in <code>onPause()</code> method). I also tried to unsubscribe using the Scheduler <code>Schedulers.immediate()</code></p>
  
  <p>But the callback is called anyway X times (where X is all the Presenters that I have subscribed to the timer), so it is not unsubscribing. Also the Log <code>"Unsubcribed from subject!"</code> is not getting called.</p>
  
  <p>How can I unsubscribe correctly from every subject?</p>
  
  <p>Thanks in advance</p>
  
  <p><strong>EDIT:</strong></p>
  
  <p><strong>Added more implementation details due to the comments:</strong></p>
  
  <p>This is the part where I create the Observable and store in a member of the <code>Singleton</code> class <code>StatusManager</code> (the status is also a singleton):</p>
  
  <pre><code>private Observable&lt;BasketStatus&gt; mObservable;
  private Status mStatus;
  
  public Observable&lt;BasketStatus&gt; start(long milliseconds, Status status, Callback callback) {
  
      if (mObservable == null) mObservable = createObservable(milliseconds, status);
  
      return register(callback);
  }
  
  private Observable&lt;BasketStatus&gt; createObservable(long milliseconds, Status status) {
  
      mStatus = status;
  
      return Observable.defer(() -&gt; Observable.timer(milliseconds, TimeUnit.MILLISECONDS).map(t -&gt; status.upgradeStatus()))
              .subscribeOn(Schedulers.newThread())
              .observeOn(AndroidSchedulers.mainThread())
              .share();
  }
  
  public Observable&lt;BasketStatus&gt; register(Callback callback) {
  
      PublishSubject&lt;Status&gt; subject = PublishSubject.create();
      subject.subscribe(status -&gt; {},
              throwable -&gt; L.LOGE(TAG, throwable.getMessage()),
              () -&gt; callback.onStatusChanged(mStatus));
  
      mObservable.subscribe(subject);
      callback.onStatusChanged(mStatus));
  
      subject.doOnUnsubscribe(() -&gt; L.LOGD(TAG, "Unsubcribed from subject!"));
      return subject.asObservable();
  }
  </code></pre>
  
  <p>After to call to the method <code>start(...)</code> from the <code>Presenter</code> that start the timer I call to <code>register(...)</code> method from the next Presenters:</p>
  
  <pre><code>class Presenter implements Callback {
  
      private Observable&lt;BasketStatus&gt; mRegister;
  
      @Inject
      public Presenter(Status status, StatusManager statusManager) {
          mRegister = statusManager.start(20000, status, this);
      }
  
      // Method called from onPause()
      public void unregisterFromBasketStatus() {
           mRegister.unsubscribeOn(Schedulers.immediate());
      }
  }
  </code></pre>
  
  <p>And the next presenter...</p>
  
  <pre><code>@Inject
  public NextPresenter(StatusManager statusManager) {
      mBasketStatusManager.register(this);
  }
  </code></pre>