关于 RX,我有一个小问题。我有一个从键盘输入的符号流,我需要将它们分成组。当流中出现 ';' 符号时,应该开始新的一组。简单来说,我需要一个操作符,类似于 Buffer,但它会在某个条件为真时触发,而不是经过一段时间延迟或事件计数后触发。有没有办法使用RX中已经存在的操作符构建它,或者我应该自己编写?
关于 RX,我有一个小问题。我有一个从键盘输入的符号流,我需要将它们分成组。当流中出现 ';' 符号时,应该开始新的一组。简单来说,我需要一个操作符,类似于 Buffer,但它会在某个条件为真时触发,而不是经过一段时间延迟或事件计数后触发。有没有办法使用RX中已经存在的操作符构建它,或者我应该自己编写?
这是一个源码。
var source = new[] { 'a', 'b', ';', 'c', 'd', 'e', ';' }.ToObservable();
以下是您所需的信息:
var groups = source
// Group the elements by some constant (0)
// and end the group when we see a semicolon
.GroupByUntil(x => 0, group => group.Where(x => x == ';'))
以下是使用方法:
groups
// Log that we're on the next group now.
.Do(x => Console.WriteLine("Group: "))
// Merge / Concat all the groups together
// {{a..b..;}..{c..d..e..;}} => {a..b..;..c..d..e..;}
.Merge()
// Ignore the semicolons? This is optional, I suppose.
.Where(x => x != ';')
// Log the characters!
.Do(x => Console.WriteLine(" {0}", x))
// Make it so, Number One!
.Subscribe();
输出:
Group:
a
b
Group:
c
d
e
//this is our incoming stream
IObservable<char> keyboardStream;
//if the observable is cold we need to do this
//in case of it being hot (as I would expect a keyboard stream to be) we do not need to do it
var persistedStream = keyboardStream.Publish().RefCount();
var bufferedBySemicolon = persistedStream.Buffer(persistedStream .Where(c=>c==';'));
keyboardStream
在不同于bufferedBySemicolon
订阅的线程上被观察/生成,则在此处使用RefCount是不确定性的,因此会出现竞争条件。 - cwharrisvar groups = Observable.Create(o => {
var publishedSource = source.Publish();
return new CompositeDisposable(
publishedSource.Buffer(publishedSource.Where(c => c == ';')).Subscribe(o),
publishedSource.Connect()
);
});