io.reactivex.functions.Action#run ( )源码实例Demo

public static <T> ObservableOperator<T,T> doOnEmpty(Action action){
    return observer -> new DisposableObserver<T>() {
        boolean isEmpty = true;
        @Override
        public void	onNext(T value) {
            isEmpty = false;
            observer.onNext(value);
        }
        @Override
        public void onError(Throwable t) {
            observer.onError(t);
        }
        @Override
        public void onComplete() {
            if (isEmpty) {
                try {
                    action.run();
                } catch (Exception e) {
                    onError(e);
                    return;
                }
            }
            observer.onComplete();
        }
    };
}
 

@BindingConversion
public static View.OnClickListener toOnClickListener(final Action listener) {
    if (listener != null) {
        return new View.OnClickListener() {
            @Override
            public void onClick(View view) {
                try {
                    listener.run();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        };
    } else {
        return null;
    }
}
 

private void writeStreamEnd(AsyncResult<Void> result) {
  try {
    Action a;
    if (result.succeeded()) {
      synchronized (this) {
        a = writeStreamEndHandler;
      }
      if (a != null) {
        a.run();
      }
    } else {
      Consumer<? super Throwable> c;
      synchronized (this) {
        c = this.writeStreamEndErrorHandler;
      }
      if (c != null) {
        c.accept(result.cause());
      }
    }
  } catch (Throwable t) {
    Exceptions.throwIfFatal(t);
    RxJavaPlugins.onError(t);
  }
}
 

private void writeStreamEnd(AsyncResult<Void> result) {
  try {
    Action a;
    if (result.succeeded()) {
      synchronized (this) {
        a = writeStreamEndHandler;
      }
      if (a != null) {
        a.run();
      }
    } else {
      Consumer<? super Throwable> c;
      synchronized (this) {
        c = this.writeStreamEndErrorHandler;
      }
      if (c != null) {
        c.accept(result.cause());
      }
    }
  } catch (Throwable t) {
    Exceptions.throwIfFatal(t);
    RxJavaPlugins.onError(t);
  }
}
 

@SuppressWarnings({"PMD.PrematureDeclaration", "PMD.PreserveStackTrace"})
private void wrapAction(Action action, String tag) {
    long startMillis = System.currentTimeMillis();
    try {
        action.run();
    } catch (Exception e) {
        throw new RuntimeException("Exception thrown during database export action: " + tag);
    }
    long endMillis = System.currentTimeMillis();

    Log.e("DatabaseExport", tag + ": " + (endMillis - startMillis) + "ms");
}