标签: rx-java2

我在Android Studio中为现有的应用程序模块添加了一个新的库模块.主要的区别是添加了RxJava 2和Retrofit 2.更新了new模块的build.gradle后,我开始得到下一个错误:

错误:任务':app:transformResourcesWithMergeJavaResForBetaNewApiDebug'的执行失败.com.android.build.api.transform.TransformException:com.android.builder.packaging.DuplicateFileException:在APK META-INF/rxjava.properties中复制的重复文件File1:C:\ Users\Gaket.gradle\_caches\modules-2\files-2.1\io.reactivex.rxjava2\rxjava\2.0.2\cfccdd18cdfbe7b4773d42c9f3512eeafbe5cbf9\rxjava-2.0.2.jar File2:C:\ Users\Gaket.gradle\_caches\modules-2\files-2.1\io.reactivex\rxjava\1.1.5\ece7b5d0870e66d8226dab6dcf47a2b12afff061\rxjava-1.1.5.jar

我看到RxJava存在一些问题(我想尝试RxJava 2,在这里我看到RxJava 1和RxJava 2).在添加所有依赖项后立即出现问题,我们不在RxJava主应用程序中使用.

更新: 正如我在下面提到的,我检查了三个主题(您可以在下面看到).这个问题:可能的重复类似于我在下面提到的那个: 这一个.他们两个都有解决方法,没有考虑我在答案中强调的真正问题.而答案是 "没有的生产就绪适配器从改造2至RxJava 2的时刻"(以下答案中描述的全部细节).

我检查了几个主题:首先 是一个具体问题的具体解决方案,第二个谈论rxbindings,但我不使用它们,其他一些也没有解决我的问题或看起来像这个完全解决方法.我试图谷歌"rxjava 2和改造2"但没有找到解决方案.

这是build.gradle图书馆:

apply plugin: 'com.android.library'
apply plugin: 'com.jakewharton.butterknife'

android {
compileSdkVersion 25
buildToolsVersion "25.0.0"

defaultConfig {
    minSdkVersion 15
    targetSdkVersion 24
    versionCode 1
    versionName "1.0"

    testInstrumentationRunner "android.support.test.runner.AndroidJUnitRunner"

}
buildTypes {
    release {
        minifyEnabled false
        proguardFiles getDefaultProguardFile('proguard-android.txt'), 'proguard-rules.pro'
    }
}
}



dependencies {
compile fileTree(dir: 'libs', include: …

我遇到的问题始终是使用combineLatest运算符获得的组合中的最后一个值.

我有2个热流道(a,b)以高频率生成事件(每100毫秒一个事件):

Flowable<OrderBook> flowA = sourceA.getObservableOrderBook(value);
Flowable<OrderBook> flowB = sourceB.getObservableOrderBook(value);

结合combineLatest,需要将近300毫秒才能完成它的工作.

Flowable<OrderBookCouple> combined = Flowable.combineLatest(flowA, flowB,        OrderBookCouple::new).observeOn(Schedulers.newThread());
combined.subscribe((bookCouple) -> {
                System.out.println("A timestamp: " + bookCouple.aOrderBook.getTimeStamp());
                System.out.println("B timestamp: " + bookCouple.bOrderBook.getTimeStamp());
                Thread.sleep(300);
            }

在一次执行组合器之后,我想处理最后生成的事件组合,意思是(lastA,lastB).

组合流的默认行为是将所有事件组合缓存在其自己的缓冲区中,以便组合流可以接收非常旧的组合,并且此时间隙正在爆炸.

我应该如何更改我的代码以禁用此缓冲区并始终接收最后一个组合？

我有一个示例Java代码使用方法引用,我想重写为Kotlin.Java版本使用方法参考,解决方案简短明了.但另一方面,我不能在Kotlin中使用方法参考.我设法编写的唯一版本如下所示.似乎Function3 { s: String, b: Boolean, i: Int -> combine(s, b, i) }可以用更干净的方式编写(如果可能的方法参考将是完美的).

我是Kotlin的新手,所以我会感激任何线索.

Java的

import io.reactivex.Observable;

public class TestJava {

    Observable<String> strings() {
        return Observable.just("test");
    }

    Observable<Boolean> booleans() {
        return Observable.just(true);
    }

    Observable<Integer> integers() {
        return Observable.just(1);
    }

    void test() {
        Observable.combineLatest(strings(), booleans(), integers(),
                this::combine);
    }

    double combine(String s, boolean b, int i) {
        return 1.0;
    }
}

科特林

import io.reactivex.Observable
import io.reactivex.functions.Function3

class TestKotlin {

    fun strings(): Observable<String> {
        return Observable.just("test")
    }

    fun booleans(): Observable<Boolean> { …

我有两个单身人士，我将其合并为一对。

假设对于这些来源，我们有两种方法：

private Single<String> single1() {} 
private Single<String> single2() {}

现在我将它们合并为 Single<Pair<String, String>

private Single<Pair<String, String> combineSingles() {
    Single.zip(single1(), single2(), BiFunction<String, String, Pair<String, String>>) (t1, t2) -> new Pair(t1, t2))
}

当其中一种方法（single1或single2）返回Single.error（）时，我得到的UndeliverableException原因错误未在zip运算符中处理。我想Single.error()从combineSingles()方法中返回，如何实现呢？

我有一个调用Web服务的方法，我认为该方法在IO线程上运行，直到服务停止并且UI冻结为止。

所以我开始了一些简单的测试来检查线程

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


public void test() {

    disposableRx.add(
            Observable.just(1, 2)
                    .subscribeOn(Schedulers.io())
                    .observeOn(AndroidSchedulers.mainThread())
                    .doOnNext(new Consumer<Integer>() {
                        @Override
                        public void accept(Integer integer) throws Exception {
                            System.out.println("Emitting item on: " + Thread.currentThread().getName());
                        }
                    })
                    .map(new Function<Integer, Integer>() {
                        @Override
                        public Integer apply(@NonNull Integer integer) throws Exception {
                            System.out.println("Processing item on: " + Thread.currentThread().getName());
                            return integer * 2;
                        }
                    })

                    .subscribeWith(new DisposableObserver<Integer>() {
                        @Override
                        public void onNext(@NonNull Integer integer) {
                            System.out.println("Consuming item on: " + Thread.currentThread().getName());
                        }

                        @Override
                        public void onError(@NonNull …

我onErrorReturn用来发出特定的项目，而不是onError如果可观察的对象遇到错误则调用它：

Observable<String> observable = getObservableSource();
observable.onErrorReturn(error -> "All Good!")
          .subscribeOn(Schedulers.io())
          .observeOn(Schedulers.trampoline())
          .subscribe(item -> onNextAction(),
              error -> onErrorAction()
          );

这工作正常，但我只想在onErrorReturn满足某些条件的情况下使用错误。就像从catch块内部抛出异常一样。

就像是：

onErrorReturn(error -> {
    if (condition) {
        return "All Good!";
    } else {
        // Don't consume error. What to do here?
        throw error; // This gives error [Unhandled Exception: java.lang.Throwable]
    }
});

有没有办法将错误从内部传播到可观察的链中，onErrorReturn就像onErrorReturn从未出现过一样？

我想缓冲元素并在x时间内没有新元素时将它们作为集合发出。怎么做？

例如给定的输入

INPUT    TIME
1        0  
2        0
3        100
4        150
5        400
6        450
7        800   

如果我的x = 200我想发射 {1, 2, 3, 4}, {5, 6}, {7}

我尝试过的方法很简单buffer()，但是随着时间的推移却无法消除抖动。我也试过throttleFirst()源和flatMap()它buffer().take(1)的来源里面flatMap，它的工作原理类似，但不完全是任意的。

getPlaces我的存储库中有一个方法：

override fun getPlaces(filter: FilterRequest): Flowable<List<Place>> {
    return from(placesApi.filter(filter))
            .doOnSuccess {
                placesDao.savePlaces(it)
            }
            .flatMapPublisher { it ->
                placesDao.getPlaces(it.map { it.placeId })
            }
}

此方法从api收集结果，然后将结果保存在数据库中，并返回一个带有flow的流，其中包含通过id从数据库中检索到的位置Flowable：

@Query("select * from Places where placeId in (:placesIds)")
fun getPlaces(placesIds: List<String>) : Flowable<List<Place>>

现在，每次我更改其中一个对象时，就可以在整个应用程序中看到所有更改。

现在，我想将这些结果与到当前位置的距离结合起来，如下所示：

 override fun addDistanceToPlaces(req: Flowable<List<Place>>): Flowable<List<Place>> {
        return req
                .zipWith(getLastLocation().toFlowable(BackpressureStrategy.LATEST),
                        BiFunction<List<Place>, Location, List<Place>> { places, location ->
                            places.forEach {
                                var placeLocation = Location(it.placeName)
                                placeLocation.latitude = it.latitude
                                placeLocation.longitude = it.longitude

                                it.distance = location.distanceTo(placeLocation)
                            }
                            places.sortedBy {
                                it.distance
                            }
                        })
                .onErrorResumeNext …

我正在使用自己的API,我希望使用RxJava链接一些分页结果.我使用基于游标的分页.(想象有50个用户):

{
    "data":{
        "status":"ok",
        "total":988, //users total
        "has_next_page":true,
        "end_cursor":"AQAxd8QPGHum7LSDz8DnwIh7yHJDM22nEjd",
        "users":[{"id":"91273813",
                "username":"codergirl",
                "full_name":"Code Girl",
                "picture_url":"https://cdn.com/21603182_7904715668509949952_n.jpg",
                },
                ...
                ]
        }
}

现在,我正在使用改造这样的前50个结果:

public class DataResponse {
    @SerializedName("end_cursor")
    private String end_cursor;

    @SerializedName("users")
    private JsonArray users;

    @SerializedName("has_next_page")
    private Boolean has_next_page;

    public boolean hasNextCursor(){
        return has_next_page;
    }
    public String endCursor(){
        if (hasNextCursor()){
            return end_cursor;
        }
        return "";
    }
    public JsonArray getUsers(){
        return users;
    }
}

然后:

public interface MyService  {
    @GET( "/users")
    Observable<DataResponse> getUsers(
            @Query("cursor") String cursor,
    );
}

和

MyService service = RetrofitClient.getInstance();
service.getUsers() …

rxjava2依赖和rxkotlin依赖之间有什么区别。如果我正在使用rxkotlin依赖项，我是否还需要添加rxjava2依赖项。

implementation 'io.reactivex.rxjava2:rxkotlin:x.y.z'
// do i need to add the below dependencies also?
implementation 'io.reactivex.rxjava2:rxjava:2.2.4'
implementation 'io.reactivex.rxjava2:rxandroid:2.1.0'