RxJava滑动窗口

Question

我有一个observable，它会发出数据，我想先缓冲它三秒钟，然后在初始缓冲区之后必须滑动一秒钟。这更像是buffer(timespan,unit,skip)跳过的时间段。

样品：

ObservableData,TimeStamp : (5,1),(10,1.5),(30,2.8),(40,3.2),(60,3.8),(90,4.2) 

ExpectedList : {5,10,30},{10,30,40,60},{30,40,60,90} 

我可以通过创建一个自定义的操作实现这一目标。我只想知道是否有任何方法可以不依赖于自定义操作符。

Answer 1

我认为这可以通过内置的运营商来解决。下面的代码演示的方法之一，虽然用于热或非轻量级冷源时，事情可能很麻烦 - 我鼓励你使用它的教育/ GET-AN-想法的目的，而不是生产中使用

@Test 
fun slidingWindow() { 
    val events = Flowable.just(
      Data(5, 1.0), 
      Data(10, 1.5), 
      Data(30, 2.8), 
      Data(40, 3.2), 
      Data(60, 3.8), 
      Data(90, 4.2)) 
      .observeOn(Schedulers.io()) 
    val windows = window(windowSize = 3, slideSize = 1, data = events).toList().blockingGet() 
    Assert.assertNotNull(windows) 
    Assert.assertFalse(windows.isEmpty()) 
} 

fun window(windowSize: Int, slideSize: Int, data: Flowable<Data>): Flowable<List<Int>> = window(
     from = 0, 
     to = windowSize, 
     slideSize = slideSize, 
     data = data) 

fun window(from: Int, to: Int, slideSize: Int, data: Flowable<Data>): Flowable<List<Int>> { 
    val window = data.takeWhile { it.time <= to }.skipWhile { it.time < from }.map { it.data } 
    val tail = data.skipWhile { it.time <= from + slideSize } 
    val nonEmptyWindow = window.toList().filter { !it.isEmpty() } 
    val nextWindow = nonEmptyWindow.flatMapPublisher { 
     window(from + slideSize, to + slideSize, slideSize, tail).observeOn(Schedulers.io()) 
    } 
    return nonEmptyWindow.toFlowable().concatWith(nextWindow) 
} 

data class Data(val data: Int, val time: Double) 

以上测试收益
[[5, 10, 30], [10, 30, 40, 60], [30, 40, 60, 90], [40, 60, 90], [90]]