在响应式扩展中实现 String Split() 运算符

Question

我希望能够解释传入的字符并“拆分”它们（在这种情况下，通过空格字符）。

var incomingCharacters = "This is a test".ToCharArray().ToObservable();

// Yields a sequence of words - "This", "is", "a", "test"
var incomingWords = incomingCharacters.Split(' ');

我做了一个操作员来做这个，但我想知道是否有更好的方法？

public static IObservable<string> Split(this IObservable<char> incomingCharacters, char s)
{
    var wordSeq = Observable.Create<string>(observer =>
        {
            // Create an inner sequence to look for word separators; publish each word to the
            // "outer sequence" as it is found
            var innerSeq = incomingCharacters
                .Concat(Observable.Return(s))           // Enables the last word to be processed
                .Scan(new StringBuilder(), (builder, c) =>
                    {
                        if (c != s)
                        {
                            builder.Append(c);
                            return builder;
                        }

                        // We encountered a "split" character; publish the current completed word
                        // and begin collecting a new one
                        observer.OnNext(builder.ToString());
                        return new StringBuilder();
                    });

            innerSeq.Subscribe(list => { });

            return Disposable.Empty;
    });

    // Return the outer sequence
    return wordSeq;
}

Answer 1

有一种更简单的方法可以使用Buffer：

public static IObservable<string> Split(
          this IObservable<char> incomingCharacters, 
          char sep)
{
    // Share a single subscription
    var oneSource = incomingCharacters.Publish().RefCount();

    // Our "stop buffering" trigger will be the separators
    var onlySeparators = oneSource
        .Where(c => c == sep);

    return oneSource
        // buffer until we get a separator
        .Buffer(onlySeparators)
        // then return a new string from the buffered chars
        .Select(carr => new string(carr.ToArray()));        
}

测试：

void Main()
{
    var feeder = new Subject<char>();   
    var query = feeder.Split(' ');

    using(query.Subscribe(Console.WriteLine))
    {
        foreach(var c in "this should split words on spaces ".ToCharArray())
        {
            feeder.OnNext(c);
        }
        Console.ReadLine();
    }       
}

输出：

this 
should 
split 
words 
on 
spaces 

编辑：一个基本的BufferUntil 实现

public static class Ext
{
    public static IObservable<IList<T>> BufferUntil<T>(
         this IObservable<T> source, 
         Func<T, bool> predicate)
    {
        var singleSource = source.Publish().RefCount();
        var trigger = singleSource.Where(predicate);
        return singleSource.Buffer(trigger);
    }

    public static IObservable<string> Split(
       this IObservable<char> incomingCharacters, 
       char sep)
    {
        return incomingCharacters
             .BufferUntil(c => c == sep)
             .Select(carr => new string(carr.ToArray()));
    }
}