具有自定义计数标准的 RxJava 缓冲区/窗口

我有一个 Observable，它发出许多对象，我想使用以下方法对这些对象进行分组：window or buffer运营。但是，不是指定count用于确定窗口中应有多少对象的参数我希望能够使用自定义标准。

例如，假设可观察对象正在发出 a 的实例Message像下面这样的类。

class Message(
   val int size: Int
)

我想根据它们的消息实例来缓冲或窗口size变量不仅仅是它们的计数。例如，获取总大小最多为 5000 的消息窗口。

// Something like this
readMessages()
    .buffer({ message -> message.size }, 5000)

是否有捷径可寻？

首先我必须承认，我不是 RxJava 专家。 我只是发现你的问题具有挑战性，并试图找到解决方案。

有一个window()带参数的函数boundaryIndicator。你必须创建一个Publisher/ Flowable如果达到窗口大小，则发出一个项目。

在示例中我创建了一个对象windowManager用作boundaryIndicator。在里面onNext回调我调用windowManager并给它一个打开新窗口的机会。

val windowManager = object {
    lateinit var emitter: FlowableEmitter<Unit>
    var windowSize: Long = 0

    fun createEmitter(emitter: FlowableEmitter<Unit>) {
        this.emitter = emitter
    }

    fun openWindowIfRequired(size: Long) {
        windowSize += size
        if (windowSize > 5) {
            windowSize = 0
            emitter.onNext(Unit)
        }
    }
}

val windowBoundary = Flowable.create<Unit>(windowManager::createEmitter, BackpressureStrategy.ERROR)

Flowable.interval(1, TimeUnit.SECONDS).window(windowBoundary).subscribe {
    it.doOnNext {
        windowManager.openWindowIfRequired(it)
    }.doOnSubscribe {
        println("Open window")
    }.doOnComplete {
        println("Close window")
    }.subscribe {
        println(it)
    }
}