Rx。 Android開発の例で、retryWhenおよびrepeatWhenを理解します。

ネットワークには、RxオペレーターのretryWhenおよびrepeatWhenに関するロシア語および英語の記事が多数あります。

それにもかかわらず、私は非常に頻繁にそれらを使用することに抵抗を感じます（複雑な構文と理解できない図のため）。



チェーンのセクションを効果的に再起動し、それらを使用してエラーおよびストリームの終了を伴う再起動の処理を委任する方法の例をいくつか示します。



例では、ラムダ（Retrolamda）を使用したJavaコードがありますが、Kotlinまたはpure Javaに書き換えることは難しくありません。

回路を再起動する必須の方法

Retrofitを使用し、 load（）メソッドで読み込みを開始するとします 。 Repository.getSomething（）は、 Single <Something> （）を返します。

@NonNull private Subscription loadingSubscription = Subscriptions.unsubscribed(); private void load() { subscription.unsubscribe(); subscription = repository .getSomething() .subscribe(result -> {}, err -> {}); } private void update() { load(); }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

更新リスナー（PullToRefreshViewなど）からupdate（）メソッドを呼び出します。これにより、 load（）メソッドが呼び出され、サブスクリプションが最初から作成されます。



私の意見では、上記のrepeatWhen（）演算子を使用して、よりリアクティブなメソッドを使用するオプションに注目します。

リアクティブチェーンの再起動方法-repeatWhen

PublishSubject updateSubjectを作成し、ラムダをオペレーターに渡します

repeatHandler-> repeatHandler.flatMap（nothing-> updateSubject.asObservable（））

 @NonNull private final PublishSubject<Void> updateSubject = PublishSubject.create(); private void load() { repository .getSomething() .repeatWhen(repeatHandler -> repeatHandler.flatMap(nothing -> updateSubject.asObservable())) .subscribe(result -> {}, err -> {}); }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

ここで、ダウンロードしたデータを更新するには、 nullをupdateSubjectに制限する必要があります 。

 private void update() { updateSubject.onNext(null); }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

このようなリアクティブメソッドは、単一の要素を発行した直後にonComplete（）を呼び出すSingleでのみ機能することを覚えておく必要があります（Observableでは機能しますが、ストリームが終了した後にのみ機能します）。

リアクティブエラー処理メソッドの再試行

同様に、エラーも処理できます。 ユーザーがネットワークを失い、エラーが発生し、getUser（）メソッドによって返されるSingle内のonError（）呼び出しが発生したとします。



この時点で、「接続の確認」というテキストを含むダイアログをユーザーに表示し、[OK]ボタンを押して、retry（）メソッドを呼び出します。

 @NonNull private final PublishSubject<Void> retrySubject = PublishSubject.create(); private void load() { repository .getSomething() .doOnError(err -> showConnectionDialog()) .retryWhen(retryHandler -> retryHandler.flatMap(nothing -> retrySubject.asObservable())) .subscribe(result -> {}, err -> {}); } private void retry() { retrySubject.onNext(null); }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

retrySubject.onNext（null）を呼び出すことにより、retryWhen（）の上のチェーン全体がソースgetUser（）に再サブスクライブし、要求を繰り返します。



このアプローチでは、 doOnError（）はretryWhen（）よりもチェーン内で高くなければならないことに注意することが重要です。後者はrepeatHandlerが発行する前にエラーを「吸収」するからです。



この特定のケースでは、パフォーマンスは向上せず、コードはさらに多くなりますが、これらの例は、リアクティブパターンで考え始めるのに役立ちます。



次の、恥知らずなまでにフェッチされた例では、 load（）メソッドで、 combineLatest演算子を使用して2つのソースを組み合わせています。



最初のソース-repository.getSomething（）はネットワークから何かをダウンロードし、2番目のlocalStorage.fetchSomethingReallyHuge（）はローカルストレージから重いものをロードします。

 public void load() { Observable.combineLatest(repository.getSomething(), localStorage.fetchSomethingReallyHuge(), (something, hugeObject) -> new Stuff(something, hugeObject)) .subscribe(stuff -> {}, err -> {}); }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

エラーを命令的な方法で処理する場合、各エラーでload（）を呼び出すことにより、両方のソースを再サブスクライブします。この例では、これは絶対に不要です。 ネットワークエラーの場合、2番目のソースはデータを正常に送信します;エラーは最初のソースでのみ発生します。 この場合、命令型メソッドも遅くなります。



リアクティブメソッドがどのようになるかを見てみましょう。

 public void load() { Observable.combineLatest( repository.getSomething() .retryWhen(retryHandler -> retryHandler.flatMap( err -> retrySubject.asObservable())), localStorage.fetchSomethingReallyHuge() .retryWhen(retryHandler -> retryHandler.flatMap( nothing -> retrySubject.asObservable())), (something, hugeObject) -> new Stuff(something, hugeObject)) .subscribe(stuff -> {}, err -> {}); }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

このアプローチの長所は、 retryWhen（）演算子に渡されたラムダがソース内のエラーの後にのみ実行されることです。したがって、ソースの1つのみが「間違っている」場合、再サブスクリプションはそのソースでのみ発生し、残りのチェーンはオーバーフローを待機します



また、両方のソース内でエラーが発生した場合、同じretryHandlerが2つの場所で機能します。

エラー処理の委任

次のステップは、再試行処理をRetryManagerに委任することです。 その前に、まだRx2に移行するための準備を少しし、Rx2で禁止されているnullオブジェクトをスレッドから削除できます。 これを行うには、クラスを作成できます。

 public class RetryEvent { }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

何もなし。 後で別のフラグを追加することもできますが、これは別の話です。 RetryManagerインターフェイスは次のようになります。

 interface RetryManager { Observable<RetryEvent> observeRetries(@NonNull Throwable error); }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

実装では、エラーをチェックし、ダイアログを表示し、スナックバーを表示し、サイレントタイムアウトを設定できます。 その後、retryHandlerでRetryEventを発行するために、これらすべてのUIコンポーネントからのコールバックをリッスンします。



このようなRetryManagerを使用した前の例は次のようになります。

 //pass this through constructor, DI or use singleton (but please don't) private final RetryManager retryManager; public void load() { Observable.combineLatest( repository.getSomething() .retryWhen(retryHandler -> retryHandler.flatMap( err -> retryManager.observeRetries())), localStorage.fetchSomethingReallyHuge() .retryWhen(retryHandler -> retryHandler.flatMap( nothing -> retryManager.observeRetries())), (something, hugeObject) -> new Stuff(something, hugeObject)) .subscribe(stuff -> {}, err -> {}); }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

このような簡単な方法では、エラー時の繰り返し処理はサードパーティのエンティティに委任され、依存関係として送信できます。



これらの例が誰かに役立ち、彼らのプロジェクトでrepeatWhen（）とretryWhen（）を試してみたいと思います。