java - SyncOnSubscribe RxJava で generateState のエラーを処理するにはどうすればよいですか?

Question

SyncOnSubscribe を作成するコツをつかもうとしていgenerateStateますが、状態の初期化中にメソッドが失敗した場合の対処方法がよくわかりません。

return Observable.create(new SyncOnSubscribe<MyState, String>() {
    @Override
    protected MyState generateState() {
        return new MyState();   // <---- what if this fails?
    }

    @Override
    protected MyState next(MyState state, Observer<? super String> observer) {
        // do something with state
    }
});

これを手に負えないように処理できるいくつかの方法を考えることができます。

1. 実行時例外がスローされた場合、ライブラリは自動的に呼び出しo.onErrorますか? （編集を参照）。
2. MyStateエラーを格納する別の変数をラップしo.onErrorて、最初に呼び出されたときに自分で呼び出すことができnextます。

これに関して提案された実践があるかどうか、私はただ興味がありますか?

ありがとう！

編集:メソッドでランタイム例外をスローしようとしましたが、メソッドで検査するエラーでgenerateStateラップすることを示していると思います。より良い提案がある場合は、コメント/回答してください。MyStatenext

public static Observable<String> getEventsOnSubscribe1() {
    return Observable.create((s) -> {
        throw new UnsupportedOperationException("getEvents3");
    });
};

/** Like 'getEventsOnSubscribe1' but wrap exception and call onError manually.  */
public static Observable<String> getEventsOnSubscribe2() {
    return Observable.create((s) -> {
        try {
            throw new UnsupportedOperationException("getEvents3");
        } catch (Exception ex) {
            s.onError(ex);
        }
    });
};

public static Observable<String> getEventsSyncOnSubscribe() {
    return Observable.create(new SyncOnSubscribe<Channel, String>() {
        @Override
        protected Channel generateState() {
            System.out.println("SyncOnSubscribe.generateState");
            throw new UnsupportedOperationException("Not supported yet.");
        }

        @Override
        protected Channel next(Channel state, Observer<? super String> observer) {
            System.out.println("SyncOnSubscribe.next");
            throw new UnsupportedOperationException("Not supported yet.");
        }

        @Override
        protected void onUnsubscribe(Channel state) {
            System.out.println("SyncOnSubscribe.onUnsubscribe");
            throw new UnsupportedOperationException("Not supported yet.");
        }
    });
}

public static void main(String[] args) throws IOException, TimeoutException {
    getEventsOnSubscribe1()
    //getEventsOnSubscribe2()
    //getEventsOnSyncSubscribe()
            .toBlocking()
            .subscribe(new Subscriber<String>() {
                @Override
                public void onCompleted() {
                    System.out.println("onCompleted");
                }

                @Override
                public void onError(Throwable e) {
                    System.out.println("onError: " + e.getLocalizedMessage());
                }

                @Override
                public void onNext(String t) {
                    System.out.println("onNext: " + t);
                }
            });
}

上記のmain関数を使用するgetEventsOnSubscribe1と、getEventsSyncOnSubscribe両方ともsubscriberを呼び出しますonErrorが、プログラムはハングしたままになります。whichを使用getEventsOnSubscribe2して例外をラップし、手動で呼び出すs.onErrorと、プログラムは終了できます。