我正在使用 Stream 的自定义实现这将流式传输 IEnumerable<T>成一个流。我正在使用这个 EnumerableStream实现以执行转换。

我正在使用它以流式传输模式通过 WCF 执行流式传输。我能够转换 IEnumerable到流没有问题。有一次，我在客户端，我可以反序列化并获取所有数据，但是我无法找到停止循环我的流的条件。我得到:

System.Runtime.Serialization.SerializationException: End of Stream encountered before parsing was completed.

这是我要实现的示例示例:

class Program
{
    public static void Main()
    {
        var ListToSend = new List<List<string>>();
        var ListToReceive = new List<List<string>>();
        ListToSend = SimulateData().ToList();
        using (Stream stream = GetStream(ListToSend))
        {
            var formatter = new BinaryFormatter();
            while (stream.CanRead || 1 == 1 || true...) // What should I put in here to stop once I read everything???
            {
                List<string> row = formatter.Deserialize(stream) as List<string>;
                ListToReceive.Add(row);
            }
            Printer(ListToReceive);
            Console.WriteLine("Done");
        }
    }

    private static void Printer(List<List<string>> data)
    {
        Console.WriteLine("Printing");
        foreach (var row in data)
        {
            foreach (var cell in row)
            {
                Console.Write(cell + "\t");
            }
            Console.WriteLine("-------------------------------------------------------------------------------");
        }
    }
    private static Stream GetStream(IEnumerable<List<string>> data)
    {
        return EnumerableStream.Create(data, DeserializerCallback);
    }

    private static List<byte> DeserializerCallback(object obj)
    {
        var binFormatter = new BinaryFormatter();
        var mStream = new MemoryStream();
        binFormatter.Serialize(mStream, obj);
        return mStream.ToArray().ToList();
    }

    private static IEnumerable<List<string>> SimulateData()
    {
        Random randomizer = new Random();
        for (var i = 0; i < 10; i++)
        {
            var row = new List<string>();
            for (var j = 0; j < 1000; j++)
            {
                row.Add((randomizer.Next(100)).ToString());
            }
            yield return row;
        }
    }
}

我没有包含自定义流。我创建了一个 fiddle对于那些想要查看完整代码的人。

• 我是否需要在自定义流本身中添加一些内容以通知已读取所有数据？
• 是不是因为反序列化器和序列化器的格式不一样(我不这么认为)。
• 我也想知道为什么当我在读取函数中放置一个断点时，缓冲区大小会随机变化。
• 我不想用 try and catch 包装代码，我想要一个不会崩溃的干净解决方案。

最佳答案

Do I need to add something in the custom stream itself to notify that all the data have been read?

可以，但是在接收到 Stream 的 WCF 场景中这无济于事。是一个不同的类。

有两种标准(官方的，设计的)方法来确定 Stream 的结束数据:

(1) ReadByte返回-1

Returns

The unsigned byte cast to an Int32, or -1 if at the end of the stream.

(2) Read用 count > 0 调用时返回 0

Returns

The total number of bytes read into the buffer. This can be less than the number of bytes requested if that many bytes are not currently available, or zero (0) if the end of the stream has been reached.

不幸的是，它们都消耗当前字节(前进到下一个字节)并将破坏反序列化器。

有哪些可能的解决方案？

首先，实现一些序列化/反序列化格式(协议(protocol))，让您知道是否有更多元素要反序列化。例如，List<T>商店 Count在元素之前，T[]商店 Length在元素等之前自EnumerableStream<T>事先不知道计数，一个简单的解决方案是在每个元素之前发出一个假字节:

private bool SerializeNext()
{
    if (!_source.MoveNext())
        return false;

    buf.Enqueue(1); // <--
    foreach (var b in _serializer(_source.Current))
        _buf.Enqueue(b);

    return true;
}

这将允许您使用

while (stream.ReadByte() != -1)
{
    // ...
}

其次，如果您想保留当前格式，更通用的解决方案是实现自定义流，它包装另一个流并实现 PeekByte与标准语义相同的方法 ReadByte ，但不消耗当前字节:

public class SequentialStream : Stream
{
    private Stream source;
    private bool leaveOpen;
    private int? nextByte;

    public SequentialStream(Stream source, bool leaveOpen = false)
    {
        if (source == null) throw new ArgumentNullException(nameof(source));
        if (!source.CanRead) throw new ArgumentException("Non readable source.", nameof(source));
        this.source = source;
        this.leaveOpen = leaveOpen;
    }

    protected override void Dispose(bool disposing)
    {
        if (disposing && !leaveOpen)
            source.Dispose();
        base.Dispose(disposing);
    }

    public override bool CanRead => true;
    public override bool CanSeek => false;
    public override bool CanWrite => false;
    public override long Length => throw new NotSupportedException();
    public override long Position { get => throw new NotSupportedException(); set => throw new NotSupportedException(); }
    public override void Flush() { }
    public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException();
    public override void SetLength(long value) => throw new NotSupportedException();
    public override void Write(byte[] buffer, int offset, int count) => throw new NotSupportedException();

    public int PeekByte()
    {
        if (nextByte == null)
            nextByte = source.ReadByte();
        return nextByte.Value;
    }

    public override int Read(byte[] buffer, int offset, int count)
    {
        if (count <= 0) return 0;
        if (nextByte != null)
        {
            if (nextByte.Value < 0) return 0;
            buffer[offset] = (byte)nextByte.Value;
            if (count > 1)
            {
                int read = source.Read(buffer, offset + 1, count - 1);
                if (read == 0)
                    nextByte = -1;
                else
                    nextByte = null;
                return read + 1;
            }
            else
            {
                nextByte = null;
                return 1;
            }
        }
        else
        {
            int read = source.Read(buffer, offset, count);
            if (read == 0)
                nextByte = -1;
            return read;
        }
    }
} 

这基本上实现了具有 0 或 1 字节预读功能的只读流。

用法是这样的:

using (var stream = new SequentialStream(GetStream(ListToSend)))
{
    // ...
    while (stream.PeekByte() != -1) 
    {
        // ...
    }
    // ...
}

附言怎么样

I also want to know why when I put a break point in the read function, the buffer size is changing randomly.

这不是随机的。 BinaryFormatter内部使用 BinaryReader读取键入的值，如 Int32 , Byte , String等，将所需的大小传递为 count ，例如4、1、字符串编码字节数(它知道是因为在实际数据之前将它们存储在流中并在尝试读取实际数据之前读取它)等。

关于c# - 使用自定义流 (IEnumerable<T>)，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/51993330/