5

私はprotobuf-netとWCFで遊んでいます。作成したコードは次のとおりです。

public class MobileServiceV2
{
    [WebGet(UriTemplate = "/some-data")]
    [Description("returns test data")]
    public Stream GetSomeData()
    {
        WebOperationContext.Current.OutgoingResponse.ContentType = "application/x-protobuf";

        var ms = new MemoryStream();
        ProtoBuf.Serializer.Serialize(ms, new MyResponse { SomeData = "Test data here" });
        return ms;
    }
}

[DataContract]
public class MyResponse
{
    [DataMember(Order = 1)] 
    public string SomeData { get; set; }
}

Fiddler を見ると、適切な送信コンテンツ タイプが表示され、すべて問題ないように見えますが、空の応答が返されます。IE はファイルをダウンロードするように要求しますが、このファイルは空です。シリアライザーが機能していませんか? それとも私はそれを正しくしないのですか？

編集：

メソッドに次のコードを追加しました。はい、正しくシリアル化されます。WCF からストリームを返す方法に問題があります。

using (var file = File.Create("C:\\test.bin"))
        {
            Serializer.Serialize(file, new MyResponse { SomeData = "Test data here" });
        }
4

3 に答える 3

10

MemoryStream に書き込み、巻き戻すだけです。この場合は Dispose()しないでください。

var ms = new MemoryStream();
Serializer.Serialize(ms, obj);
ms.Position = 0;
return ms;

ただし、これはメモリにバッファリングすることを意味します。それを避けるためにブードゥー教を考え出すこともできますが、それは非常に複雑です。

于 2012-07-20T16:42:20.447 に答える
3

質問を正しく理解できれば、ストリーミング WCF バインディングを使用しようとしています。その場合、データをチャンクに分割して個別にシリアライズし、クライアントで同じ方法でデシリアライズしてみてください。唯一の注意点は、受信側の WCF 提供の Stream 実装です。それをラップして、自分で読み取りを管理する必要があります。以下は、これを容易にするために使用するクラスです。

    public static class StreamingUtility
{

    public static IEnumerable<T> FromStream<T>(this Stream value, Action<T> perItemCallback = null)
    {

        List<T> result = new List<T>();
        StreamProxy sp = new StreamProxy(value);
        try
        {
            while (sp.CanRead)
            {

                T v = ProtoBuf.Serializer.DeserializeWithLengthPrefix<T>((Stream)sp, ProtoBuf.PrefixStyle.Base128);
                if (perItemCallback != null)
                    perItemCallback(v);

                result.Add(v);
            }
        }
        catch { }

        return result;
    }

    public static StreamingContent<T> SingleToStream<T>(this T value)
    {
        return new StreamingContent<T>(new T[] { value });
    }

    public static StreamingContent<T> ToStream<T>(this IEnumerable<T> value)
    {
        return new StreamingContent<T>(value);
    }

    public class StreamingContent<T> : Stream
    {
        private bool _canRead = true;
        private ManualResetEventSlim _dataIsReady = new ManualResetEventSlim(false);
        private bool _noMoreAdditions = false;
        private long _readingOffset = 0;

        //private IFormatter _serializer = new BinaryFormatter(null, new StreamingContext(StreamingContextStates.CrossMachine));
        private IEnumerable<T> _source = null;

        private MemoryStream _stream = new MemoryStream();

        public static StreamingContent<T> Clone(Stream origin)
        {
            return new StreamingContent<T>(origin);
        }

        private StreamingContent(Stream origin)
        {
            byte[] b = new byte[65536];

            while (true)
            {
                int count = origin.Read(b, 0, b.Length);

                if (count > 0)
                {
                    _stream.Write(b, 0, count);
                }
                else
                    break;
            }
            _noMoreAdditions = true;
        }

        public StreamingContent(IEnumerable<T> source)
        {
            if (!s_initialized)
            {
                StreamingUtility.Initialize();

                StreamingUtility.s_initialized = true;
            }

            _source = source.ToList();
            if (source.Count() > 0)
            {
                new Thread(new ParameterizedThreadStart(obj =>
                {
                    StreamingContent<T> _this = obj as StreamingContent<T>;
                    foreach (T item in _this._source)
                    {
                        lock (_this._stream)
                        {
                            if (_this._noMoreAdditions) break;
                            _stream.Seek(0, SeekOrigin.End);

                            ProtoBuf.Serializer.SerializeWithLengthPrefix<T>(_this._stream, item, ProtoBuf.PrefixStyle.Base128);

                            //_serializer.Serialize(_this._stream, item);
                            _dataIsReady.Set();
                        }
                    }

                    lock (_this._stream)
                    {
                        _this._noMoreAdditions = true;
                        _dataIsReady.Set();
                    }
                })) { IsBackground = true }.Start(this);
            }
            else
            {
                _canRead = false;
            }
        }

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

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

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

        public override long Length
        {
            get
            {
                while (!_noMoreAdditions) Thread.Sleep(20);
                return _stream.Length;
            }
        }

        public override long Position
        {
            get
            {
                throw new Exception("This stream does not support getting the Position property.");
            }
            set
            {
                throw new Exception("This stream does not support setting the Position property.");
            }
        }

        public override void Close()
        {
            lock (_stream)
            {
                _noMoreAdditions = true;
                _stream.Close();
            }
        }

        public override void Flush()
        {
        }

        public override int Read(byte[] buffer, int offset, int count)
        {
            if (!CanRead) return 0;

            bool wait = false;

            lock (_stream)
            {
                wait = !_dataIsReady.IsSet && !_noMoreAdditions;
            }

            if (wait)
            {
                _dataIsReady.Wait();
            }

            lock (_stream)
            {
                if (!_noMoreAdditions)
                    _dataIsReady.Reset();

                if (_stream.Length > _readingOffset)
                {
                    _stream.Seek(_readingOffset, SeekOrigin.Begin);
                    int res = _stream.Read(buffer, 0, count);

                    if (_noMoreAdditions && count + _readingOffset >= _stream.Length)
                        _canRead = false;

                    _readingOffset += res;

                    return res;
                }
            }

            return 0;
        }

        public override long Seek(long offset, SeekOrigin origin)
        {
            throw new Exception("This stream does not support seeking.");
        }

        public override void SetLength(long value)
        {
            throw new Exception("This stream does not support setting the Length.");
        }

        public override void Write(byte[] buffer, int offset, int count)
        {
            throw new Exception("This stream does not support writing.");
        }

        protected override void Dispose(bool disposing)
        {
            try
            {
                lock (_stream)
                {
                    _noMoreAdditions = true;
                    _stream.Close();
                }
            }
            catch { }
        }
    }

    private class StreamProxy : Stream
    {
        private bool _canRead = true;
        private bool _endOfMessage = false;
        private Stream _internalStream;
        private int _readPosition = 0;
        private MemoryStream _storage = new MemoryStream();
        private int _writePosition = 0;

        public StreamProxy(Stream internalStream)
        {
            _internalStream = internalStream;
            byte[] initialRequest = new byte[1000];

            int length = _internalStream.Read(initialRequest, 0, 1000);

            if (length != 0)
                _storage.Write(initialRequest, 0, length);
            else
                _canRead = false;

            _writePosition = length;
        }

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

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

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

        public override long Length
        {
            get { throw new NotImplementedException(); }
        }

        public override long Position
        {
            get
            {
                return _readPosition;
            }
            set
            {
                throw new NotImplementedException();
            }
        }

        public override void Flush()
        {
        }

        public override int ReadByte()
        {
            byte[] res = new byte[1];
            int g = Read(res, 0, 1);
            return res[0];
        }

        public override int Read(byte[] buffer, int offset, int count)
        {


            int res = 0;
            if (_readPosition + count > _writePosition)
            {
                /// add extra bytes to see if more data is available and we need to allow next read
                int readSize = _readPosition + count - _writePosition;

                if (readSize < 1024)
                    readSize = 1024;

                byte[] read = new byte[readSize];
                res = _internalStream.Read(read, 0, readSize);
                if (res > 0)
                {
                    _storage.Seek(_writePosition, SeekOrigin.Begin);
                    _writePosition += res;
                    _storage.Write(read, 0, res);

                }
                else if (res == 0)/// If the read returned 0, we're at the end
                {
                    _endOfMessage = true;

                }

                if (res > 0 && res < readSize)
                {
                    read = new byte[1024];
                    res = _internalStream.Read(read, 0, 1024);
                    if (res > 0)
                    {
                        _storage.Seek(_writePosition, SeekOrigin.Begin);
                        _writePosition += res;
                        _storage.Write(read, 0, res);

                    }
                    else if (res == 0)/// If the read returned 0, we're at the end
                    {
                        _endOfMessage = true;

                    }
                }
            }

            _storage.Seek(_readPosition, SeekOrigin.Begin);
            res = _storage.Read(buffer, offset, count);
            _readPosition += res;


            /// If end of message was reached and all the data was read from the
            /// internal storage, mark CanRead as false
            if (_readPosition >= _writePosition && _endOfMessage)
                _canRead = false;

            return res;
        }

        public override long Seek(long offset, SeekOrigin origin)
        {
            throw new NotImplementedException();
        }

        public override void SetLength(long value)
        {
            throw new NotImplementedException();
        }

        public override void Write(byte[] buffer, int offset, int count)
        {
            throw new NotImplementedException();
        }
    }
}

これを使用するには、次のように WCF インターフェイス メソッドを呼び出すだけです。

IEnumerable<SomeType> collection = ...
clannel.Method(collection.ToStream());

そして、受信側で次のように読み取ります。

public void Method(Stream stream){
   IEnumerable<SomeType> coll = stream.FromStream<SomeType>();
}

この実装はまだテスト中なので、ご意見をお待ちしております。

于 2013-08-16T16:04:17.113 に答える
0

この方法を試してください

var ms = new MemoryStream();
using (var file = File.Create("C:\\test.bin"))
{
Serializer.Serialize(file, new MyResponse { SomeData = "Test data here" });
file.CopyTo(ms); 
}

return ms;//stream
于 2012-07-20T15:43:46.050 に答える