Annulla attività su timeout in RxJava

Question

Sto sperimentando con RxJava e la classe CompletableFuture di Java 8 e non riesco a capire come gestire le condizioni di timeout.

import static net.javacrumbs.futureconverter.java8rx.FutureConverter.toObservable; 

// ... 

    Observable<String> doSomethingSlowly() { 
     CompletableFuture<PaymentResult> task = CompletableFuture.supplyAsync(() -> { 
      // this call may be very slow - if it takes too long, 
      // we want to time out and cancel it. 
      return processor.slowExternalCall(); 

     }); 

     return toObservable(task); 
    } 

    // ... 

    doSomethingSlowly() 
     .single() 
     .timeout(3, TimeUnit.SECONDS, Observable.just("timeout")); 

Questo fondamentalmente funziona (se viene raggiunto il timeout di tre secondi, "timeout" è pubblicato). Vorrei tuttavia in aggiunta cancellare il futuro compito che ho inserito in un Observable - è possibile con un approccio RxJava-centrico?

So che un'opzione potrebbe essere gestire il timeout da solo, utilizzando task.get(3, TimeUnit.SECONDS), ma mi chiedo se sia possibile eseguire tutte le operazioni di gestione delle attività in RxJava.

Answer 1

Sì, puoi farlo. Aggiungere Subscription allo Subscriber.

Ciò consente di ascoltare le disiscrizioni, che avverranno se si chiama esplicitamente subscribe().unsubscribe() o se Observable viene completato correttamente o con un errore.

Se si vede una disiscrizione prima che il futuro sia completato, è possibile supporre che sia a causa di un esplicito unsubscribe o di un timeout.

public class FutureTest { 
    public static void main(String[] args) throws IOException { 
     doSomethingSlowly() 
       .timeout(1, TimeUnit.SECONDS, Observable.just("timeout")) 
       .subscribe(System.out::println); 
     System.in.read(); // keep process alive 
    } 

    private static Observable<String> doSomethingSlowly() { 
     CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> { 
      try { 
       Thread.sleep(2000); 
      } catch (InterruptedException e) { 
      } 
      return "Something"; 

     }); 
     return toObservable(future); 
    } 

    private static <T> Observable<T> toObservable(CompletableFuture<T> future) { 
     return Observable.create(subscriber -> { 
      subscriber.add(new Subscription() { 
       private boolean unsubscribed = false; 
       @Override 
       public void unsubscribe() { 
        if (!future.isDone()){ 
         future.cancel(true); 
        } 
        unsubscribed = true; 
       } 

       @Override 
       public boolean isUnsubscribed() { 
        return unsubscribed; 
       } 
      }); 

      future.thenAccept(value -> { 
       if (!subscriber.isUnsubscribed()){ 
        subscriber.onNext(value); 
        subscriber.onCompleted(); 
       } 
      }).exceptionally(throwable -> { 
       if (!subscriber.isUnsubscribed()) { 
        subscriber.onError(throwable); 
       } 
       return null; 
      }); 
     }); 
    } 
}