下面列出了怎么用io.reactivex.observers.DisposableSingleObserver的API类实例代码及写法，或者点击链接到github查看源代码。
private void requestWeathers(String cityName) {
ApiService apiService = ApiClient.getClient().create(ApiService.class);
disposable.add(
apiService.getMultipleDaysWeather(
cityName, Constants.UNITS, defaultLang, 16, apiKey)
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribeWith(new DisposableSingleObserver<MultipleDaysWeatherResponse>() {
@Override
public void onSuccess(MultipleDaysWeatherResponse response) {
handleMultipleDaysResponse(response);
swipeContainer.setRefreshing(false);
}
@Override
public void onError(Throwable e) {
swipeContainer.setRefreshing(false);
Log.e("MainActivity", "onError: " + e.getMessage());
}
})
);
}
private void getFiveDaysWeather(String cityName) {
disposable.add(
apiService.getMultipleDaysWeather(
cityName, Constants.UNITS, defaultLang, 5, apiKey)
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribeWith(new DisposableSingleObserver<MultipleDaysWeatherResponse>() {
@Override
public void onSuccess(MultipleDaysWeatherResponse response) {
handleFiveDayResponse(response, cityName);
}
@Override
public void onError(Throwable e) {
e.printStackTrace();
}
})
);
}
private void getFiveDaysHourlyWeather(String cityName) {
disposable.add(
apiService.getFiveDaysWeather(
cityName, Constants.UNITS, defaultLang, apiKey)
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribeWith(new DisposableSingleObserver<FiveDayResponse>() {
@Override
public void onSuccess(FiveDayResponse response) {
handleFiveDayHourlyResponse(response);
}
@Override
public void onError(Throwable e) {
e.printStackTrace();
}
})
);
}
void sendSMS() {
if (smsSender != null) {
// started sending before, just republish state
return;
}
smsSender = d2.smsModule().smsSubmitCase();
reportState(State.STARTED, 0, 0);
Single<Integer> convertTask = chooseConvertTask();
if (convertTask == null) return;
disposables.add(convertTask.subscribeOn(Schedulers.newThread()
).observeOn(Schedulers.newThread()
).subscribeWith(new DisposableSingleObserver<Integer>() {
@Override
public void onSuccess(Integer count) {
reportState(State.CONVERTED, 0, count);
reportState(State.WAITING_COUNT_CONFIRMATION, 0, count);
}
@Override
public void onError(Throwable e) {
reportError(e);
}
}));
}
private void fetchData() {
getView().showProgressBar();
getView().hideListView();
getCompositeDisposable().add(getDataProvider().getPermissionGroups(false)
.subscribeOn(Schedulers.computation())
.observeOn(AndroidSchedulers.mainThread())
.subscribeWith(new DisposableSingleObserver<ArrayList<PermissionGroupDetails>>() {
@Override
public void onSuccess(ArrayList<PermissionGroupDetails> permissionGroupDetails) {
getView().hideProgressBar();
getView().showListView();
permissionList = permissionGroupDetails;
getView().notifyListAdapter();
}
@Override public void onError(Throwable e) {
}
}));
}
private void extractAPK() {
getCompositeDisposable().add(createDirectoryAndExtractApk().subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribeWith(new DisposableSingleObserver<String>() {
@Override public void onSuccess(String filePath) {
getView().onExtractionComplete(filePath);
}
@Override public void onError(Throwable e) {
if (e instanceof IllegalStateException) {
getView().showFileExitsAlert();
return;
}
getView().showError("APK extraction failed");
}
}));
}
private void getAppDetails() {
if (appDetailList != null) {
return;
}
getView().showProgressBar();
getView().hideListView();
getCompositeDisposable().add(getDataProvider().getAppDetailsList(packages, false)
.subscribeOn(Schedulers.computation())
.observeOn(AndroidSchedulers.mainThread())
.subscribeWith(new DisposableSingleObserver<ArrayList<AppDetails>>() {
@Override public void onSuccess(ArrayList<AppDetails> appDetails) {
getView().hideProgressBar();
getView().showListView();
if (appDetailList != appDetails) {
appDetailList = appDetails;
getView().notifyListAdapter();
}
}
@Override public void onError(Throwable e) {
}
}));
}
private void fetchRepositoryDetails() {
ApolloCall<EntryDetailQuery.Data> entryDetailQuery = application.apolloClient()
.query(new EntryDetailQuery(repoFullName))
.cacheControl(CacheControl.CACHE_FIRST);
//Example call using Rx2Support
disposables.add(Rx2Apollo.from(entryDetailQuery)
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribeWith(new DisposableSingleObserver<Response<EntryDetailQuery.Data>>() {
@Override
public void onSuccess(Response<EntryDetailQuery.Data> dataResponse) {
setEntryData(dataResponse.data());
}
@Override
public void onError(Throwable e) {
Log.e(TAG, e.getMessage(), e);
}
}));
}
@Override
protected void protectedRun(final ObservableEmitter<BluetoothGatt> emitter, final QueueReleaseInterface queueReleaseInterface) {
final Action queueReleaseAction = new Action() {
@Override
public void run() {
queueReleaseInterface.release();
}
};
final DisposableSingleObserver<BluetoothGatt> disposableGattObserver = getConnectedBluetoothGatt()
.compose(wrapWithTimeoutWhenNotAutoconnecting())
// when there are no subscribers there is no point of continuing work -> next will be disconnect operation
.doFinally(queueReleaseAction)
.subscribeWith(disposableSingleObserverFromEmitter(emitter));
emitter.setDisposable(disposableGattObserver);
if (autoConnect) {
// with autoConnect the connection may be established after a really long time
queueReleaseInterface.release();
}
}
private DisposableSingleObserver<List<Issue>> getIssuesObserver() {
return new DisposableSingleObserver<List<Issue>>() {
@Override
public void onSuccess(List<Issue> issues) {
IssueAdapter listAdapter = (IssueAdapter) list.getAdapter();
MainActivity.this.issues.clear();
MainActivity.this.issues.addAll(new ArrayList<Issue>(issues));
list.getAdapter().notifyDataSetChanged();
}
@Override
public void onError(Throwable e) {
Toast.makeText(MainActivity.this, e.getMessage(), Toast.LENGTH_SHORT).show();
}
};
}
private void getCurrentWeather(String cityName, boolean isSearch) {
apiKey = getResources().getString(R.string.open_weather_map_api);
disposable.add(
apiService.getCurrentWeather(
cityName, Constants.UNITS, defaultLang, apiKey)
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribeWith(new DisposableSingleObserver<CurrentWeatherResponse>() {
@Override
public void onSuccess(CurrentWeatherResponse currentWeatherResponse) {
isLoad = true;
storeCurrentWeather(currentWeatherResponse);
storeCityInfo(currentWeatherResponse);
swipeContainer.setRefreshing(false);
if (isSearch) {
prefser.remove(Constants.LAST_STORED_MULTIPLE_DAYS);
}
}
@Override
public void onError(Throwable e) {
swipeContainer.setRefreshing(false);
try {
HttpException error = (HttpException) e;
handleErrorCode(error);
} catch (Exception exception) {
e.printStackTrace();
}
}
})
);
}
public void refreshList() {
disposable = Single.fromCallable(() -> {
Timber.d("fetching ride row list");
List<Ride> rideRowList = App.INSTANCE.db.rideDao().getAll();
Timber.d("rideRowList.size() = " + rideRowList.size());
return rideRowList;
})
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribeWith(new DisposableSingleObserver<List<Ride>>() {
@Override
public void onSuccess(List<Ride> rides) {
for (Ride ride : rides) {
Timber.d("ride = " + ride);
}
rideListAdapter.getRideList().clear();
rideListAdapter.getRideList().addAll(rides);
rideListAdapter.notifyDataSetChanged();
}
@Override
public void onError(Throwable e) {
Log.e(TAG, "onError: ", e);
}
});
}
public static <T> DisposableSingleObserver<T> disposableSingleObserverFromEmitter(final SingleEmitter<T> emitter) {
return new DisposableSingleObserver<T>() {
@Override
public void onSuccess(T t) {
emitter.onSuccess(t);
}
@Override
public void onError(Throwable e) {
emitter.tryOnError(e);
}
};
}
public static <T> DisposableSingleObserver<T> disposableSingleObserverFromEmitter(final ObservableEmitter<T> emitter) {
return new DisposableSingleObserver<T>() {
@Override
public void onSuccess(T t) {
emitter.onNext(t);
emitter.onComplete();
}
@Override
public void onError(Throwable e) {
emitter.tryOnError(e);
}
};
}
@Override
protected void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
setContentView(R.layout.activity_main);
ButterKnife.bind(this);
LinearLayoutManager layoutManager = new LinearLayoutManager(this);
recyclerView.setLayoutManager(layoutManager);
RxPaletteSampleApplication.get()
.getImgurApi()
.getAlbum("jx90V")
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new DisposableSingleObserver<ImgurResponse<Album>>() {
@Override
public void onSuccess(ImgurResponse<Album> albumImgurResponse) {
Adapter adapter = new Adapter(albumImgurResponse.data.images);
progressBar.setVisibility(View.GONE);
recyclerView.setVisibility(View.VISIBLE);
recyclerView.setAdapter(adapter);
}
@Override
public void onError(Throwable e) {
Toast.makeText(MainActivity.this, "Error: " + e, Toast.LENGTH_LONG).show();
Log.e("ERROR", "OnError", e);
}
});
}
public void bind(IJoinableGame joinableGame) {
IMapDefinition mapDefinition = joinableGame.getMap();
if (mapDefinition == null)
return;
hostNameTextView.setText(joinableGame.getName());
mapNameTextView.setText(mapDefinition.getMapName());
playerCountTextView.setText(mapDefinition.getMinPlayers() + "-" + mapDefinition.getMaxPlayers());
if (subscription != null) {
subscription.dispose();
}
subscription = PreviewImageConverter.toBitmap(mapDefinition.getImage(), limitImageLoadingSemaphore)
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribeWith(new DisposableSingleObserver<Bitmap>() {
@Override
public void onSuccess(Bitmap bitmap) {
mapPreviewImageView.setImageBitmap(bitmap);
}
@Override
public void onError(Throwable e) {
mapPreviewImageView.setImageDrawable(null);
}
});
}
public void bind(IMapDefinition mapDefinition) {
mapPreviewImageView.setImageDrawable(null);
nameTextView.setText(mapDefinition.getMapName());
playerCountTextView.setText(mapDefinition.getMinPlayers() + "-" + mapDefinition.getMaxPlayers());
if (showMapDates()) {
dateTextView.setText(dateFormat.format(mapDefinition.getCreationDate()));
}
if (subscription != null) {
subscription.dispose();
}
subscription = PreviewImageConverter.toBitmap(mapDefinition.getImage(), limitImageLoadingSemaphore)
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribeWith(new DisposableSingleObserver<Bitmap>() {
@Override
public void onSuccess(Bitmap bitmap) {
mapPreviewImageView.setImageBitmap(bitmap);
}
@Override
public void onError(Throwable e) {
mapPreviewImageView.setImageDrawable(null);
}
});
}
private DisposableSingleObserver<ResponseBody> getCommentObserver() {
return new DisposableSingleObserver<ResponseBody>() {
@Override
public void onSuccess(ResponseBody value) {
commentEditText.setText("");
Toast.makeText(MainActivity.this, "Comment created", Toast.LENGTH_LONG).show();
}
@Override
public void onError(Throwable e) {
e.printStackTrace();
Toast.makeText(MainActivity.this, "Can not create comment", Toast.LENGTH_SHORT).show();
}
};
}
public void execute(DisposableSingleObserver<T> singleObserver, Params params) {
Single<T> single = buildUseCase(params)
.subscribeOn(schedulerProvider.io())
.observeOn(schedulerProvider.ui());
disposables.add(single.subscribeWith(singleObserver));
}
/**
* Emits BluetoothGatt and completes after connection is established.
*
* @return BluetoothGatt after connection reaches {@link com.polidea.rxandroidble2.RxBleConnection.RxBleConnectionState#CONNECTED}
* state.
* @throws com.polidea.rxandroidble2.exceptions.BleDisconnectedException if connection was disconnected/failed before
* it was established.
*/
@NonNull
private Single<BluetoothGatt> getConnectedBluetoothGatt() {
// start connecting the BluetoothGatt
// note: Due to different Android BLE stack implementations it is not certain whether `connectGatt()` or `BluetoothGattCallback`
// will emit BluetoothGatt first
return Single.create(new SingleOnSubscribe<BluetoothGatt>() {
@Override
public void subscribe(final SingleEmitter<BluetoothGatt> emitter) {
final DisposableSingleObserver<BluetoothGatt> disposableGattObserver = getBluetoothGattAndChangeStatusToConnected()
// when the connected state will be emitted bluetoothGattProvider should contain valid Gatt
.delaySubscription(
rxBleGattCallback
.getOnConnectionStateChange()
.filter(new Predicate<RxBleConnection.RxBleConnectionState>() {
@Override
public boolean test(RxBleConnection.RxBleConnectionState rxBleConnectionState) {
return rxBleConnectionState == CONNECTED;
}
})
)
// disconnect may happen even if the connection was not established yet
.mergeWith(rxBleGattCallback.<BluetoothGatt>observeDisconnect().firstOrError())
.firstOrError()
.subscribeWith(disposableSingleObserverFromEmitter(emitter));
emitter.setDisposable(disposableGattObserver);
connectionStateChangedAction.onConnectionStateChange(CONNECTING);
/*
* Apparently the connection may be established fast enough to introduce a race condition so the subscription
* must be established first before starting the connection.
* https://github.com/Polidea/RxAndroidBle/issues/178
* */
final BluetoothGatt bluetoothGatt = connectionCompat
.connectGatt(bluetoothDevice, autoConnect, rxBleGattCallback.getBluetoothGattCallback());
/*
* Update BluetoothGatt when connection is initiated. It is not certain
* if this or RxBleGattCallback.onConnectionStateChange will be first.
* */
bluetoothGattProvider.updateBluetoothGatt(bluetoothGatt);
}
});
}
public void loadIssues(String username, String password, DisposableSingleObserver<List<Issue>> observer, String owner, String repository) {
githubApi.getIssues(Credentials.basic(username, password), owner, repository)
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(observer);
}