1

我有一个类可以对大量数据进行一些长时间运行的处理，并将输出写入我提供的流中。我正在尝试将 WCF 放在前面（使用命名管道），但无法弄清楚如何返回流。到目前为止，我有这样的事情：

interface IProcessor { Stream GetStream(); }

class Host {
  static void Main(string[] args) {
    using (ServiceHost sh = new ServiceHost(typeof(Processor), new Uri[]{new Uri("net.pipe://localhost")})) {
      var binding = new NetNamedPipeBinding();
      binding.TransferMode = TransferMode.StreamedResponse;
      sh.AddServiceEndpoint(typeof(IProcessor), binding, "foo");
      sh.Open();
      Console.WriteLine("Waiting...");
      Console.ReadLine();
      sh.Close();
    }
  }
}

class Processor : IProcessor {
  Stream GetStream() {
    var SourceProcessor = new SourceProcessor(...);
    var s = new MemoryStream();
    new Task(() => { SourceProcessor.Run(s); }).Start();
    return s;
  }
}

class Client {
  static void Main(string[] args) {
    Console.WriteLine("Starting...");
    var binding = new NetNamedPipeBinding();
    binding.TransferMode = TransferMode.StreamedResponse;
    ChannelFactory<IProcessor> f = new ChannelFactory<IProcessor>(binding, new EndpointAddress("net.pipe://localhost/foo"));
    Console.WriteLine("Creating channel...");
    IProcessor eps = f.CreateChannel();
    Console.WriteLine("Getting stream.");
    Stream s = eps.GetStream();
    StreamReader sr = new StreamReader(s);
    while (!sr.EndOfStream) Console.WriteLine(sr.ReadLine());
    Console.ReadLine();
  }
}

一切都经历了这些动作，但当然没有任何源数据能够传递给客户端。我对如何做到这一点感到困惑（也许我做不到），因为我需要返回流并运行任务并可能等待任务完成。如果我只是调用 SourceProcessor.Run(s) 而不是在任务中，据称它会阻塞和缓冲，但我不确定如何让它等到任务完成，同时还返回流供客户端读取...

4

1 回答 1

4

Read(问题是，如果 WCF 调用并且调用返回 0 字节，WCF 会认为该流已“完成” 。MemoryStream很乐意这样做，如果没有可用数据，它不会阻止读取。

您的问题的根源是 WCF 的读取MemoryStream速度比您写入它的速度更快，并且认为它已“完成”，修复它的方法是您需要返回不同类型的Stream块而不是在没有时返回 0可用的数据。.NET 没有内置任何东西可以做到这一点，您需要找到第 3 方类或自己制作（它可能很简单，就像从 MemoryStream 派生并覆盖Read以阻止读取，直到设置“完成”标志（有关类似行为，请参阅BlockingCollection<T>及其方法））。CompleteAdding()

为了好玩，我把它放在一起，它完全未经测试，但它可能会满足你的需要。

using System;
using System.Collections.Concurrent;
using System.IO;

namespace Example
{
    public class BufferStream : Stream
    {
        public BufferStream()
        {
            _data = new BlockingCollection<byte[]>();
        }


        /// <param name="boundedCapacity">The maximum number of calls to <see cref="Write"/> that can be made without
        /// the buffer being drained.</param>
        public BufferStream(int boundedCapacity)
        {
            _data = new BlockingCollection<byte[]>(boundedCapacity);
        }

        private readonly BlockingCollection<byte[]> _data;
        private byte[] _currentBlock = null;
        private int _currentBlockIndex = 0;

        public int BoundedCapacity { get { return _data.BoundedCapacity; } }

        public int BufferedWrites { get { return _data.Count; } }

        public bool IsAddingCompleted
        {
            get { return _data.IsAddingCompleted; }
        }

        public bool IsCompleted
        {
            get { return _data.IsCompleted; }
        }

        public void CompleteAdding()
        {
            _data.CompleteAdding();
        }

        public override void Write(byte[] buffer, int offset, int count)
        {
            var localArray = new byte[count];

            //Copy the data in to a new buffer of exactly the count size.
            Array.Copy(buffer, offset, localArray, 0, count);

            _data.Add(localArray);
        }


        public override int Read(byte[] buffer, int offset, int count)
        {
            if (_currentBlock == null || _currentBlockIndex == _currentBlock.Length)
            {
                if (!GetNextBlock()) 
                    return 0;
            }

            int minCount = Math.Min(count, _currentBlock.Length - _currentBlockIndex);

            Array.Copy(_currentBlock, _currentBlockIndex, buffer, offset, minCount);
            _currentBlockIndex += minCount;

            return minCount;
        }

        /// <summary>
        /// Loads the next block in to <see cref="_currentBlock"/>.
        /// </summary>
        /// <returns>True if the next block was retrieved.</returns>
        private bool GetNextBlock()
        {
            if (!_data.TryTake(out _currentBlock))
            {
                //The TryTake failed, the collection is empty.

                //See if we are in the completed state.
                if (_data.IsCompleted)
                {
                    return false;
                }

                //Wait for more data to show up.
                try
                {
                    _currentBlock = _data.Take();
                }
                catch (InvalidOperationException)
                {
                    //If the blocking collection was marked complete while we where waiting Take throws a InvalidOperationException
                    return false;
                }
            }

            _currentBlockIndex = 0;
            return true;
        }

        #region Constant functions

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

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

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

        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 long Length
        {
            get { throw new NotSupportedException(); }
        }

        public override long Position
        {
            get { throw new NotSupportedException(); }
            set { throw new NotSupportedException(); }
        }

        #endregion
    }
}

这比派生的优点MemoryStream是一旦 WCF 读取了一个值，它就不再需要保留在内存中（返回 aStream而不是 a的全部意义byte[]）

于 2015-04-07T15:08:55.590 回答