Serializzazione di oggetti pigri e basati su stream con protobuf-net

Question

Stiamo sviluppando un servizio WCF per lo streaming di una grande quantità di dati, pertanto abbiamo scelto di utilizzare la funzionalità WCF Streaming combinata con una serializzazione protobuf-net.

Contesto:

In generale l'idea è di serializzare gli oggetti nel servizio, scriverli in un flusso e inviare. All'altro capo il chiamante riceverà un oggetto Stream e può leggere tutti i dati.

Così attualmente il codice di metodo di servizio sembra un po 'come questo:

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}; 
} 

L'oggetto BusinessResponse è serializzato a un MemoryStream e che viene restituito da un metodo. Sul lato client il codice chiamante appare così:

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; 

Così, quando serializer.Deserialize() si chiama legge da un flusso methodResult.ResultBody, allo stesso tempo sul WCF lato servizio sta leggendo un MemoryStream, che è stato restituito da a TestMethod.

Problema:

Quello che vorremmo raggiungere è quello di sbarazzarsi di un MemoryStream e prima serializzazione di tutto l'oggetto sul lato di servizio in una sola volta. Dato che usiamo lo streaming vorremmo evitare di tenere un oggetto serializzato in memoria prima di inviarlo.

Idea:

La soluzione ideale sarebbe quella di restituire un, oggetto Stream misura vuoto (da TestMethod()) con un riferimento a un oggetto che deve essere serializzato (oggetto 'BusinessResponse' nell'esempio). Quindi, quando WCF chiama un metodo Read() del mio stream, internizzo in serie un pezzo di un oggetto usando protobuf-net e lo restituisco al chiamante senza memorizzarlo nella memoria.

E ora c'è un problema, perché ciò di cui abbiamo effettivamente bisogno è la possibilità di serializzare un oggetto pezzo per pezzo nel momento in cui viene letto lo stream. Capisco che questo è un modo totalmente diverso di serializzazione: invece di spingere un oggetto su un serializzatore, vorrei richiedere un contenuto serializzato pezzo per pezzo.

Questo tipo di serializzazione è in qualche modo possibile utilizzando protobuf-net?

Answer 1

Ho preparato un codice che è probabilmente sulla falsariga dell'idea del cancello di 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(); 
    } 
} 

Questa è una classe di flusso semplice che implementa solo lettura e scrittura (e fine). I blocchi di lettura mentre nessun dato è disponibile e la scrittura si blocca mentre i dati sono disponibili. In questo modo è coinvolto solo un buffer di byte. La copia di linq del resto è aperta per l'ottimizzazione ;-) Il metodo End viene aggiunto in modo che non si verifichi il blocco in cui viene eseguita la lettura quando non sono disponibili dati e non viene più scritto alcun dato.

È necessario scrivere su questo flusso da un thread separato.Lo mostro qui sotto:

// 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); 

Spero che questo risolva il problema :-) È stato divertente codificarlo comunque.

Cordiali saluti, Jacco