标签: rx-java2

我想使用swagger codegen生成sdk,它可以让我生成带有Observable的sdk作为回调,如下所示:

@POST("oauth/token")
Observable <TokenResponse> getRepository(@Query("grant_type")String grantType);

我已经使用RxJava 1了一段时间,但我想查看RxJava 2.在RxJava 1中,我可以按如下方式发出每个列表项:

List<String> list = ...
Observable.from(list)
    .filter(str -> str.contains("Help")
    .subscribe(...);

但是,如何使用RxJava2实现相同的功能呢？我试过使用以下内容,但我似乎无法通过以下内容:

Observable.fromArray(list)
// this now passes a list into the stream - there is no Observable::from

我有这个代码在Rx Java 1中包装一个回调并且它编译得很好,但是现在我已经切换到RX Java 2它不能编译...... Rx Java 2中的等价物是什么？

return Observable.fromEmitter(new Action1<AsyncEmitter<Integer>>() {
            @Override
            public void call(AsyncEmitter<Integer> emitter) {

                transObs.setTransferListener(new TransferListener() {
                    @Override
                    public void onStateChanged(int id, TransferState state) {
                        if (state == TransferState.COMPLETED)
                            emitter.onCompleted();
                    }

                    @Override
                    public void onProgressChanged(int id, long bytesCurrent, long bytesTotal) {

                    }

                    @Override
                    public void onError(int id, Exception ex) {
                        emitter.onError(ex);
                    }
                });

                emitter.setCancellation(new AsyncEmitter.Cancellable() {
                    @Override
                    public void cancel() throws Exception {

                        transObs.cleanTransferListener();
                    }
                });
            }
        }, AsyncEmitter.BackpressureMode.BUFFER);

更新:

我想出了这个,但你是否需要处理背压,因为它是一个oncreate电话？

 return Observable.create(new ObservableOnSubscribe<List<DigitsUser>>() {

        @Override …

当我在带有API 21的手机上的改装API中使用Observable时出现此错误:

java.lang.NoClassDefFoundError: io.reactivex.Observable

但是在使用API​​ 19的电话或使用API​​ 23的模拟器上它可以工作.

这是我的API接口:

import io.reactivex.Observable;
import retrofit2.Response;
import retrofit2.http.Body;
import retrofit2.http.POST;
public interface ApiService {
    /**
     * Created by Mohsen on 5/10/2017.
     *
     */
    @POST("/cp/api/")
    Observable<Response<Integer>> Get_BuyBox_Count(@Body Object request);
}

这是我的改造设置:

    @Provides
    @Application_Scope
    @Store_Retrofit_Qualifier
    public Retrofit Store_retrofit(OkHttpClient client) {
        return new Retrofit.Builder()
                .baseUrl(Urls.Sotre_Base_Url)
                .client(client)
                .addCallAdapterFactory(RxJava2CallAdapterFactory.create())
                .addConverterFactory(JSONConverterFactory.create())
                .addConverterFactory(GsonConverterFactory.create())
                .build();
    }

我的依赖:

    compile 'com.squareup.retrofit2:retrofit:2.2.0'
    compile 'com.squareup.retrofit2:converter-gson:2.1.0'
    compile 'io.reactivex.rxjava2:rxjava:2.0.1'
    compile 'io.reactivex.rxjava2:rxandroid:2.0.1'
    compile 'com.jakewharton.retrofit:retrofit2-rxjava2-adapter:1.0.0'

我正在尝试通过仅创建一个实例并将其通过流上的方法通过它们的实例来重用Observerfor Single和Observablestream ：DisposableSingleObserver/DisposableObserversubscribeWith()

public class SomeClass {
    private DisposableSingleObserver<Object> observer;

    public SomeClass() {
        observer = new DisposableSingleObserver<Object>() {
            @Override
            public void onSuccess(Object object) {
                ...
            }

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

    public void doSomeStuff() {
        singleStream.subscribeOn(...)
            .observeOn(...)
            .subscribeWith(observer);
    }
}

上面的代码导致ProtocolViolationException当我尝试通过消息多次订阅单个观察者实例时：

io.reactivex.exceptions.ProtocolViolationException：不允许多次使用com.package.name.SomeClass $ 1进行订阅。请创建com.package.name.SomeClass $ 1的新实例，然后将其订阅到目标源。

因此，我对代码进行了如下修改：

public class SomeClass {
    ...

    public void doSomeStuff() {
        if (observer != null) {
            observer.dispose();
        }

        singleStream.subscribeOn(...)
            .observerOn(...)
            .subscribeWith(observer); …

嗨,我正在阅读rxjava2哪些基本上是异步操作.我发现它有一个运算符概念,AsyncTask在android中没有.

还有什么rxjava2不同AsyncTask？

我只是将代码切换asynctask到,rxjava2并且我在我的nexus中随机获取此异常:

在Galaxy s6 Edge中调用Camera.release()之后正在使用相机

以下是我的代码 -

班级相机:

 public class Cameras {


    private static final String TAG = Cameras.class.getSimpleName();

    private static final String SP_CAMERA_ID = "camera_id";

    private static final int NO_NEXT_TASK = 0;

    private static final int NEXT_TASK_RELEASE_COMPLETE = 1;

    private static final int NEXT_TASK_SWITCH_COMPLETE = 2;

    private static final int NEXT_TASK_START_PREVIEW = 3;


    private Camera camera;

    private int currentCameraId = -1;


    private Camera.PreviewCallback previewCallback;

    private byte[] buffer1, buffer2, buffer3;

    private SurfaceTexture surfaceTexture;


    private Listener …

在创建像这样的Observable时:

public void foo() {

    Observable observable = Observable.fromCallable(() -> {
        bar();
        return "";
      })
      .doOnSubscribe(disposable -> System.out.println("onSubscribe"))
      .doOnDispose(() -> System.out.println("onDispose"));

    Disposable disposable = observable.subscribe();
    disposable.dispose();    
}

private void bar() {
    System.out.println("bar");
}

doOnSubcribe叫,doOnDispose不叫.

这是为什么？

我正在尝试在RxJava中找到一个以特定方式限制的运算符:

• 发出一个元素
• 在元素发出后的一段时间内,过滤掉所有其他发射元素

我似乎无法找到符合此行为的一个.我看了一些类似的,但似乎没有一个是正确的.

• debounce/ throttleWithTimeout- 发出在它们之间短时间内发出的元素序列中的最后一个元素

• sample/ throttleLast它查看设置的时间间隔并发出每个时间间隔中的最后一个元素.

• throttleFirst它查看设定的时间间隔并在每个时间间隔内发出第一个元素.这似乎是最接近我想要的,但并不完全相同.

是否有任何我可以使用的RxJava运算符与此匹配？它似乎是一个有用的用例.

TO BE CLEAR:

The most likely part of the code which has the problem is the connect function, which you can find in the code block.

EDIT:

I've had a good dig through LogCat and found something interesting (this occurred the exact moment enableNetwork was called):

2018-12-04 20:13:14.508 1315-7000/? I/WifiService: enableNetwork uid=10158 disableOthers=true
2018-12-04 20:13:14.508 1315-1607/? D/WifiStateMachine: connectToUserSelectNetwork netId 49, uid 10158, forceReconnect = false
2018-12-04 20:13:14.541 1315-1607/? D/WifiConfigStore: Writing to stores completed in 14 ms.
2018-12-04 20:13:14.541 1315-1607/? E/WifiConfigManager: …