c# - 将非终止流中的对象集合转换为 IObservable

Question

我有一个 Stream包含序列化为 JSON 的对象集合，例如:

{"Name":"Alice","Age":20}{"Name":"Bob","Age":30}{"Name":"Charlie","Age":35}{"Name":"Danielle","Age":50}...

这个流可能非常大，所以我想在项目进入时对其进行处理，而不是将整个流读入一个字符串。此外，如果流是网络流，它将无限期打开。

有些对象可能非常复杂，所以我想利用 JSON.NET 提供的 JSON 序列化。

有没有一种方法可以使用 Newtonsoft Json 将流转换为 IObservable<Person>并在元素到达时对其进行 react 性处理？

Answer 1

如果您可以修改您的输入，使其看起来像一个 JSON 数组，那么您可以直接使用 JsonTextReader 读取数组部分，然后使用 JsonSerializer 读取数组中的对象。

这样，可以流式传输结果，而您不必处理所有反序列化。

生成 IObservable 的方法可能类似于:

public static IObservable<T> DeserializeAsObservable<T>(this TextReader reader)
{
    var jsonReader = new JsonTextReader(reader);

    if (!jsonReader.Read() || jsonReader.TokenType != JsonToken.StartArray)
        throw new Exception();

    var serializer = JsonSerializer.Create(null);

    return Observable.Create<T>(o => 
        {
            while (true)
            {
                if (!jsonReader.Read())
                    o.OnError(new Exception());
                if (jsonReader.TokenType == JsonToken.EndArray)
                    break;
                var obj = serializer.Deserialize<T>(jsonReader);
                o.OnNext(obj);
            }
            o.OnCompleted();
            return Disposable.Empty;
        });
}

(显然只有在将 new Exception() 替换为更合适的异常类型后才能使用它。)