Rx Java Android: ошибка несовместимых типов наблюдателя

Question

У меня есть следующий сетевой вызов Asyn c, который я хотел бы сделать, используя Rx Java и Rx Android. Я включил последние RxJava (3.0.1) и RxAndroid (3.0.0) в зависимости проекта, но все еще получаю следующую ошибку в блоке Subscription subscription = ....:

Несовместимые типы

Обязательно: org.reactivestreams.Subscription

Найдено: void

Почему это так? Может кто-нибудь сказать мне, как заставить этот Rx Java наблюдаемый работать?

Любая помощь оценена, спасибо.

Subscription subscription = getServerResponse(et_id.getText().toString())// error highlighted in this whole block
       .subscribeOn(Schedulers.io())
       .observeOn(AndroidSchedulers.mainThread())
       .subscribe(new Observer<String>() { 
            @Override
            public void onSubscribe(@NonNull Disposable d) {
            }

            @Override
            public void onNext(String bitmap) {
               // Handle result of network request
            }

            @Override
            public void onError(@NonNull Throwable e) {
               // Update user interface to handle error
            }

            @Override
            public void onComplete() {
               // Update user interface if needed
            }
       });

public Observable<String> getServerResponse(String string) {
    return Observable.fromCallable(new Callable<String>() {
        @Override
        public String call() throws Exception {
            BufferedReader inFromServer = null;
            String response = "";
            Socket clientSocket = null;
            try {
                clientSocket = new Socket(serverHostname, serverPort);
                DataOutputStream outToServer = new DataOutputStream(clientSocket.getOutputStream());
                inFromServer = new BufferedReader(new InputStreamReader(clientSocket.getInputStream()));
                outToServer.writeBytes(string + "\n");
                response = inFromServer.readLine();
                clientSocket.close();
            } catch (IOException e) {
                e.printStackTrace();
                response = MainActivity.this.getString(R.string.result_serverError);
            }
            return response;
        }
    });
}

Весь мой импорт для реактивных java:

import io.reactivex.rxjava3.android.schedulers.AndroidSchedulers;
import io.reactivex.rxjava3.annotations.NonNull;
import io.reactivex.rxjava3.core.Observable;
import io.reactivex.rxjava3.core.Observer;
import io.reactivex.rxjava3.disposables.Disposable;
import io.reactivex.rxjava3.schedulers.Schedulers;

Answer 1

попробуйте использовать одноразовый композит. Это было бы что-то вроде этого:

private CompositeDisposable compositeDisposable = CompositeDisposable();
compositeDisposable.add(getServerResponse(et_id.getText().toString())
   .subscribeOn(Schedulers.io())
   .observeOn(AndroidSchedulers.mainThread())
   .subscribe(new Observer<String>() { 
        @Override
        public void onSubscribe(@NonNull Disposable d) {
        }

        @Override
        public void onNext(String bitmap) {
           // Handle result of network request
        }

        @Override
        public void onError(@NonNull Throwable e) {
           // Update user interface to handle error
        }

        @Override
        public void onComplete() {
           // Update user interface if needed
        }
   }));

Не забудьте сделать CompositeDisposable.dispose () для метода onDestroy вашей деятельности.

Дайте мне знать, если это работает для вас!