TCL模式匹配(TCL pattern matching)

 这对我有用。 
 
 首先，将byte []转换为字符串并将字符串拆分为\r （Regex Split保留分隔符）。 
 
 现在有一串字符串，其中一些以\r结尾。 
 
 然后康卡特，以保持他们的秩序。 另外，由于strings需要在下一步中变为热点，请发布它们。 
 
var strings = bytes.
  Select(arr => (Regex.Split(Encoding.Default.GetString(arr, 0, arr.Length - 1), "(\r)")).
    Where(s=> s.Length != 0).
    ToObservable()).
  Concat().
  Publish().
  RefCount();
 
 创建一个字符串窗口，当字符串以\r结尾时结束。 strings需要很热，因为它用于Window内容和窗口结束触发器。 
 
var linewindows = strings.Window(strings.Where(s => s.EndsWith("\r")));
 
 将每个Window聚合为一个字符串。 
 
var lines = linewindows.SelectMany(w => w.Aggregate((l, r) => l + r));
 
 lines是IObservable<String> ，每个字符串包含一行。 
 
 为了测试这个，我使用以下生成器来生成IObservable<byte[]> 
 
var bytes = Observable.
Range(1, 10).
SelectMany(i => Observable.
    Return((byte)('A' + i)).
    Repeat(24).
    Concat(Observable.
        Return((byte)'\r'))).
Window(17).
SelectMany(w => w.ToArray());

This works for me.
 
First, Convert the byte[] into a string and Split the string on \r (Regex Split keeps the delimiters). 
 
Now there is a stream of Strings some of which end in \r. 
 
Then Concat, to keep them in order. Also, since strings needs to be Hot for the next step, Publish them. 
 
var strings = bytes.
  Select(arr => (Regex.Split(Encoding.Default.GetString(arr, 0, arr.Length - 1), "(\r)")).
    Where(s=> s.Length != 0).
    ToObservable()).
  Concat().
  Publish().
  RefCount();
 
Make a Window of strings that ends when a string ends with \r. strings needs to be hot since it is used for both the Window contents and the end-of-window trigger.
 
var linewindows = strings.Window(strings.Where(s => s.EndsWith("\r")));
 
Aggregate each Window into a single string.
 
var lines = linewindows.SelectMany(w => w.Aggregate((l, r) => l + r));
 
lines is IObservable<String> and each string contains one line.
 
To test this I used the following generator to produce IObservable<byte[]>
 
var bytes = Observable.
Range(1, 10).
SelectMany(i => Observable.
    Return((byte)('A' + i)).
    Repeat(24).
    Concat(Observable.
        Return((byte)'\r'))).
Window(17).
SelectMany(w => w.ToArray());