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

private Observable<String> getNeteaseNetworkThumbObservable(UriRequest request) {
  return Observable.concat(Observable.create(new ObservableOnSubscribe<String>() {
    @Override
    public void subscribe(ObservableEmitter<String> emitter) throws Exception {
      String imageUrl = SPUtil
          .getValue(App.getContext(), SPUtil.COVER_KEY.NAME, request.getNeteaseCacheKey(), "");
      if (!TextUtils.isEmpty(imageUrl) && UriUtil.isNetworkUri(Uri.parse(imageUrl))) {
        emitter.onNext(imageUrl);
      }
      emitter.onComplete();
    }
  }), Observable.just(isAutoDownloadCover() && !TextUtils.isEmpty(request.getNeteaseSearchKey()))
      .filter(aBoolean -> aBoolean)
      .flatMap(aBoolean -> HttpClient.getInstance()
          .getNeteaseSearch(request.getNeteaseSearchKey(), 0, 1, request.getNeteaseType())
          .map(responseBody -> parseNeteaseNetworkImageUrl(request, responseBody))
          .firstElement().toObservable()));
}
 

@MainThread
protected NetworkBoundResource() {
    Observable<Resource<ResultType>> source;
    if (shouldFetch()) {
        source = createCall()
                .subscribeOn(Schedulers.io())
                .doOnNext(apiResponse -> saveCallResult(processResponse(apiResponse)))
                .flatMap(apiResponse -> loadFromDb().toObservable().map(Resource::success))
                .doOnError(t -> onFetchFailed())
                .onErrorResumeNext(t -> {
                    return loadFromDb()
                            .toObservable()
                            .map(data -> Resource.error(t.getMessage(), data));

                })
                .observeOn(AndroidSchedulers.mainThread());
    } else {
        source = loadFromDb()
                .toObservable()
                .map(Resource::success);
    }

    result = Observable.concat(
            loadFromDb()
                    .toObservable()
                    .map(Resource::loading)
                    .take(1),
            source
    );
}
 

@NonNull @CheckReturnValue @TargetApi(M) Observable<Permission> requestOnM(@NonNull final String... permissions) {
  final List<Observable<Permission>> list = new ArrayList<>(permissions.length);
  final List<String> unrequestedPermissions = new ArrayList<>();

  // In case of multiple permissions, we create an observable for each of them.
  // At the end, the observables are combined to have a unique response.

  for (final String permission : permissions) {
    if (isGranted(permission)) {
      list.add(Observable.just(Permission.granted(permission)));
    } else if (isRevokedByPolicy(permission)) {
      list.add(Observable.just(Permission.revokedByPolicy(permission)));
    } else {
      PublishSubject<Permission> subject = currentPermissionRequests.get(permission);

      // Create a new subject if not exists
      if (subject == null) {
        unrequestedPermissions.add(permission);
        subject = PublishSubject.create();
        currentPermissionRequests.put(permission, subject);
      }

      list.add(subject);
    }
  }

  if (!unrequestedPermissions.isEmpty()) {
    final String[] permissionsToRequest = unrequestedPermissions.toArray(new String[0]);
    startShadowActivity(permissionsToRequest);
  }

  return Observable.concat(Observable.fromIterable(list));
}
 

@TargetApi(Build.VERSION_CODES.M)
private Observable<Permission> requestImplementation(final String... permissions) {
    List<Observable<Permission>> list = new ArrayList<>(permissions.length);
    List<String> unrequestedPermissions = new ArrayList<>();

    // 在多个权限的情况下,我们为每个人创造一个可观测的。
    // 最后,可见结合独特的响应。
    for (String permission : permissions) {
        mRxPermissionsFragment.log("Requesting permission " + permission);
        if (isGranted(permission)) {
            // 返回一个授予权限。
            list.add(Observable.just(new Permission(permission, true, false)));
            continue;
        }

        if (isRevoked(permission)) {
            // 返回一个拒绝权限。
            list.add(Observable.just(new Permission(permission, false, false)));
            continue;
        }

        PublishSubject<Permission> subject = mRxPermissionsFragment.getSubjectByPermission(permission);
        //  如果不存在创建一个新的 subject
        if (subject == null) {
            unrequestedPermissions.add(permission);
            subject = PublishSubject.create();
            mRxPermissionsFragment.setSubjectForPermission(permission, subject);
        }

        list.add(subject);
    }

    if (!unrequestedPermissions.isEmpty()) {
        String[] unrequestedPermissionsArray = unrequestedPermissions.toArray(new String[unrequestedPermissions.size()]);
        requestPermissionsFromFragment(unrequestedPermissionsArray);
    }
    return Observable.concat(Observable.fromIterable(list));
}
 

@TargetApi(Build.VERSION_CODES.M)
private Observable<Permission> requestImplementation(final String... permissions) {
    List<Observable<Permission>> list = new ArrayList<>(permissions.length);
    List<String> unrequestedPermissions = new ArrayList<>();

    // In case of multiple permissions, we create an Observable for each of them.
    // At the end, the observables are combined to have a unique response.
    for (String permission : permissions) {
        mRxPermissionsFragment.get().log("Requesting permission " + permission);
        if (isGranted(permission)) {
            // Already granted, or not Android M
            // Return a granted Permission object.
            list.add(Observable.just(new Permission(permission, true, false)));
            continue;
        }

        if (isRevoked(permission)) {
            // Revoked by a policy, return a denied Permission object.
            list.add(Observable.just(new Permission(permission, false, false)));
            continue;
        }

        PublishSubject<Permission> subject = mRxPermissionsFragment.get().getSubjectByPermission(permission);
        // Create a new subject if not exists
        if (subject == null) {
            unrequestedPermissions.add(permission);
            subject = PublishSubject.create();
            mRxPermissionsFragment.get().setSubjectForPermission(permission, subject);
        }

        list.add(subject);
    }

    if (!unrequestedPermissions.isEmpty()) {
        String[] unrequestedPermissionsArray = unrequestedPermissions.toArray(new String[unrequestedPermissions.size()]);
        requestPermissionsFromFragment(unrequestedPermissionsArray);
    }
    return Observable.concat(Observable.fromIterable(list));
}
 

@Override
protected Observable<Buffer> concat(Observable<Buffer> obs1, Observable<Buffer> obs2) {
  return Observable.concat(obs1, obs2);
}