如何處理在地圖中的異常（）可觀察的RxJava

Question

我想這樣做：

Observable.just(bitmap) 
      .map(new Func1<Bitmap, File>() { 
       @Override 
       public File call(Bitmap photoBitmap) { 

        //File creation throws IOException, 
        //I just want it to hit the onError() inside subscribe() 

        File photoFile = new File(App.getAppContext().getCacheDir(), "userprofilepic_temp.jpg"); 
        if(photoFile.isFile()) {//delete the file first if it exists otherwise the new file won't be created 
         photoFile.delete(); 
        } 
        photoFile.createNewFile(); //saves the file in the cache dir 

        FileOutputStream fos = new FileOutputStream(photoFile); 
        photoBitmap.compress(Bitmap.CompressFormat.JPEG, 90, fos);//jpeg format 
        fos.close(); 

        return photoFile; 

       } 
      }) 
      .subscribe(//continue implementation...); 

基本上在call()方法，它可以拋出異常。我如何讓觀察者在onError()中處理它。或者，這不是考慮這個問題的正確方法嗎？

Answer 1

rx將總是捕獲錯誤，即使這是RuntimeException。 所以你可以在catch塊中拋出某種運行時異常。這實際上應該如何工作。

Observable.just(bitmap) 
       .map(b -> { 
        try { 
         // do some work which throws IOException 
         throw new IOException("something went wrong"); 
        } catch (IOException e) { 
         throw new RXIOException(e); 
         // Or you can use 
         throw Exceptions.propagate(e); 
         // This helper method will wrap your exception with runtime one 
        } 
       }).subscribe(o -> { 
        // do something here 
       }, exception -> exception.printStackTrace()); 

public static class RXIOException extends RuntimeException { 
     public RXIOException(IOException throwable) { 
      super(throwable); 
     } 
} 

Answer 2

隨着1.0.15，有fromCallable工廠方法，讓你運行一個Callable實例，你可以拋出checked異常以及每個用戶：

Observable.fromCallable(() -> {  
    File photoFile = new File(App.getAppContext().getCacheDir(), 
     "userprofilepic_temp.jpg"); 
    if (photoFile.isFile()) { 
     //delete the file if it exists otherwise the new file won't be created 
     photoFile.delete(); 
    } 
    photoFile.createNewFile(); //saves the file in the cache dir 

    FileOutputStream fos = new FileOutputStream(photoFile); 
    photoBitmap.compress(Bitmap.CompressFormat.JPEG, 90, fos);//jpeg format 
    fos.close(); 

    return photoFile; 
}) 
.subscribe(...) 

編輯：

source.flatMap(v -> { 
    try { 
     //... 
     return Observable.just(result); 
    } catch (Exception e) { 
     return Observable.error(e); 
    } 
}) 
.subscribe(...); 

Answer 3

剛剛創建的helper類此樣板提取到另一個地方：

public class RxRethrow { 
    public static <T, R> Func1<T, R> rethrow(Func1R<T, R> catchedFunc) { 
     return t -> { 
      try { 
       return catchedFunc.call(t); 
      } catch (Exception e) { 
       throw Exceptions.propagate(e); 
      } 
     }; 
    } 

    public interface Func1R<T, R> extends Function { 
     R call(T t) throws Exception; 
    } 
} 

你可以這樣調用：

.map(RxRethrow.rethrow(products -> mapper.writer(schema).writeValueAsString(products))) 

Answer 4

我不知道情況是怎樣的時這個問題首先被問及並得到解答，但RxJava目前包含一個輔助方法用於這個確切目的： Exceptions.propagate(Throwable t)

RxJava Javadoc

方便的方法來直接拋出RuntimeException和錯誤或包裹的任何其它異常類型爲一個RuntimeException。