Java 类io.reactivex.observers.ResourceObserver 实例源码

public static void main(String[] args) {
        Observable<Long> source =
                Observable.interval(1, TimeUnit.SECONDS);
        ResourceObserver<Long> myObserver = new
                ResourceObserver<Long>() {
                    @Override
                    public void onNext(Long value) {
                        System.out.println(value);
                    }

                    @Override
                    public void onError(Throwable e) {
                        e.printStackTrace();
                    }

                    @Override
                    public void onComplete() {
                        System.out.println("Done!");
                    }
                };
//capture Disposable
        Disposable disposable = source.subscribeWith(myObserver);
    }

void setMode(final OpEntryInfo info) {
  Helper.setMode(context, appInfo.packageName, info)
      .subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread())
      .subscribe(new ResourceObserver<OpsResult>() {
        @Override
        public void onNext(OpsResult value) {

        }

        @Override
        public void onError(Throwable e) {
          mView.updateItem(info);
        }

        @Override
        public void onComplete() {

        }
      });
}

@SuppressLint("MissingPermission")
@Override
protected void locationSettingSuccess(final ResourceObserver<Location> emitter, GoogleApiClient client) {
    this.client = client;
    locationListener = new LocationListener() {
        @Override
        public void onLocationChanged(Location location) {
            emitter.onNext(location);
            if (locationRequest.getNumUpdates() == 1) {
                emitter.onComplete();
            }
        }
    };

    //noinspection MissingPermission
    LocationServices.FusedLocationApi.requestLocationUpdates(client, locationRequest, locationListener);
}

void handleWord(final String text) {
  if (TextUtils.isEmpty(text)) {
    mAdapter.clear();
    return;
  }

  search(text)
      .subscribeOn(Schedulers.io())
      .observeOn(AndroidSchedulers.mainThread()).subscribe(new ResourceObserver<List<AppInfo>>() {


    @Override
    protected void onStart() {
      super.onStart();
    }

    @Override
    public void onNext(List<AppInfo> value) {

      mAdapter.kw = text;
      mAdapter.showItems(value);
      if (!value.isEmpty()) {
        recyclerView.scrollToPosition(0);
      }
    }

    @Override
    public void onError(Throwable e) {

    }

    @Override
    public void onComplete() {

    }
  });


}

void changeMode(final int groupPosition, final int childPosition,
    final PermissionChildItem info) {

  info.opEntryInfo.changeStatus();

  Helper.setMode(context, info.appInfo.packageName, info.opEntryInfo)
      .subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread())
      .subscribe(new ResourceObserver<OpsResult>() {
        @Override
        public void onNext(OpsResult value) {

          mView.changeTitle(groupPosition, childPosition, info.opEntryInfo.isAllowed());
        }

        @Override
        public void onError(Throwable e) {
          try {
            info.opEntryInfo.changeStatus();
            mView.refreshItem(groupPosition, childPosition);
          } catch (Exception e2) {
            e2.printStackTrace();
          }
        }

        @Override
        public void onComplete() {

        }
      });
}

void restoreOps(final RestoreModel model) {
  final int size = model.preAppInfos.size();
  final AtomicInteger progress = new AtomicInteger();
  mView.showProgress(true, size);

  Observable.fromIterable(model.preAppInfos).observeOn(Schedulers.io())
      .flatMap(new Function<PreAppInfo, ObservableSource<OpsResult>>() {
        @Override
        public ObservableSource<OpsResult> apply(@NonNull PreAppInfo appInfo) throws Exception {
          return Helper.setModes(context, appInfo.getPackageName(), AppOpsManager.MODE_IGNORED,
              appInfo.getOps());
        }
      }).observeOn(AndroidSchedulers.mainThread()).subscribe(new ResourceObserver<OpsResult>() {
    @Override
    public void onNext(@NonNull OpsResult opsResult) {
      mView.setProgress(progress.incrementAndGet());
    }

    @Override
    public void onError(@NonNull Throwable e) {
      progress.incrementAndGet();
    }

    @Override
    public void onComplete() {
      mView.showProgress(false, 0);
      Toast.makeText(context, "恢复成功", Toast.LENGTH_LONG).show();
    }
  });
}

private void handleLocationSettings(final ResourceObserver<T> emitter, final GoogleApiClient client) {
    if (activityResultCanceled(STATE_NAME)) {
        emitter.onError(new LocationSettingDeniedException());
        return;
    }
    LocationSettingsRequest.Builder builder = new LocationSettingsRequest.Builder()
        .setAlwaysShow(true)
        .addLocationRequest(locationRequest);

    PendingResult<LocationSettingsResult> pendingResult = LocationServices.SettingsApi.checkLocationSettings(client, builder.build());
    pendingResult.setResultCallback(new ResultCallback<LocationSettingsResult>() {
        @Override
        public void onResult(LocationSettingsResult result) {
            final Status status = result.getStatus();
            switch (status.getStatusCode()) {
                case LocationSettingsStatusCodes.SUCCESS:
                    locationSettingSuccess(emitter, client);
                    break;
                case LocationSettingsStatusCodes.RESOLUTION_REQUIRED:
                    resolveResolutionRequired(emitter, status);
                    emitter.onComplete();  // silently close our api connection, full restart through responsehandler required, so we can close this connection.
                    break;
                case LocationSettingsStatusCodes.SETTINGS_CHANGE_UNAVAILABLE:
                    emitter.onError(new RxPlayServices.RxLocationError(LocationSettingsStatusCodes.SETTINGS_CHANGE_UNAVAILABLE));
                    break;
            }
        }
    });
}

protected void resolveResolutionRequired(ResourceObserver<T> subscriber, Status status) {
    recieveStateResponse(STATE_NAME);
    try {
        status.startResolutionForResult(activity, getRequestCode());
    } catch (IntentSender.SendIntentException e) {
        subscriber.onError(new RxPlayServices.RxLocationError(e));
    }
}

void load() {
  observable = Helper.getAppPermission(context, appInfo.packageName,
      PreferenceManager.getDefaultSharedPreferences(context)
          .getBoolean("key_show_no_prems", false));

  observable.subscribeOn(Schedulers.io())
      .observeOn(AndroidSchedulers.mainThread())
      .subscribe(new ResourceObserver<List<OpEntryInfo>>() {

        @Override
        protected void onStart() {
          super.onStart();
        }

        @Override
        public void onNext(List<OpEntryInfo> opEntryInfos) {

          if (opEntryInfos != null && !opEntryInfos.isEmpty()) {
            if (autoDisabled) {

              if (sortByMode) {
                reSortByModePerms(opEntryInfos);
              } else {
                mView.showProgress(false);
                mView.showPerms(opEntryInfos);
              }
            } else {
              autoDisable();
            }

          } else {
            mView.showError(context.getString(R.string.no_perms));
          }
          loadSuccess = true;
        }

        @Override
        public void onError(Throwable e) {
          mView.showError(getHandleError(e));

          loadSuccess = false;
        }

        @Override
        public void onComplete() {
        }
      });
}

private void loadData(final boolean isFirst) {
  boolean showSysApp = PreferenceManager.getDefaultSharedPreferences(getApplicationContext())
      .getBoolean("show_sysapp", false);
  Helper.getInstalledApps(getApplicationContext(), showSysApp)
      .map(Helper.getSortComparator(getApplicationContext())).subscribeOn(Schedulers.io())
      .observeOn(AndroidSchedulers.mainThread()).subscribe(new ResourceObserver<List<AppInfo>>() {

    @Override
    protected void onStart() {
      super.onStart();
      if(isFirst){
        mProgressBar.setVisibility(View.VISIBLE);
        recyclerView.setVisibility(View.GONE);
      }
    }

    @Override
    public void onNext(List<AppInfo> value) {
      adapter.showItems(value);
      mSearchHandler.setBaseData(new ArrayList<AppInfo>(value));

      invalidateOptionsMenu();

    }

    @Override
    public void onError(Throwable e) {
      e.printStackTrace();
      mSwipeRefreshLayout.setRefreshing(false);
      Toast.makeText(getApplicationContext(), e.getMessage(), Toast.LENGTH_LONG).show();

      invalidateOptionsMenu();
    }

    @Override
    public void onComplete() {
      mProgressBar.setVisibility(View.GONE);
      recyclerView.setVisibility(View.VISIBLE);
      mSwipeRefreshLayout.setRefreshing(false);

      if (isFirst) {
        mSwipeRefreshLayout.setEnabled(true);
      }

      invalidateOptionsMenu();
    }
  });
  loadUsers();
}

@Override
public void onGoogleApiClientReady(final ResourceObserver<String> emitter, final GoogleApiClient client) {
    if (activityResultCanceled(STATE_NAME)) {
        // always check and handle responses after recieveStateResponse
        emitter.onError(new GoogleLoginCanceledException());
        return;
    }
    final Disposable disposable = Observable.create(new ObservableOnSubscribe<String>() {
        @Override
        public void subscribe(ObservableEmitter<String> subscriber) {
            //noinspection MissingPermission
            final String accountName = Plus.AccountApi.getAccountName(client);
            final Account account = new Account(accountName, GoogleAuthUtil.GOOGLE_ACCOUNT_TYPE);
            final String scopes = "oauth2:" + GOOGLE_PLUS_SCOPES;
            try {
                subscriber.onNext(GoogleAuthUtil.getToken(activity.getApplicationContext(), account, scopes));
                resetClientForDemoPurposes(client);
                subscriber.onComplete();
            } catch (Exception e) {
                subscriber.onError(e);
            }
        }
    }).subscribeOn(Schedulers.computation()).observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Consumer<String>() {
                @Override
                public void accept(String s) {
                    emitter.onNext(s);
                }
            }, new Consumer<Throwable>() {
                @Override
                public void accept(Throwable throwable) {
                    if (throwable instanceof UserRecoverableAuthException) {
                        // always trigger recieveStateResponse before startActivityForResult (or permission...)
                        recieveStateResponse(STATE_NAME);
                        activity.startActivityForResult(((UserRecoverableAuthException) throwable).getIntent(), getRequestCode());
                        emitter.onComplete();
                    } else {
                        emitter.onError(throwable);
                    }
                }
            });
    emitter.add(disposable);
}

@Override
public void subscribe(final ObservableEmitter<T> emitter) {
    if (activityResultCanceled(STATE_NAME)) {
        emitter.onError(new PlayServicesConnectionCanceledException());
        return;
    }
    GoogleApiClient.Builder builder = new GoogleApiClient.Builder(activity);
    for (Api<? extends Api.ApiOptions.NotRequiredOptions> service : services) {
        builder.addApi(service);
    }
    if (scopes != null) {
        for (Scope scope : scopes) {
            builder.addScope(scope);
        }
    }
    builder.addConnectionCallbacks(new GoogleApiClient.ConnectionCallbacks() {
        @Override
        public void onConnected(Bundle bundle) {
            ResourceObserver<T> resourceObserver = new ResourceObserver<T>() {
                @Override
                public void onNext(T t) {
                    emitter.onNext(t);
                }

                @Override
                public void onError(Throwable e) {
                    emitter.onError(e);
                }

                @Override
                public void onComplete() {
                    emitter.onComplete();
                }
            };


            onGoogleApiClientReady(resourceObserver, client);
        }

        @Override
        public void onConnectionSuspended(int reason) {
            emitter.onError(new GoogleApiConnectionSuspended(reason));
        }
    });
    builder.addOnConnectionFailedListener(new GoogleApiClient.OnConnectionFailedListener() {
        @Override
        public void onConnectionFailed(final ConnectionResult connectionResult) {
            if (connectionResult.hasResolution()) {
                resolveConnectionFailed(connectionResult, emitter);
            } else {
                emitter.onError(new GoogleApiConnectionFailed(connectionResult));
            }
        }
    });
    client = builder.build();
    client.connect();
    emitter.setCancellable(new Cancellable() {
        @Override
        public void cancel() throws Exception {
            onUnsubscribe();
        }
    });
}

@Override
public void onGoogleApiClientReady(ResourceObserver<GoogleApiClient> emitter, GoogleApiClient client) {
    emitter.onNext(client);
}

@Override
public void onGoogleApiClientReady(ResourceObserver<T> emitter, GoogleApiClient client) {
    handleLocationSettings(emitter, client);
}

public abstract void onGoogleApiClientReady(ResourceObserver<T> emitter, GoogleApiClient client);

protected abstract void locationSettingSuccess(ResourceObserver<T> subscriber, GoogleApiClient client);