问题描述
我在下面创建了 Observable 构造函数,其工作原理如所述.有谁知道使用 RxJs 附带的运算符是否有更简洁的方法来实现相同的行为?我在看 bufferToggle 接近所需的行为,但我需要在缓冲区关闭时传递发出的值.
I created the Observable constructor below that works as described. Does anyone know if there is a more concise way of achieving the same behaviour using the operators that come with RxJs? I was looking at bufferToggle which is close to the required behaviour, but I need the emitted values to be passed through when the buffer is closed.
函数描述:如果condition
发出true
,则缓冲发出的source
值,并通过发出的source
值source
值如果 condition
发出 false
.如果条件在 true
后发出 false
,缓冲区会按照接收到的顺序释放每个值.缓冲区被初始化为传递发出的 source
值,直到 condition
发出 true
.
Function Description: Buffers the emitted source
values if the condition
emits true
, and passes through the emitted source
values if the condition
emits false
. If the condition emits false
after being true
, the buffer releases each value in the order that they were received. The buffer is initialised to pass through the emitted source
values until the condition
emits true
.
function bufferIf<T>(condition: Observable<boolean>, source: Observable<T>): Observable<T> {
return new Observable<T>(subscriber => {
const subscriptions: Subscription[] = [];
const buffer = [];
let isBufferOpen = false;
subscriptions.push(
// handle source events
source.subscribe(value => {
// if buffer is open, or closed but buffer is still being
// emptied from previously being closed.
if (isBufferOpen || (!isBufferOpen && buffer.length > 0)) {
buffer.push(value);
} else {
subscriber.next(value);
}
}),
// handle condition events
condition.do(value => isBufferOpen = value)
.filter(value => !value)
.subscribe(value => {
while (buffer.length > 0 && !isBufferOpen) {
subscriber.next(buffer.shift());
}
})
);
// on unsubscribe
return () => {
subscriptions.forEach(sub => sub.unsubscribe());
};
});
}
编辑
为了回应评论,以下与上面的功能相同,但采用 RxJs 运算符的形式,并更新为使用 RxJx 6+ pipeabale 运算符:
Edit
In response to comment, the following is the same function as the one above but in the form of an RxJs Operator and updated to use RxJx 6+ pipeabale Operators:
function bufferIf<T>(condition: Observable<boolean>): MonoTypeOperatorFunction<T> {
return (source: Observable<T>) => {
return new Observable<T>(subscriber => {
const subscriptions: Subscription[] = [];
const buffer: T[] = [];
let isBufferOpen = false;
subscriptions.push(
// handle source events
source.subscribe(value => {
// if buffer is open, or closed but buffer is still being
// emptied from previously being closed.
if (isBufferOpen || (!isBufferOpen && buffer.length > 0)) {
buffer.push(value);
} else {
subscriber.next(value);
}
}),
// handle condition events
condition.pipe(
tap(con => isBufferOpen = con),
filter(() => !isBufferOpen)
).subscribe(() => {
while (buffer.length > 0 && !isBufferOpen) {
subscriber.next(buffer.shift());
}
})
);
// on unsubscribe
return () => subscriptions.forEach(sub => sub.unsubscribe());
});
}
}
推荐答案
我找到了一种基于运算符而不是订阅的解决方案,但不愿称其为更简洁.
I've found a solution based on operators rather than subscriptions, but hesitate to call it more concise.
请注意,如果可以保证缓冲区开/关流始终以关闭(即奇数次发射)结束,则可以删除 endToken.
Note, the endToken can be removed if it's possible to guarantee the buffer on/off stream always ends with an off (i.e an odd number of emits).
console.clear()
const Observable = Rx.Observable
// Source and buffering observables
const source$ = Observable.timer(0, 200).take(15)
const bufferIt$ = Observable.timer(0, 500).map(x => x % 2 !== 0).take(6)
// Function to switch buffering
const endToken = 'end'
const bufferScanner = { buffering: false, value: null, buffer: [] }
const bufferSwitch = (scanner, [src, buffering]) => {
const onBufferClose = (scanner.buffering && !buffering) || (src === endToken)
const buffer = (buffering || onBufferClose) ? scanner.buffer.concat(src) : []
const value = onBufferClose ? buffer : buffering ? null : [src]
return { buffering, value, buffer }
}
// Operator chain
const output =
source$
.concat(Observable.of(endToken)) // signal last buffer to emit
.withLatestFrom(bufferIt$) // add buffering flag to stream
.scan(bufferSwitch, bufferScanner) // turn buffering on and off
.map(x => x.value) // deconsruct bufferScanner
.filter(x => x) // ignore null values
.mergeAll() // deconstruct buffer array
.filter(x => x !== endToken) // ignore endToken
// Proof
const start = new Date()
const outputDisplay = output.timestamp()
.map(x => 'value: ' + x.value + ', elapsed: ' + (x.timestamp - start) )
const bufferDisplay = bufferIt$.timestamp()
.map(x => (x.value ? 'buffer on' : 'buffer off') + ', elapsed: ' + (x.timestamp - start) )
bufferDisplay.merge(outputDisplay)
.subscribe(console.log)
<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/5.5.2/Rx.js"></script>
脚注
我也找到了一个基于 buffer()
的解决方案,但我不相信它在高频源下是稳定的.某些缓冲区配置似乎有些问题(即声明看起来很合理,但测试显示偶尔会出现干扰缓冲区操作的延迟).
I also found a solution based on buffer()
, but am not convinced it's stable with a high-frequency source. There seems to be something hokey with certain buffer configurations (i.e the declaration looks sound but tests show occasional delays which interfere with buffer operation).
无论如何,供参考,
/*
Alternate with buffered and unbuffered streams
*/
const buffered =
source$.withLatestFrom(bufferIt$)
.filter(([x, bufferIsOn]) => bufferIsOn)
.map(x => x[0])
.buffer(bufferIt$.filter(x => !x))
.filter(x => x.length) // filter out empty buffers
.mergeAll() // unwind the buffer
const unbuffered =
source$.withLatestFrom(bufferIt$)
.filter(([x, bufferIsOn]) => !bufferIsOn)
.map(x => x[0])
const output = buffered.merge(unbuffered)
这篇关于RxJs:条件为真时缓冲事件,条件为假时传递事件的文章就介绍到这了,希望我们推荐的答案对大家有所帮助,也希望大家多多支持!