将 Hot and Cold Observable 与缓冲区合并答案

【问题标题】：Merge Hot and Cold Observable with buffer将 Hot and Cold Observable 与缓冲区合并
【发布时间】：2018-02-21 02:27:20
【问题描述】：

我有 2 个数据源。

获取项目列表当前状态的服务调用
这些项目的更新流

我正在尝试以在拨打服务电话时不会丢失任何更新的方式合并它们。

我做了一个简单的测试来进行实验，我认为它表明了我想要的

private val coldObservable = Observable.just(1, 2, 3, 4, 5)
private val subject = PublishSubject.create<Int>()
private val hotObservable = subject.hide()

@Test
fun test() {
    subject.onNext(10)

    val test = Flowable.concat(
            coldObservable.toFlowable(BackpressureStrategy.BUFFER), hotObservable.toFlowable(BackpressureStrategy.BUFFER)
    )
            .doOnSubscribe { subject.onNext(20) }
            .test()

    test.await(1, TimeUnit.SECONDS)
    test.assertNotComplete()
    test.assertNotTerminated()
    subject.onComplete()
    test.assertComplete()
    test.assertValues(1,2,3,4,5)
}

按原样，此测试通过。但我真正想要的是

test.assertValues(1,2,3,4,5,10,20)

我认为背压很容易让我坚持排放，但我猜不是因为它没有订阅。

有没有办法将这两个源合并成一个流？

【问题讨论】：

标签： java android kotlin rx-java2 reactive-streams

【解决方案1】：

这是一个与我正在寻找的效果相似的解决方案。它存储实时数据，然后在服务调用完成后对其进行处理

@Test
fun test2() {
    val randomString = "abc123"
    val realLog = ConcurrentHashMap<String, Boolean>()

    var isLive = false

    val subject = PublishSubject.create<Pair<String, Boolean>>()

    val subscription = subject.hide()
            .doOnNext {
                if (isLive) {
                    realLog[it.first] = it.second
                } else {

                    realLog[it.first+ randomString] = it.second
                }
            }.subscribe()

    subject.onNext(Pair("one", false))

    val coldTest = Observable.just("one","two","three")
            .map { Pair(it, true) }
            .doOnSubscribe {
                subject.onNext(Pair("twenty", false))
            }
            .doOnNext{
                realLog[it.first] = it.second
            }
            .doOnComplete {
                val iterator = realLog.keys.iterator()
                while(iterator.hasNext()){
                    val oldKey = iterator.next()
                    if(oldKey.contains(randomString)){
                        val newKey = oldKey.removeSuffix(randomString)
                        realLog[newKey] = realLog[oldKey]!!
                        realLog.remove(oldKey)
                    }
                }
                isLive = true
                subject.onNext(Pair("Thirty", false))
            }
            .test()

    coldTest.awaitTerminalEvent()
    coldTest.assertComplete()

    Assert.assertFalse(realLog["Thirty"]!!)

它不会发出单个流，我仍然不确定这是否可能。

【讨论】：

【解决方案2】：

我不确定我是否理解您的所有代码（我自己是 rx 的新手）。但是您的用例似乎与我的用例相同。

我的用例是这样的（因此您可以决定它是否相关）：我使用 HTTP 请求来获取数据结构的初始状态。但是一个网络套接字流不断地向该数据发送更新。如果我首先初始化数据结构，然后使用流对其进行更新，则更新将在其间的短时间内丢失。所以我需要先获取流，缓存其更新数据，同时等待 HTTP 请求返回数据结构，然后将来自流（在缓存中）的更新积压应用到数据上。随着来自套接字的数据进入，从那里开始更新。

我不知道 Kotlin，所以在 Java 中：

public static void main(String[] args) throws InterruptedException {
    //This one would correspond to the continuous stream of updates.
    //It sends Long, one each second
    Observable<Long> listUpdater = Observable.interval(1, TimeUnit.SECONDS);
    //Acts as both observable and observer, that replays everything to its own
    //observers.
    Subject<Long> replaySubjbect = ReplaySubject.create();
    //Start getting items from listUpdater immediately
    listUpdater.subscribe(replaySubjbect);
    //To show that listUpdater moves along regardless of what you are doing, as would
    //be the case for example with a websocket.
    listUpdater.subscribe(i -> System.out.println("I am also watching, i=" + i));
    //This callable corresponds to whatever returns the initial state of the list,
    //for example some http request
    Callable<List<Long>> clbl = () -> Arrays.asList(1L, 2L, 3L, 4L, 5L);
    //This is where the merging takes place, although we are not using one of the merging
    //methods, but scanWith. 
    Observable<List<Long>> theStream = replaySubjbect.scanWith(clbl, (List<Long> list, Long l) -> {
        //This updater adds l to each element of the list
        System.out.println("I am adding " + l + " to each element in the list");
        return list.stream().mapToLong(k -> k + l).boxed().collect(Collectors.toList());            
    });
    //Simulate a response time for the http request, or whatever it is the callable is doing
    Thread.sleep(3000);
    //Now we get the stream of lists, updated for each of the update data sent by the
    //original listUpdater
    theStream.subscribe(list -> System.out.println("theStream sent me this: " + list));
    //Just to see how it works, we sleep some more
    Thread.sleep(5000); 
}

这个输出：

I am also watching, i=0
I am also watching, i=1
I am also watching, i=2
theStream sent me this: [1, 2, 3, 4, 5] <- cahce retrieval starts
I am adding 0 to each element in the list 
theStream sent me this: [1, 2, 3, 4, 5] 
I am adding 1 to each element in the list 
theStream sent me this: [2, 3, 4, 5, 6] 
I am adding 2 to each element in the list 
theStream sent me this: [4, 5, 6, 7, 8]
I am also watching, i=3     
I am adding 3 to each element in the list <- from here on, update list as updates come in
theStream sent me this: [7, 8, 9, 10, 11]
I am also watching, i=4
I am adding 4 to each element in the list
theStream sent me this: [11, 12, 13, 14, 15]
I am also watching, i=5
I am adding 5 to each element in the list
theStream sent me this: [16, 17, 18, 19, 20]
I am also watching, i=6
I am adding 6 to each element in the list
theStream sent me this: [22, 23, 24, 25, 26]
I am also watching, i=7
I am adding 7 to each element in the list
theStream sent me this: [29, 30, 31, 32, 33]

扫描： http://reactivex.io/RxJava/javadoc/io/reactivex/Observable.html#scanWith-java.util.concurrent.Callable-io.reactivex.functions.BiFunction-

您也可以使用扫描（直接提供初始数据，而不是使用 Callable）： http://reactivex.io/RxJava/javadoc/io/reactivex/Observable.html#scan-R-io.reactivex.functions.BiFunction-

【讨论】：