io.reactivex.Observable#error ( )源码实例Demo

@Override
@RestrictTo(RestrictTo.Scope.LIBRARY_GROUP)
public synchronized <T> Observable<T> queue(final Operation<T> operation) {
    if (!shouldRun) {
        return Observable.error(disconnectionException);
    }

    return Observable.create(new ObservableOnSubscribe<T>() {
        @Override
        public void subscribe(ObservableEmitter<T> emitter) {
            final FIFORunnableEntry entry = new FIFORunnableEntry<>(operation, emitter);
            emitter.setCancellable(new Cancellable() {
                @Override
                public void cancel() {
                    if (queue.remove(entry)) {
                        logOperationRemoved(operation);
                    }
                }
            });

            logOperationQueued(operation);
            queue.add(entry);
        }
    });
}
 

/**
 * login with auth data.
 * @param authData auth data.
 * @param platform platform string.
 * @param failOnNotExist flag to indicate to exit if failed or not.
 * @return observable instance.
 */
public Observable<AVUser> loginWithAuthData(final Map<String, Object> authData, final String platform,
                                            final boolean failOnNotExist) {
  if (null == authData || authData.isEmpty()) {
    return Observable.error(new IllegalArgumentException(String.format(ILLEGALARGUMENT_MSG_FORMAT, "authData")));
  }
  if (StringUtil.isEmpty(platform)) {
    return Observable.error(new IllegalArgumentException(String.format(ILLEGALARGUMENT_MSG_FORMAT, "platform")));
  }

  HashMap<String, Object> data = createUserMapAFAP(getUsername(), null, getEmail(), getMobilePhoneNumber(), null);
  Map<String, Object> authMap = new HashMap<String, Object>();
  authMap.put(platform, authData);
  data.put(AUTHDATA_TAG, authMap);
  JSONObject param = new JSONObject(data);
  return PaasClient.getStorageClient().signUpWithFlag(param, failOnNotExist).map(new Function<AVUser, AVUser>() {
    @Override
    public AVUser apply(AVUser avUser) throws Exception {
      AVUser.this.resetByRawData(avUser);
      AVUser.changeCurrentUser(AVUser.this, true);
      return AVUser.this;
    }
  });
}
 

private Observable<AVStatus> sendInBackground(String inboxType, AVQuery query) {
  if (!checkCurrentUserAuthenticated()) {
    return Observable.error(ErrorUtils.sessionMissingException());
  }
  setSource(AVUser.currentUser());

  Map<String, Object> param = new HashMap<>();
  param.put("data", this.serverData);
  param.put("inboxType", inboxType);

  Map<String, Object> queryCondition = query.assembleJsonParam();
  param.put("query", queryCondition);
  return PaasClient.getStorageClient().postStatus(param).map(new Function<AVStatus, AVStatus>() {
    @Override
    public AVStatus apply(AVStatus avStatus) throws Exception {
      AVStatus.this.mergeRawData(avStatus, true);
      return avStatus;
    }
  });
}
 

public Observable<AVUser> loginWithAuthData(final Map<String, Object> authData, final String platform,
                                            final String unionId, final String unionIdPlatform,
                                            final boolean asMainAccount, final boolean failOnNotExist) {
  if (null == authData || authData.isEmpty()) {
    return Observable.error(new IllegalArgumentException(String.format(ILLEGALARGUMENT_MSG_FORMAT, "authData")));
  }
  if (StringUtil.isEmpty(unionId)) {
    return Observable.error(new IllegalArgumentException(String.format(ILLEGALARGUMENT_MSG_FORMAT, "unionId")));
  }
  if (StringUtil.isEmpty(unionIdPlatform)) {
    return Observable.error(new IllegalArgumentException(String.format(ILLEGALARGUMENT_MSG_FORMAT, "unionIdPlatform")));
  }
  authData.put(AUTHDATA_ATTR_UNIONID, unionId);
  authData.put(AUTHDATA_ATTR_UNIONID_PLATFORM, unionIdPlatform);
  if (asMainAccount) {
    authData.put(AUTHDATA_ATTR_MAIN_ACCOUNT, asMainAccount);
  }
  return loginWithAuthData(authData, platform, failOnNotExist);
}
 

/**
 * 搜索
 */
public Observable<List<SearchBookBean>> searchBook(String content, int page) {
    if (bookSourceBean == null || isEmpty(bookSourceBean.getRuleSearchUrl())) {
        return Observable.create(emitter -> {
            emitter.onNext(new ArrayList<>());
            emitter.onComplete();
        });
    }
    BookList bookList = new BookList(tag, name, bookSourceBean, false);
    try {
        AnalyzeUrl analyzeUrl = new AnalyzeUrl(bookSourceBean.getRuleSearchUrl(), content, page, headerMap, tag);
        return getResponseO(analyzeUrl)
                .flatMap(bookList::analyzeSearchBook);
    } catch (Exception e) {
        return Observable.error(e);
    }
}
 

/**
 * Call this method to send the SMS. You must call a "convert" method before to specify the data to send. This
 * method will fail if the app is not granted the permissions required to use SMS in the device: READ_PHONE_STATE,
 * SEND_SMS, READ_SMS and RECEIVE_SMS.
 * @return {@code Observable} emitting the sending states.
 */
public Observable<SmsRepository.SmsSendingState> send() {
    if (smsParts == null || smsParts.isEmpty()) {
        return Observable.error(new IllegalStateException("Convert method should be called first"));
    }
    return checkPreconditions(
    ).andThen(
            localDbRepository.addOngoingSubmission(submissionId, getSubmissionType())
    ).andThen(
            localDbRepository.getGatewayNumber()
    ).flatMapObservable(number ->
            smsRepository.sendSms(number, smsParts, SENDING_TIMEOUT)
    ).flatMap(state -> {
        if (!finishedSending && state.getSent() == state.getTotal()) {
            finishedSending = true;
            return converter.updateSubmissionState(State.SENT_VIA_SMS).andThen(
                    localDbRepository.removeOngoingSubmission(submissionId)
            ).andThen(Observable.just(state));
        }
        return Observable.just(state);
    });
}
 

/**
 * Delete all objects in async mode.
 * @param objects object collection.
 * @return observable instance.
 */
public static Observable<AVNull> deleteAllInBackground(Collection<? extends AVObject> objects) {
  if (null == objects || objects.isEmpty()) {
    return Observable.just(AVNull.getINSTANCE());
  }
  String className = null;
  Map<String, Object> ignoreParams = new HashMap<>();
  StringBuilder sb = new StringBuilder();
  for (AVObject o : objects) {
    if (StringUtil.isEmpty(o.getObjectId()) || StringUtil.isEmpty(o.getClassName())) {
      return Observable.error(new IllegalArgumentException("Invalid AVObject, the class name or objectId is blank."));
    }
    if (className == null) {
      className = o.getClassName();
      sb.append(o.getObjectId());
    } else if (className.equals(o.getClassName())) {
      sb.append(",").append(o.getObjectId());
    } else {
      return Observable.error(new IllegalArgumentException("The objects class name must be the same."));
    }
  }
  return PaasClient.getStorageClient().deleteObject(className, sb.toString(), ignoreParams);
}
 

public Observable<AVUser> dissociateWithAuthData(final String platform) {
  if (StringUtil.isEmpty(platform)) {
    return Observable.error(new IllegalArgumentException(String.format(ILLEGALARGUMENT_MSG_FORMAT, "platform")));
  }

  String objectId = getObjectId();
  if (StringUtil.isEmpty(objectId) || !isAuthenticated()) {
    return Observable.error(new AVException(AVException.SESSION_MISSING,
            "the user object missing a valid session"));
  }
  this.remove(AUTHDATA_TAG + "." + platform);
  return this.saveInBackground().map(new Function<AVObject, AVUser>() {
    public AVUser apply(@NonNull AVObject var1) throws Exception {
      Map<String, Object> authData = (Map<String, Object>) AVUser.this.get(AUTHDATA_TAG);
      if (authData != null) {
        authData.remove(platform);
      }
      return AVUser.this;
    }
  });
}
 

/**
 * Save object in background.
 * @param option save option.
 * @return observable instance.
 */
public Observable<? extends AVObject> saveInBackground(final AVSaveOption option) {
  Map<AVObject, Boolean> markMap = new HashMap<>();
  if (hasCircleReference(markMap)) {
    return Observable.error(new AVException(AVException.CIRCLE_REFERENCE, "Found a circular dependency when saving."));
  }

  Observable<List<AVObject>> needSaveFirstly = generateCascadingSaveObjects();
  return needSaveFirstly.flatMap(new Function<List<AVObject>, Observable<? extends AVObject>>() {
    @Override
    public Observable<? extends AVObject> apply(List<AVObject> objects) throws Exception {
      logger.d("First, try to execute save operations in thread: " + Thread.currentThread());
      for (AVObject o: objects) {
        o.save();
      }
      logger.d("Second, save object itself...");
      return saveSelfOperations(option);
    }
  });
}
 

/**
 * get status count.
 * @return observable instance.
 */
@Override
public Observable<Integer> countInBackground() {
  if (null == this.owner && null == this.source) {
    return Observable.error(ErrorUtils.illegalArgument("source or owner is null, please initialize correctly."));
  }
  if (null != this.owner) {
    return Observable.error(ErrorUtils.invalidStateException("countInBackground doesn't work for inbox query," +
            " please use unreadCountInBackground."));
  }

  Map<String, String> query = assembleParameters();
  query.put("count", "1");
  query.put("limit", "0");
  return PaasClient.getStorageClient().queryCount(AVStatus.CLASS_NAME, query);
}
 

public static Observable<Boolean> importReplaceRule(String text) {
    if (TextUtils.isEmpty(text)) return null;
    text = text.trim();
    if (text.length() == 0) return null;
    if (StringUtils.isJsonType(text)) {
        return importReplaceRuleO(text)
                .compose(RxUtils::toSimpleSingle);
    }
    if (NetworkUtils.isUrl(text)) {
        return BaseModelImpl.getInstance().getRetrofitString(StringUtils.getBaseUrl(text), "utf-8")
                .create(IHttpGetApi.class)
                .get(text, AnalyzeHeaders.getMap(null))
                .flatMap(rsp -> importReplaceRuleO(rsp.body()))
                .compose(RxUtils::toSimpleSingle);
    }
    return Observable.error(new Exception("不是Json或Url格式"));
}
 

@NonNull
private static Observable<List<Media>> queueObservableForSchemeAndData(
        @NonNull final QueueProviderFiles queueProvider,
        @NonNull final Pair<Uri, String> data) {
    switch (data.second) {
        case "file":
            return queueFromFileActionView(queueProvider, data.first);

        default:
            return Observable.error(new IOException("Unhandled Uri scheme: " + data.second));
    }
}
 

/**
 * 章节缓存
 */
@Override
public Observable<BookContentBean> getBookContent(BookInfoBean bookInfo, ChapterBean chapter) {
    try {
        IStationBookModel bookModel = getBookSourceModel(bookInfo.getTag());
        return bookModel.getBookContent(bookInfo.getChapterListUrl(), chapter)
                .flatMap(bookContentBean -> saveChapterInfo(bookInfo, bookContentBean));
    } catch (Exception e) {
        return Observable.error(e);
    }
}
 

/**
 * 获取目录
 */
@Override
public Observable<List<ChapterBean>> getChapterList(final BookShelfBean bookShelfBean) {
    try {
        final AnalyzeUrl analyzeUrl = new AnalyzeUrl(bookShelfBean.getNoteUrl(), bookShelfBean.getBookInfoBean().getChapterListUrl(), headerMap(true));
        final BookChapters bookChapters = new BookChapters(tag, bookSourceBean);
        return toObservable(analyzeUrl)
                .flatMap(response -> bookChapters.analyzeChapters(response, analyzeUrl.getQueryUrl(), bookShelfBean));
    } catch (Exception e) {
        Logger.e(TAG, "chapterList", e);
        return Observable.error(e);
    }
}
 

/**
 * Creates a new AesEncryptionObservable that will listen to fingerprint authentication
 * to encrypt the given data.
 *
 * @param context   context to use
 * @param keyName   keyName to use for the decryption
 * @param encrypted data to encrypt  @return Observable {@link FingerprintEncryptionResult}
 * @return Observable result of the decryption
 */
static Observable<FingerprintDecryptionResult> create(Context context, String keyName, String encrypted) {
	try {
		return Observable.create(new RsaDecryptionObservable(new FingerprintApiWrapper(context),
				new RsaCipherProvider(context, keyName),
				encrypted,
				new Base64Provider()));
	} catch (Exception e) {
		return Observable.error(e);
	}
}
 

/**
 * 发现
 */
public Observable<List<SearchBookBean>> findBook(String url, int page) {
    if (bookSourceBean == null) {
        return Observable.error(new NoSourceThrowable(tag));
    }
    BookList bookList = new BookList(tag, name, bookSourceBean, true);
    try {
        AnalyzeUrl analyzeUrl = new AnalyzeUrl(url, null, page, headerMap, tag);
        return getResponseO(analyzeUrl)
                .flatMap(bookList::analyzeSearchBook);
    } catch (Exception e) {
        return Observable.error(new Throwable(String.format("%s错误:%s", url, e.getLocalizedMessage())));
    }
}
 

@Override
public Observable<ChapterBean> getAudioBookContent(BookInfoBean bookInfo, ChapterBean chapter) {
    try {
        IStationBookModel bookModel = getBookSourceModel(bookInfo.getTag());
        return ((IAudioBookChapterModel) bookModel).getAudioBookContent(bookInfo.getChapterListUrl(), chapter);
    } catch (Exception e) {
        return Observable.error(e);
    }
}
 

@Override
public Observable<ScanResult> scanBleDevices(ScanSettings scanSettings, ScanFilter... scanFilters) {
    return Observable.error(new RuntimeException("not implemented")); // TODO [DS]
}
 

@RequestMapping(method = RequestMethod.GET, value = "/throw")
public ObservableDeferredResult<Object> error() {
    return new ObservableDeferredResult<>(Observable.error(new RuntimeException("Unexpected")));
}
 

/**
 * Encrypt data with the given {@link EncryptionMethod}. Depending on the given method, the
 * fingerprint sensor might be enabled and waiting for the user to authenticate before the
 * encryption step. All encrypted data can only be accessed again by calling
 * {@link #decrypt(EncryptionMethod, Context, String, String)} with the same
 * {@link EncryptionMethod} that was used for encryption of the given value.
 * <p>
 * Take more details about the encryption method and how they behave from {@link EncryptionMethod}
 * <p>
 * The resulting {@link FingerprintEncryptionResult} will contain the encrypted data as a String
 * and is accessible via {@link FingerprintEncryptionResult#getEncrypted()} if the
 * operation was successful. Save this data where you please, but don't change it if you
 * want to decrypt it again!
 *
 * @param method    the encryption method to use
 * @param context   context to use
 * @param keyName   name of the key to store in the Android {@link java.security.KeyStore}
 * @param toEncrypt data to encrypt
 * @param keyInvalidatedByBiometricEnrollment whether or not the key will be invalidated when fingerprints are added
 *                                            or changed. Works only on Android N(API 24) and above.
 * @return Observable {@link FingerprintEncryptionResult} that will contain the encrypted data.
 * Will complete once the operation was successful or failed entirely.
 */
public static Observable<FingerprintEncryptionResult> encrypt(@NonNull EncryptionMethod method,
																															@NonNull Context context,
																															@Nullable String keyName,
																															@NonNull char[] toEncrypt,
																															boolean keyInvalidatedByBiometricEnrollment) {
	switch (method) {
		case AES:
			return AesEncryptionObservable.create(context, keyName, toEncrypt, keyInvalidatedByBiometricEnrollment);
		case RSA:
			return RsaEncryptionObservable.create(context, keyName, toEncrypt, keyInvalidatedByBiometricEnrollment);
		default:
			return Observable.error(new IllegalArgumentException("Unknown encryption method: " + method));
	}
}