标签: rx-java2

我正在从Realm读取/写入这个异常

06-19 09:49:26.352 11404-11404/******E/ContentValues:loadData:OnError Realm从错误的线程访问.Realm对象只能在创建它们的线程上访问.java.lang.IllegalStateException:从错误的线程访问Realm.Realm对象只能在创建它们的线程上访问.在io.realm.BaseRealm.checkIfValid(BaseRealm.java:385)的io.realm.alalm.OrderedRealmCollectionImpl.size(OrderedRealmCollectionImpl.java:307)中的io.realm.RealmResults.isLoaded(RealmResults.java:115). RealmResults.size(RealmResults.java:60)位于/******.lambda $ loadData $ 0(SplashPresenter.java:42)的/./的.***$ Lambda $ 1.test(未知来源)at io.reactivex.internal.operators.observable.ObservableFilter $ FilterObserver.onNext(ObservableFilter.java:45)at io.reactivex.observers.SerializedObserver.onNext(SerializedObserver.java) :111)ato.reactivex.internal.operators.observable.ObservableDelay $ DelayObserver $ 1.run(ObservableDelay.java:84)at io.reactivex.internal.schedulers.ScheduledRunnable.run(ScheduledRunnable.java:59)at io.reactivex .internal.schedulers.ScheduledRunnable.call(ScheduledRunnable.java:51)at java.util.concurrent.FutureTask.run(FutureTask.java:237)at java.util.concurrent.ScheduledThreadPoolExecutor $ Schedul edFutureTask.run(ScheduledThreadPoolExecutor.java:272)java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1133)java.lang上的java.util.concurrent.ThreadPoolExecutor $ Worker.run(ThreadPoolExecutor.java:607) .Thread.run(Thread.java:761)

这是代码:

  mSubscribe = Observable.just(readData())
            .delay(DELAY, TimeUnit.SECONDS)
            .filter(value -> !value.isEmpty())
            .switchIfEmpty(createRequest())
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread()).subscribe(data -> {
                getView().hideLoading();
                writeData(data);
            }, 
           (throwable -> {
            }));

读数据

  private List<CategoryModel> readData() {
    Realm defaultInstance = Realm.getDefaultInstance();
    List<CategoryModel> title = defaultInstance.where(CategoryModel.class).findAllSorted("title");

    defaultInstance.close();
    return title;
}

写数据

private void writeData(List<CategoryModel> categoryModels) {

        try {
            Realm …

我使用 .just() 运算符编写了一个 Observable ，如下代码所示。我要做的是向操作员添加延迟，以便在将整数值发送给观察者之前，它应该延迟几秒钟

就像是：

mAsyncObservable = rx.Observable.create(new rx.Observable.OnSubscribe<Integer>() {
        @Override
        public void call(Subscriber<? super Integr> subscriber) {
            SystemClock.sleep(3000);

            subscriber.onNext(10);
            subscriber.onCompleted();
        }
    });

如何使用 .just() 运算符转换相同的代码？

代码：

Subscriber<Integer> integerSubscriber = new Subscriber<Integer>() {
    private final String SubscriberTag = "integerSubscriber";
    @Override
    public void onStart() {
        super.onStart();
        Log.w(TAG, "onStart->" + SubscriberTag);
    }

    @Override
    public void onCompleted() {
        Log.w(TAG, "onCompleted->" + SubscriberTag);
    }

    @Override
    public void onError(Throwable e) {
        Log.w(TAG, "onError->" + SubscriberTag + e.getMessage().toString());
    }

    @Override
    public void onNext(Integer i) …

我是否需要包装Android SharedPreferences类？如果是的话,能否请您提供一个简单的工作示例？

我知道如何使用它SharedPreferences,但是当它包装它并提供Dagger 2和RxJava2时,我很困惑.

我正在尝试解决一个问题，即 Retrofit 2 返回 null 作为响应。有趣的是，在模拟器中一切正常，数据被接收并显示，但是当我构建 APK 时，同样的事情因 NullPointerException 而崩溃。我已经评论了发生这种情况的行。

我的 build.gradle 依赖项：

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

初始化改造：

retrofit = new Retrofit.Builder()
        .baseUrl(<url>)
        .addConverterFactory(GsonConverterFactory.create())
        .addCallAdapterFactory(RxJava2CallAdapterFactory.create())
        .build();

界面：

@GET(<url>)
Flowable<DataResponse> getDataFromApi(@Query("offset") int offset, @Query("limit") int limit);

API调用：

retrofit
  .getDataFromApi(0, 10)
  .subscribeOn(Schedulers.io())
  .observeOn(AndroidSchedulers.mainThread())
  .subscribe(new Subscriber<DataResponse>() {
        @Override
        public void onSubscribe(Subscription s) {
            s.request(Long.MAX_VALUE);
        }

        @Override
        public void onNext(DataResponse dataResponse) {
          // HERE I GET NULL IN PRODUCTION   
        }

数据响应POJO：

public class ServicesResponse {
  private List<Item> items; …

我知道如何在RxJava 2 中做到这一点。

而且我知道RxKotlin如何帮助解决类似问题。

但似乎 RxKotlin.Observables 没有列表重载的这个辅助函数，我无法弄清楚。你会怎么做？

因为我是新手RxJava试图运行以下代码,但它显示我

Cannot resolve symbol `Obserable.onSubscribe`

代码如下

Observable<String> fetchFromGoogle = Observable.create(new Observable.OnSubscribe<String>() {
            @Override
            public void call(Subscriber<? super String> subscriber) {
                try {

                }catch(Exception e){
                    subscriber.onError(e); // In case there are network errors
                }
            }
        });

已添加以下条目 gradle

compile 'io.reactivex.rxjava2:rxandroid:2.0.1'
compile 'io.reactivex.rxjava2:rxjava:2.1.8'

对于java 8的兼容性

 compileOptions {
        sourceCompatibility JavaVersion.VERSION_1_8
        targetCompatibility JavaVersion.VERSION_1_8
    }

我有一个Fragment A使用Retrofit在中调用API的onCreateView。
结果将用于显示对象列表。

Fragment A还有一个按钮，可以Fragment B通过提交FragmentTransaction和自身导航到。（已Fragment A onDestroyView触发）

一切正常，但是当我从后Fragment B到Fragment A（Fragment A onCreateView触发），API不叫。

这只是发生如果我叫Disposable.dispose()的onDestroyView，这是企图以防止内存泄漏。

我期望什么

由于Observable每次onCreateView触发都会创建新的，因此应该将它们与先前处理的无关Disposables。因此，将再次调用API并触发回调。对象列表应刷新。

我观察到的

onCreateView触发后，告诉Retrofit调用API的行也被触发。但是没有.map()触发任何回调（包括）。由于我也记录了网络流量，因此，我很确定实际上没有进行网络通信。看起来RxJava或Retrofit决定停止流。

问题

如果不使用，当然可以避免Disposable.dispose()。但是我确实想防止内存泄漏。
正确的方法是什么？

我的代码（简体）

片段A：

public class BeansFragment extends BaseFragment {

   ...

    @Nullable
    @Override
    public View onCreateView(LayoutInflater inflater, @Nullable ViewGroup container, @Nullable Bundle savedInstanceState) {
        ...
        retrofit.getSomeBeans()
                .subscribeOn(Schedulers.io())
                .unsubscribeOn(Schedulers.io()) …

我有一个带有 DiffUtil 功能的 RecyclerView 适配器。当 DiffUtil 完成其魔法时，我想使用 DataObserver 来通知片段。但是，看起来 DiffUtil 不会调用观察者。

适配器

private val items: MutableList<Message>
private val updater = PublishSubject.create<MutableList<Message>>()

init {
    items = ArrayList()

    updater
        .debounce(500, TimeUnit.MILLISECONDS)
        .subscribeOn(Schedulers.computation())
        .map { Pair(it, DiffUtil.calculateDiff(DiffUtilCallback(it, items), true)) }
        .doOnNext { setItems(it.first) }
        .map { it.second }
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe { diffResult.dispatchUpdatesTo(this) }
}

fun updateData(msgs: MutableList<Message>) = updater.onNext(msgs)

private fun setItems(newItems: MutableList<Message>) {
    items.clear()
    items.addAll(newItems)
}

分段

adapter.registerAdapterDataObserver(object : RecyclerView.AdapterDataObserver() {
    override fun onChanged() {
        //is not called
    }
})

如何使 AdapterDataObserver …

我正在使用带有 rxjava 的改造 2 来发送休息请求。和 Gson 作为 json 解析。

如果 Gson 无法解析 json 字符串，则抛出java.lang.IllegalStateException一些消息。我的问题是我不知道这个错误发生在哪个网址上。

如果错误类型是，HttpException我可以从中获取 url，但在其他类型中java.lang.IllegalStateException，TimeoutException只能访问错误消息，而我看不到 url。

我怎样才能在 rxjava 中实现 url 并改造捕获。

那是我的代码：

mainApi.getSomething(offset, limit, distance, latitude, longitude)
            .subscribeOn(Schedulers.newThread())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe({it:Response<Something>
                val entities = it.body()?.entities
                callBack.onResponse(it.raw(), entities)
            }, {//it:throwable
                if (t is HttpException) {
                   Log.d("logUrl", "HttpException : ${t.response().raw().request().url()} : ${t.code()} :${t.message()}")
                } else//something like IllegalStateException or TimeoutException
                     Log.d("logUrl", "Failure : ${t.localizedMessage}")
                })
               })

我想调用一个 APIapi1并将API返回的输出传递给第二个 API api2。第一个 API 是获取请求，第二个 API 是POST请求。因此，api1返回 aSingle<String>并api2返回 a Completable。

函数看起来像这样：

// api1
public Single<String> getToken() {
  ...
}
// api2
public Completable saveTokenToBackend(String token, String userId) {
 ...
}

我想将这两个操作链接在一起。订阅者只关心获取令牌并保存它的过程是否成功。因此，最终操作链的返回类型应该是 a Completable。但是，当我这样做时，API 要求api2停止发生。仅api1根据日志成功运行。

Single<Completable> r1 = getToken().map(t -> saveTokenToBackend(t, userId));
Completable r2 = Completable.fromSingle(r1);

我在这里更广泛的问题是，我如何将响应从 Single 链接到 Completable？

第二个问题是为什么上面的代码不起作用？

：：编辑：：

根据评论中的建议，我尝试了：

public Completable getAndSaveToken() {
    getToken().flatMapCompletable(t -> saveTokenToBackend(t, "dummyuser");
}

在我的应用程序代码中，我正在做： …