rx-java - Rxを活用するための接続方法の書き換え

Question

ライブラリ内のメソッドを Rx を利用するように書き直しています。以下のコード例はオリジナルの方法です。

public void connect(ConnectionListener connectionListener) {

    //......

    RxBleDevice device = mRxBleClient.getBleDevice(builder.toString());

    mEstablishedConnection = device.establishConnection(mContext, false)
            .observeOn(AndroidSchedulers.mainThread())
            .doOnError(throwable -> {
                throwable.printStackTrace();
                connectionListener.onFailed();
                Log.d(TAG, "Error establishing a connection" + throwable.getMessage());
            })
            .subscribe(rxBleConnection -> {
                mConnection = rxBleConnection;
                connectionListener.onConnected();
                setNotifications();
                Log.d(TAG, "Connection established. Status: " + device.getConnectionState().toString());
            }, throwable -> {
                if (throwable != null) {
                    throwable.printStackTrace();
                }
            });
}

Subscription私の最初の試みは、に保存する代わりに a を返すことでしたmEstablishedConnection。これにより、ユーザーは登録を解除して切断をトリガーできます。

public Subscription connect() {

    //..

    RxBleDevice device = mRxBleClient.getBleDevice(builder.toString());

    return device.establishConnection(mContext, false)
            .observeOn(AndroidSchedulers.mainThread())
            .doOnError(throwable -> {
                throwable.printStackTrace();
                Log.d(TAG, "Error establishing a connection" + throwable.getMessage());
            })
            .flatMap(rxBleConnection -> {
                mConnection = rxBleConnection;
                return Observable.empty();
            })
            .subscribe(o -> {
                setNotifications();
                Log.d(TAG, "Connection established. Status: " + device.getConnectionState().toString());
            }, throwable -> {
                if (throwable != null) {
                    throwable.printStackTrace();
                }
            });
}

上記の問題は、エラーを発信者に適切に伝達できないことです。このメソッドをリアクティブにしてRxBleConnection、サードパーティのクラスである を返すだけでなく、呼び出し元がエラーを受信できるようにするにはどうすればよいですか?