Rx java - Как получить текущий и предыдущий пункт?

Question

Как мне настроить операторы set op, чтобы я всегда получал предыдущее и текущее значение? Если возможно, я хочу избежать создания состояния вне канала.

- time ->
1      2      3      4
|      |      |      |
Operations
       |      |      |
       (1,2)  (2,3)  (3,4)

Обратите внимание, что каждое значение, кроме первого и последнего, должно появляться дважды, поэтому простой буфер не подходит.

Я думал об объединении skip с merge и buffer, но слияние, похоже, не гарантирует упорядочивания.

val s = PublishSubject.create<Int>()
s.mergeWith(s.skip(1)).buffer(2).subscribe{i -> print(i)}
s.onNext(1)
s.onNext(2)
s.onNext(3)
s.onNext(4)

output:
[1, 2] [2 , 3] [3, 4]

val o = Observable.just(1,2,3,4)
o.mergeWith(o.skip(1)).buffer(2).subscribe{i -> print(i)}

вывод:
[1, 2] [3, 4] [2, 3] [4]

(подошва 4 в порядке и ожидалась)

Answer 1

Похоже, вы все еще можете использовать буфер:

Observable.just(1, 2, 3, 4)
    .buffer(2, 1)
    .subscribe { println(it) }

// prints
// [1, 2]
// [2, 3]
// [3, 4]
// [4]