PendingObservableReplayer.java

import java.util.concurrent.atomic.AtomicReference;
import rx.Observable;
import rx.Subscriber;
import rx.functions.Action0;
import rx.functions.Func0;

/**
 * This custom {@link Observable.Transformer} can be used to replay an in flight {@link Observable}
 * source among all {@link Subscriber}s subscribed before its termination.
 */
class PendingObservableReplayer<T> implements Observable.Transformer<T, T> {

  private final AtomicReference<Observable<T>> pendingObservable;

  PendingObservableReplayer() {
    this.pendingObservable = new AtomicReference<>(null);
  }

  @Override
  public Observable<T> call(final Observable<T> source) {
    return Observable.defer(new Func0<Observable<T>>() {
      @Override
      public Observable<T> call() {
        Observable<T> sharable = source
            .doOnTerminate(clearPending())
            .replay()
            .autoConnect();
        if (pendingObservable.compareAndSet(null, sharable)) {
          return sharable;
        }
        return pendingObservable.get();
      }
    });
  }

  private Action0 clearPending() {
    return new Action0() {
      @Override
      public void call() {
        pendingObservable.set(null);
      }
    };
  }
}

PendingObservableReplayerTest.java

import java.util.Random;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.ArgumentCaptor;
import org.mockito.Captor;
import org.mockito.Mock;
import org.mockito.runners.MockitoJUnitRunner;
import rx.Observable;
import rx.Observer;
import rx.functions.Action0;
import rx.functions.Func0;

import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;

@RunWith(MockitoJUnitRunner.class)
public class PendingObservableReplayerTest {

  @Mock private Action0 sourceSubscribed;
  @Mock private Observer<Integer> observer1;
  @Mock private Observer<Integer> observer2;
  @Captor private ArgumentCaptor<Integer> captor1;
  @Captor private ArgumentCaptor<Integer> captor2;
  private Random random;

  private PendingObservableReplayer<Integer> replayer;

  @Before
  public void setUp() {
    random = new Random();
    replayer = new PendingObservableReplayer<>();
  }

  @Test
  public void shouldSubscribeToOriginalSourceOnlyOnceWhenSubscribedTwiceBeforeTermination() {
    Observable<Integer> source = random()
        .concatWith(Observable.<Integer>never())
        .doOnSubscribe(sourceSubscribed)
        .compose(replayer);

    source.subscribe();
    source.subscribe();

    verify(sourceSubscribed).call();
  }

  @Test
  public void shouldReceiveSameEmissionsFromOriginalSourceWhenSubscribedTwiceBeforeTermination() {
    Observable<Integer> source = random()
        .concatWith(random())
        .concatWith(Observable.<Integer>never())
        .compose(replayer);

    source.subscribe(observer1);
    source.subscribe(observer2);

    verify(observer1, times(2)).onNext(captor1.capture());
    verify(observer2, times(2)).onNext(captor2.capture());
    assertThat(captor1.getAllValues()).isEqualTo(captor2.getAllValues());
  }

  @Test
  public void shouldReceiveDifferentEmissionsFromOriginalSourceWhenSubscribedTwiceAfterTermination() {
    Observable<Integer> source = random()
        .concatWith(random())
        .compose(replayer);

    source.subscribe(observer1);
    source.subscribe(observer2);

    verify(observer1, times(2)).onNext(captor1.capture());
    verify(observer2, times(2)).onNext(captor2.capture());
    assertThat(captor1.getAllValues()).isNotEqualTo(captor2.getAllValues());
  }

  private Observable<Integer> random() {
    return Observable.defer(new Func0<Observable<Integer>>() {
      @Override
      public Observable<Integer> call() {
        return Observable.just(random.nextInt(1000));
      }
    });
  }

}

@jonreeve aparently this is happening as you mentioned! I'll try your workaround and see if it fixes the issue