将 IObservable<byte[]>拆分为字符，然后拆分为行

Question

Rx很棒，但有时很难找到优雅的方式来做某事。这个想法很简单。我接收到带有字节[]的事件，此数组可能包含一行、多行或一行的一部分。我想要的是找到一种方法，使Line的IObservable为IObservable<String>，其中序列的每个元素都是一条线。

几个小时后，我发现最接近的解决方案非常丑陋，而且不正常，因为扫描在每个字符上触发OnNext：

//Intermediate subject use to transform byte[] into char
var outputStream = new Subject<char>();
_reactiveSubcription = outputStream
    //Scan doesn't work it trigger OnNext on every char
    //Aggregate doesn't work neither as it doesn't return intermediate result
    .Scan(new StringBuilder(), (builder, c) => c == 'r' ? new StringBuilder() : builder.Append((char)c))
    .Subscribe(this);

Observable.FromEventPattern<ShellDataEventArgs>(shell, "DataReceived")
            //Data is a byte[]
            .Select(_ => _.EventArgs.Data)
            .Subscribe(array => array.ToObservable()
            //Convert into char
            .ForEach(c => outputStream.OnNext((char)c)));

注：_reactiveSubcription应为IObservable<String>。

在不考虑字符编码问题的情况下，我缺少什么来完成这项工作？

Answer 1

这对我有用。

首先，将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需要是热的，因为它用于窗口内容和窗口结束触发器。

var linewindows = strings.Window(strings.Where(s => s.EndsWith("r")));

将每个窗口聚合为一个字符串。

var lines = linewindows.SelectMany(w => w.Aggregate((l, r) => l + r));

CCD_ 9是CCD_。

为了测试这一点，我使用以下生成器生成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());