RxJava отказывается от одного наблюдателя от другого

Два потока не нужны. Просто используйте один. Типичная конструкция будет включать Observable.using() для создания сокета, установления наблюдаемого в этом сокете и обработки закрытия сокета, а затем связывания этого с retry() (обычно с задержкой).

 return Observable.using(new Func0<XMPPTCPConnection>() {
        @Override
        public XMPPTCPConnection call() {
            L.log("creating connection");
            connection = _createConnection();
            return connection;
        }
    }, new Func1<XMPPTCPConnection, Observable<?>>() {
        @Override
        public Observable<?> call(final XMPPTCPConnection connection) {

            try {
                _authenticate(connection, username, password);
            } catch (Exception e) {
                throw new RuntimeException(e);
            }

            return Observable.just(connection)
                    .repeatWhen(new RepeatWhenOperator(-1, 1000))
                    .map(new Func1<XMPPTCPConnection, Object>() {
                        @Override
                        public Object call(XMPPTCPConnection connection) {
                            if (!connection.isConnected()) {
                                throw new RuntimeException("Disconnected");
                            }

                            return null;
                        }
                    });
        }
    }, new Action1<XMPPTCPConnection>() {
        @Override
        public void call(XMPPTCPConnection connection) {
            L.log("disposing");
            _disconnect();
        }
    })
            .subscribeOn(Schedulers.io())
            .unsubscribeOn(Schedulers.io())
            .retry();

Итак, на основе Observable.using() я придумал это решение. Это работает, но я не доволен решением - то есть опрос каждую секунду, живо ли соединение или нет. XMPPTCPConnection предоставляет мне прослушиватель, когда соединение обрывается, это кажется лучшим решением - просто не уверен, как это включить...

Метод _disconnect() на самом деле предназначен для отключения и удаления; это может быть плохим именем в текущих обстоятельствах, но отключение, которое вызывает исключение RuntimeException, происходит за пределами этой системы.

Любые идеи или улучшения были бы очень признательны!