在返回WCF流的同时将流写入文件？

Question

我有一个WCF服务，可以从远程FTP服务器流传输大文件（100MB +）。

[ServiceContract]
public interface IStreamService
{
    [OperationContract]
    Stream GetDataFromFtp();
}

public class StreamService : IStreamService
{
    public Stream GetDataFromFtp()
    {
        Stream ftpStream = Retr(...);
        return ftpStream;
    }
}

作为WCF将文件流传输到客户端时，我希望将其流式传输到本地缓存中，以便未来的请求不需要返回远程FTP - 我只需从磁盘上提供它们即可。

但是，在不将整个100MB文件流缓冲到内存中之前，我无法实现这一点。

我尝试使用一个简单的捕获流包装器，每次读取都进行写入：

public class CapturingStreamWrapper : Stream
{
    private readonly Stream stream;
    private readonly Stream captureStream;

    public override int Read(byte[] buffer, int offset, int count)
    {
        int readBytes = stream.Read(buffer, offset, count);
        captureStream.Write(buffer, offset, readBytes);

        return readBytes;
    }
}

public class StreamService : IStreamService
{
    public Stream GetDataFromFtp()
    {
        Stream ftpStream = Retr(...);
        Stream cacheStream = File.OpenWrite(...);
        return new CapturingStreamWrapper(ftpStream, cacheStream);
    }
}

但是这似乎行不通。

此外，这并没有提供任何错误处理 - 我需要一个catch块来删除任何半写文件，如果客户端传输失败（即事务性缓存）。我不确定这将如何工作，因为我不知道在WCF生命周期中何时调用/清理流。

有什么想法可以在向客户端流回的同时将数据流到文件中吗？

- Richard Dingwall

Answer 1

我最终编写了几个相互连接的流类——一个在读取时将数据传输到另一个。抱歉，代码粘贴有点长：

/// <summary>
/// A stream that, as it reads, makes those bytes available on an ouput
/// stream. Thread safe.
/// </summary>
public class CacheStream : Stream
{
    private readonly Stream stream;

    public CacheStream(Stream stream)
    {
        if (stream == null) throw new ArgumentNullException("stream");
        this.stream = stream;
        OutputStream = new CacheOutputStream(this);
    }

    public event EventHandler<BytesReadEventArgs> BytesRead = delegate { };
    public event EventHandler Closing = delegate { };

    public Stream OutputStream { get; private set; }

    public override void Flush()
    {
        stream.Flush();
    }

    public override long Seek(long offset, SeekOrigin origin)
    {
        throw new InvalidOperationException("Cannot seek in CachingStream.");
    }

    public override void SetLength(long value)
    {
        stream.SetLength(value);
    }

    public override int Read(byte[] buffer, int offset, int count)
    {
        int numberOfBytesRead = stream.Read(buffer, offset, count);

        if (numberOfBytesRead > 0)
            PipeToOutputStream(buffer, offset, numberOfBytesRead);

        return numberOfBytesRead;
    }

    private void PipeToOutputStream(byte[] buffer, int offset, int numberOfBytesRead)
    {
        var tmp = new byte[numberOfBytesRead];
        Array.Copy(buffer, offset, tmp, 0, numberOfBytesRead);
        BytesRead(this, new BytesReadEventArgs(tmp));
    }

    public override void Write(byte[] buffer, int offset, int count)
    {
        throw new InvalidOperationException("Cannot write in CachingStream.");
    }

    public override bool CanRead
    {
        get { return stream.CanRead; }
    }

    public override bool CanSeek
    {
        get { return false; }
    }

    public override bool CanWrite
    {
        get { return false; }
    }

    public override long Length
    {
        get { return stream.Length; }
    }

    public override long Position
    {
        get { return stream.Position; }
        set { throw new InvalidOperationException("Cannot set position in CachingStream."); }
    }

    public override void Close()
    {
        Closing(this, EventArgs.Empty);
        base.Close();
    }

    protected override void Dispose(bool disposing)
    {
        base.Dispose(disposing);
        OutputStream.Dispose();
    }
}

并且

/// <summary>
/// Output portion of CacheStream. Streams bytes from a queue of buffers.
/// Thread safe.
/// </summary>
public class CacheOutputStream : Stream
{
    private volatile int position;
    private volatile int length;
    private volatile bool sourceIsClosed;

    // No Deque<T> in the BCL yet, but LinkedList is more or less the same.
    private readonly LinkedList<byte[]> buffers = new LinkedList<byte[]>();

    public CacheOutputStream(CacheStream stream)
    {
        if (stream == null) throw new ArgumentNullException("stream");

        stream.BytesRead += (o, e) => AddToQueue(e.Buffer);
        stream.Closing += (o, e) => sourceIsClosed = true;
    }

    private void AddToQueue(byte[] buffer)
    {
        if (buffer.Length == 0)
            return;

        lock (buffers)
        {
            buffers.AddLast(buffer);
            length += buffer.Length;
        }
    }

    public override int Read(byte[] buffer, int offset, int count)
    {
        if (buffer == null) throw new ArgumentNullException("buffer");

        bool noMoreBuffersAvailable = HasNoMoreBuffersAvailable();

        // Guard clause - closed and nothing more to write.
        if (noMoreBuffersAvailable && sourceIsClosed)
            return 0;

        if (noMoreBuffersAvailable)
        {
            // Not closed yet! Block infinitely until we get closed or have some data.
            while (HasNoMoreBuffersAvailable())
            {
                if (sourceIsClosed)
                    return 0;

                Thread.Sleep(TimeSpan.FromMilliseconds(50));
            }
        }

        byte[] currentBuffer = GetCurrentBuffer();
        int numberOfBytesRead = DoRead(buffer, count, currentBuffer, offset);

        PutLeftoverBytesAtFrontOfQueue(currentBuffer, numberOfBytesRead);

        return numberOfBytesRead;
    }

    // Check if caller didn't have enough space to fit the buffer.
    // Put the remaining bytes at the front of the queue.
    private void PutLeftoverBytesAtFrontOfQueue(byte[] currentBuffer, int numberOfBytesRead)
    {
        if (currentBuffer == null) throw new ArgumentNullException("currentBuffer");

        if (numberOfBytesRead == currentBuffer.Length)
            return; // Clean read!

        var remainingBuffer = new byte[currentBuffer.Length - numberOfBytesRead];
        Array.Copy(currentBuffer, numberOfBytesRead, remainingBuffer, 0, remainingBuffer.Length);

        lock (buffers)
            buffers.AddFirst(remainingBuffer);
    }

    private int DoRead(byte[] buffer, int count, byte[] currentBuffer, int offset)
    {
        int maxNumberOfBytesWeCanWrite = Math.Min(count, currentBuffer.Length);

        Array.Copy(currentBuffer, 0, buffer, offset, maxNumberOfBytesWeCanWrite);
        position += maxNumberOfBytesWeCanWrite;

        return maxNumberOfBytesWeCanWrite;
    }

    private byte[] GetCurrentBuffer()
    {
        byte[] currentBuffer;

        lock (buffers)
        {
            currentBuffer = buffers.First.Value;
            buffers.RemoveFirst();
        }

        return currentBuffer;
    }

    private bool HasNoMoreBuffersAvailable()
    {
        lock (buffers)
            return buffers.Count == 0;
    }

    public override void Flush() { }

    public override long Seek(long offset, SeekOrigin origin)
    {
        throw new InvalidOperationException("Cannot seek in CachingStream.");
    }

    public override void SetLength(long value)
    {
        throw new InvalidOperationException("Cannot set length in CachingStream.");
    }

    public override void Write(byte[] buffer, int offset, int count)
    {
        throw new InvalidOperationException("Cannot write in a CachingStream.");
    }

    public override bool CanRead
    {
        get { return true; }
    }

    public override bool CanSeek
    {
        get { return false; }
    }

    public override bool CanWrite
    {
        get { return false; }
    }

    public override long Length
    {
        get { return length; }
    }

    public override long Position
    {
        get { return position; }
        set { throw new InvalidOperationException("Cannot set position in CachingStream."); }
    }
}

- Richard Dingwall

---

我正准备试一试这个.. 你也有 BytesReadEventArgs 吗？你觉得这个怎么样？ - makerofthings7

嗨，Richard。我知道这可能已经很旧了，但是想知道你在WCF实现中是否使用了流传输模式。 - tinonetic