如何处理在地图中的异常（）可观察的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。