title: rxJava. Help to understand how publish and unsubscribe work
body: <p>I observe strange behavior when using <code>publish()</code> in conjunction with <code>observeOn</code> and <code>subscribeOn</code>. Please take a look at the folowing examples.</p>
  
  <p>Code:</p>
  
  <pre><code>ConnectableObservable&lt;String&gt; observable = Observable.create(
      new Observable.OnSubscribe&lt;String&gt;() {
          @Override
          public void call(Subscriber&lt;? super String&gt; subscriber) {
              int i=0;
              Log.d("testTag:", "start call");
              while (!subscriber.isUnsubscribed()) {
                  subscriber.onNext("item "+i);
                  i++;
              }
              Log.d("testTag:", "completed");
              subscriber.onCompleted();
          }
      }
  ).publish();
  
  observable
      .take(10)
      .subscribe(
          new Action1&lt;String&gt;() {
              @Override
              public void call(String s) {
                  Log.d("testTag:", "item received 1 : " + String.valueOf(s));
              }
          });
  
  observable.connect();
  </code></pre>
  
  <p>Output:</p>
  
  <pre><code>start call
  item received 1 : item 0
  item received 1 : item 1
  item received 1 : item 2
  item received 1 : item 3
  item received 1 : item 4
  item received 1 : item 5
  item received 1 : item 6
  item received 1 : item 7
  item received 1 : item 8
  item received 1 : item 9
  </code></pre>
  
  <p>Condition <code>!subscriber.isUnsubscribed()</code> never happens. This is totally fine. Strange thing happens when I add <code>observeOn(Schedulers.newThread())</code> and <code>subscribeOn(Schedulers.newThread())</code>. Take a look:</p>
  
  <pre><code>ConnectableObservable&lt;String&gt; observable = Observable.create(
      new Observable.OnSubscribe&lt;String&gt;() {
          @Override
          public void call(Subscriber&lt;? super String&gt; subscriber) {
              int i=0;
              Log.d("testTag:", "start call");
              while (!subscriber.isUnsubscribed()) {
                  subscriber.onNext("item "+i);
                  i++;
              }
              Log.d("testTag:", "completed");
              subscriber.onCompleted();
          }
      }
  )
  .observeOn(Schedulers.newThread())
  .subscribeOn(Schedulers.newThread())
  .publish();
  
  observable
      .take(10)
      .subscribe(
          new Action1&lt;String&gt;() {
              @Override
              public void call(String s) {
                  Log.d("testTag:", "item received 1 : " + String.valueOf(s));
              }
          });
  
  observable.connect();
  </code></pre>
  
  <p>Output:</p>
  
  <pre><code>start call
  item received 1 : item 0
  item received 1 : item 1
  item received 1 : item 2
  item received 1 : item 3
  item received 1 : item 4
  item received 1 : item 5
  item received 1 : item 6
  item received 1 : item 7
  item received 1 : item 8
  item received 1 : item 9
  completed
  </code></pre>
  
  <p>Please help me understand why condition <code>!subscriber.isUnsubscribed()</code> become <code>true</code>.</p>
  
  <p>PS. I understand that if I want check <code>!subscriber.isUnsubscribed()</code> I should use <code>.publish().refCount()</code> instead of <code>connect()</code>. My goal is understand current behavior.</p>