Lazy, secuencia de objetos de transmisión de serialización con protobuf-net

Question

Estamos desarrollando un servicio WCF para transmitir una gran cantidad de datos, por lo tanto, hemos elegido utilizar la funcionalidad WCF Streaming combinada con una serialización protobuf-net.

Contexto:

general una idea es serializar objetos en el servicio, escribirlos en una corriente y enviar. En el otro extremo, la persona que llama recibirá un objeto Stream y podrá leer todos los datos.

Así Actualmente el código método de servicio se ve algo como esto:

public Result TestMethod(Parameter parameter) 
{ 
    // Create response 
    var responseObject = new BusinessResponse { Value = "some very large data"}; 

    // The resposne have to be serialized in advance to intermediate MemoryStream 
    var stream = new MemoryStream(); 
    serializer.Serialize(stream, responseObject); 
    stream.Position = 0; 

    // ResultBody is a stream, Result is a MessageContract 
    return new Result {ResultBody = stream}; 
} 

El objeto BusinessResponse es serializado a un MemoryStream y que se volvió de un método. En el lado del cliente el código de llamada se ve así:

var parameter = new Parameter(); 

// Call the service method 
var methodResult = channel.TestMethod(parameter); 

// protobuf-net deserializer reads from a stream received from a service. 
// while reading is performed by protobuf-net, 
// on the service side WCF is actually reading from a 
// memory stream where serialized message is stored 
var result = serializer.Deserialize<BusinessResponse>(methodResult.ResultBody); 
return result; 

Así que cuando serializer.Deserialize() se llama lo hace desde una corriente methodResult.ResultBody, en el mismo tiempo en el WCF lado de servicio está leyendo un MemoryStream, que se ha vuelto de a TestMethod.

Problema:

Lo que nos gustaría lograr es la de deshacerse de un MemoryStream e inicial serialización de todo el objeto en el lado del servicio a la vez. Como utilizamos la transmisión, nos gustaría evitar tener un objeto serializado en la memoria antes de enviarlo.

Idea:

La solución perfecta sería devolver un objeto corriente a medida vacío (de TestMethod()) con una referencia a un objeto que va a ser serializado (objeto 'BusinessResponse' en mi ejemplo) Entonces, cuando WCF llama a un método Read() de mi transmisión, internamente serializo una parte de un objeto usando protobuf-net y lo devuelvo a la persona que llama sin almacenarlo en la memoria.

Y ahora hay un problema, porque lo que realmente necesitamos es una posibilidad de serializar un objeto pieza por pieza en el momento en que se lee la secuencia. Entiendo que esta es una forma de serialización totalmente diferente: en lugar de enviar un objeto a un serializador, me gustaría solicitar un contenido serializado pieza por pieza.

¿Es ese tipo de serialización de alguna manera posible usando protobuf-net?

Answer 1

He cocinado algún código que probablemente esté en la línea de la idea de la puerta de Marc.

public class PullStream : Stream 
{ 
    private byte[] internalBuffer; 
    private bool ended; 
    private static ManualResetEvent dataAvailable = new ManualResetEvent(false); 
    private static ManualResetEvent dataEmpty = new ManualResetEvent(true); 

    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() 
    { 
     throw new NotImplementedException(); 
    } 

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

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

    public override int Read(byte[] buffer, int offset, int count) 
    { 
     dataAvailable.WaitOne(); 
     if (count >= internalBuffer.Length) 
     { 
      var retVal = internalBuffer.Length; 
      Array.Copy(internalBuffer, buffer, retVal); 
      internalBuffer = null; 
      dataAvailable.Reset(); 
      dataEmpty.Set(); 
      return retVal; 
     } 
     else 
     { 
      Array.Copy(internalBuffer, buffer, count); 
      internalBuffer = internalBuffer.Skip(count).ToArray(); // i know 
      return count; 
     } 
    } 

    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) 
    { 
     dataEmpty.WaitOne(); 
     dataEmpty.Reset(); 

     internalBuffer = new byte[count]; 
     Array.Copy(buffer, internalBuffer, count); 

     Debug.WriteLine("Writing some data"); 

     dataAvailable.Set(); 
    } 

    public void End() 
    { 
     dataEmpty.WaitOne(); 
     dataEmpty.Reset(); 

     internalBuffer = new byte[0]; 

     Debug.WriteLine("Ending writes"); 

     dataAvailable.Set(); 
    } 
} 

Esta es una clase de descendiente de flujo simple que solo implementa lectura y escritura (y final). Los bloques de lectura mientras no hay datos disponibles y los bloques de escritura mientras los datos están disponibles. De esta forma, solo hay un buffer de bytes involucrado. La copia de linq del resto está abierta para la optimización ;-) El método End se agrega para que no ocurra ningún bloqueo donde Read se realiza cuando no hay datos disponibles y ya no se escribirán más datos.

Tienes que escribir en esta corriente desde un hilo separado.Muestro esto a continuación:

// create a large object 
    var obj = new List<ToSerialize>(); 
    for(int i = 0; i <= 1000; i ++) 
     obj.Add(new ToSerialize { Test = "This is my very loooong message" }); 
    // create my special stream to read from 
    var ms = new PullStream(); 
    new Thread(x => 
    { 
     ProtoBuf.Serializer.Serialize(ms, obj); 
     ms.End(); 
    }).Start(); 
    var buffer = new byte[100]; 
    // stream to write back to (just to show deserialization is working too) 
    var ws = new MemoryStream(); 
    int read; 
    while ((read = ms.Read(buffer, 0, 100)) != 0) 
    { 
     ws.Write(buffer, 0, read); 
     Debug.WriteLine("read some data"); 
    } 
    ws.Position = 0; 
    var back = ProtoBuf.Serializer.Deserialize<List<ToSerialize>>(ws); 

Espero que esto resuelva su problema :-) Fue divertido codificar esto de todos modos.

Saludos, Jacco