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));
      }
    });
  }

}

I have an (incredibly unlikely) edge case for you.

1. Use the same transformer twice in quick succession, on different threads. Could be a thread pool Scheduler like io() I guess.
2. First one (A) terminates while the second (B) is being processed by the transformer.
3. (A) triggers clearPending(), while (B) has just failed the compareAndSet on line 28.
4. (A) now clears the AtomicReference from line 40, before (B) evaluates the get() on line 31.
5. (B) will now return null.

To address it while still avoiding locking, I'd suggest something like this to replace lines 28-31:

Observable<T> current;
do {
    cache.compareAndSet(null, sharable);
    current = cache.get();
} while (current == null);
return current;

With the above, it's impossible to return null. It's highly unlikely to loop, but if it does it's certainly not going to many times. I'm basing this on the sort of thing AtomicReference does with getAndSet, but the emphasis here is on returning a non-null value.

Whaddya think? Am I missing something obvious?

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