Observable.from（Future）の使い方は？

Question

私はAndroidでRxJavaを学んでいます。私はネットワーキングでそれを使用したいので、私はメインスレッド上のものを行うことはできません。

私は次のコードを持っている：

final String RXTAG = "Rx"; 
Log.d(RXTAG, "Starting Rx experiment"); 

final FutureTask<String> future = new FutureTask<>(new Callable<String>() { 
    @Override 
    public String call() throws InterruptedException { 
     Log.d(RXTAG, "Callable called on thread " + Thread.currentThread().getName()); 
     Utils.assertNotUIThread(); 
     Thread.sleep(TimeUnit.SECONDS.toMillis(1)); // Simulates network latency 
     return "hello"; 
    } 
}); 

Observable.from(future, Schedulers.io()).timeout(5, TimeUnit.SECONDS).subscribe(
     new Action1<String>() { 
      @Override 
      public void call(final String s) { 
       Log.d(RXTAG, "Next " + s); 
      } 
     }, 
     new Action1<Throwable>() { 
      @Override 
      public void call(final Throwable throwable) { 
       Log.w(RXTAG, throwable); 
      } 
     }, 
     new Action0() { 
      @Override 
      public void call() { 
       Log.d(RXTAG, "Completed"); 
      } 
     } 
); 

をしかし、5 secundsとコーラブルはログを呼び出した後、それはTimeoutExceptionで終わる示したことはありません。何が問題なのですか？

Answer 1

自分でFutureを実行する必要があります。しかし、幸運にも、メインスレッドでCallableを実行するか、またはExecutorを使用する必要があります。

// Extract Callable from FutureTask 

Single.fromCallable(callable).timeout(5, TimeUnit.SECONDS).subscribeOn(Schedulers.io()).subscribe(
    new Action1<String>() { 
     @Override 
     public void call(final String s) { 
      Log.d(RXTAG, "Success " + s); 
     } 
    }, 
    new Action1<Throwable>() { 
     @Override 
     public void call(final Throwable throwable) { 
      Log.w(RXTAG, throwable); 
     } 
    } 
); 

をそして今、それが期待通りに動作します：

代わりに、私は以下のソリューションが付属して

07-29 13:41:26.516 D/Rx: Starting Rx experiment 
07-29 13:41:26.547 D/Rx: Callable called on thread RxIoScheduler-2 
07-29 13:41:27.550 D/Rx: Success hello 

Answer 2

試みは前に加入したコードの下に追加します。

.doOnSubscribe(disposable -> future.run()) 
.subscribeOn(Schedulers.io()) 

Rxjava2コード：

Observable.fromFuture(future, Schedulers.io()) 
.doOnSubscribe(disposable -> future.run()) 
.subscribeOn(Schedulers.io()) 
.subscribe(someConsumer()). 

はうまく動作します。