publish-merge.java

    @Override
    public ObservableSource<C> apply(final Observable<A> x) {
        Objects.requireNonNull(x, "x is null");
        return Observable.create(new ObservableOnSubscribe<C>() {

            transient volatile Optional<A> last = Optional.empty();

            private void sendPrevious(@NonNull final ObservableEmitter<C> e) {
                if (last.isPresent()) {
                    e.onNext(last.get());
                }
            }

            @Override
            public void subscribe(@NonNull final ObservableEmitter<C> e) throws Exception {
                x.subscribe(next -> {
                    sendPrevious(e);
                    last = Optional.of(next);
                }, error -> {
                    sendPrevious(e);
                    e.onError(error);
                }, () -> {
                    sendPrevious(e);
                    if (last.isPresent()) {
                        f.apply(last.get()).subscribe(e::onNext, e::onError, e::onComplete);
                    } else {
                        e.onComplete();
                    }
                });
            }
        });
    }