C#: ¿Implementando NetworkStream.Peek?

byte

Question

Actualmente, no hay un método NetworkStream.Peek en C #. ¿Cuál es la mejor manera de implementar un método de este tipo que funciona como NetworkStream.ReadByte excepto que el byte devuelto no se elimina realmente del Stream ?

Answer 1

Aquí hay una implementación muy simple de PeekStream que le permite ver un cierto número de bytes solo al inicio de la transmisión (en lugar de poder mirar en cualquier momento). Los bytes observados se devuelven como un Stream para minimizar los cambios en el código existente.

Así es como lo usas:

Stream nonSeekableStream = ...;
PeekStream peekStream = new PeekStream(nonSeekableStream, 30); // Peek max 30 bytes
Stream initialBytesStream = peekStream.GetInitialBytesStream();
ParseHeaders(initialBytesStream);  // Work on initial bytes of nonSeekableStream
peekStream.Read(...) // Read normally, the read will start from the beginning

GetInitialBytesStream() devuelve una secuencia de búsqueda que contiene hasta bytes iniciales peekSize de la secuencia subyacente (menos si la secuencia es más corta que peekSize ).

Debido a su simplicidad, la lectura de PeekStream debería ser solo un poco más lenta (si es que lo hace) que la lectura de la secuencia subyacente directamente.

public class PeekStream : Stream
{
    private Stream m_stream;
    private byte[] m_buffer;
    private int m_start;
    private int m_end;

    public PeekStream(Stream stream, int peekSize)
    {
        if (stream == null)
        {
            throw new ArgumentNullException("stream");
        }
        if (!stream.CanRead)
        {
            throw new ArgumentException("Stream is not readable.");
        }
        if (peekSize < 0)
        {
            throw new ArgumentOutOfRangeException("peekSize");
        }
        m_stream = stream;
        m_buffer = new byte[peekSize];
        m_end = stream.Read(m_buffer, 0, peekSize);
    }

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

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

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

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

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

    public MemoryStream GetInitialBytesStream()
    {
        return new MemoryStream(m_buffer, 0, m_end, false);
    }

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

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

    public override int Read(byte[] buffer, int offset, int count)
    {
        // Validate arguments
        if (buffer == null)
        {
            throw new ArgumentNullException("buffer");
        }
        if (offset < 0)
        {
            throw new ArgumentOutOfRangeException("offset");
        }
        if (offset + count > buffer.Length)
        {
            throw new ArgumentOutOfRangeException("count");
        }

        int totalRead = 0;

        // Read from buffer
        if (m_start < m_end)
        {
            int toRead = Math.Min(m_end - m_start, count);
            Array.Copy(m_buffer, m_start, buffer, offset, toRead);
            m_start += toRead;
            offset += toRead;
            count -= toRead;
            totalRead += toRead;
        }

        // Read from stream
        if (count > 0)
        {
            totalRead += m_stream.Read(buffer, offset, count);
        }

        // Return total bytes read
        return totalRead;
    }

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

    public override int ReadByte()
    {
        if (m_start < m_end)
        {
            return m_buffer[m_start++];
        }
        else
        {
            return m_stream.ReadByte();
        }
    }

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

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

Descargo de responsabilidad: PeekStream anterior se toma de un programa de trabajo, pero no está probado exhaustivamente, por lo que puede contener errores. A mí me funciona, pero es posible que descubras algunos casos de esquina donde falla.

Answer 2

Me encontré con el mismo ''buscar número mágico'' y luego decido a qué procesador de transmisión enviaría el requerimiento de transmisión ''y desafortunadamente no puedo salir de ese problema - como se sugiere en los comentarios a la respuesta de Aaronaught - pasando los bytes ya consumidos en los métodos de procesamiento de flujo en parámetros separados, ya que esos métodos son dados y esperan System.IO.Stream y nada más.

Resolví esto creando una clase PeekableStream más o menos universal que envuelve un Stream. Funciona para NetworkStreams, pero también para cualquier otro Stream, siempre que Stream.Lo puede leer .

Editar

Alternativamente, puede usar el nuevo ReadSeekableStream y hacer

var readSeekableStream = new ReadSeekableStream(networkStream, /* >= */ count);
...
readSeekableStream.Read(..., count);
readSeekableStream.Seek(-count, SeekOrigin.Current);

En cualquier caso, aquí viene PeekableStream :

/// <summary>
/// PeekableStream wraps a Stream and can be used to peek ahead in the underlying stream,
/// without consuming the bytes. In other words, doing Peek() will allow you to look ahead in the stream,
/// but it won''t affect the result of subsequent Read() calls.
/// 
/// This is sometimes necessary, e.g. for peeking at the magic number of a stream of bytes and decide which
/// stream processor to hand over the stream.
/// </summary>
public class PeekableStream : Stream
{
    private readonly Stream underlyingStream;
    private readonly byte[] lookAheadBuffer;

    private int lookAheadIndex;

    public PeekableStream(Stream underlyingStream, int maxPeekBytes)
    {
        this.underlyingStream = underlyingStream;
        lookAheadBuffer = new byte[maxPeekBytes];
    }

    protected override void Dispose(bool disposing)
    {
        if (disposing)
            underlyingStream.Dispose();

        base.Dispose(disposing);
    }

    /// <summary>
    /// Peeks at a maximum of count bytes, or less if the stream ends before that number of bytes can be read.
    /// 
    /// Calls to this method do not influence subsequent calls to Read() and Peek().
    /// 
    /// Please note that this method will always peek count bytes unless the end of the stream is reached before that - in contrast to the Read()
    /// method, which might read less than count bytes, even though the end of the stream has not been reached.
    /// </summary>
    /// <param name="buffer">An array of bytes. When this method returns, the buffer contains the specified byte array with the values between offset and
    /// (offset + number-of-peeked-bytes - 1) replaced by the bytes peeked from the current source.</param>
    /// <param name="offset">The zero-based byte offset in buffer at which to begin storing the data peeked from the current stream.</param>
    /// <param name="count">The maximum number of bytes to be peeked from the current stream.</param>
    /// <returns>The total number of bytes peeked into the buffer. If it is less than the number of bytes requested then the end of the stream has been reached.</returns>
    public virtual int Peek(byte[] buffer, int offset, int count)
    {
        if (count > lookAheadBuffer.Length)
            throw new ArgumentOutOfRangeException("count", "must be smaller than peekable size, which is " + lookAheadBuffer.Length);

        while (lookAheadIndex < count)
        {
            int bytesRead = underlyingStream.Read(lookAheadBuffer, lookAheadIndex, count - lookAheadIndex);

            if (bytesRead == 0) // end of stream reached
                break;

            lookAheadIndex += bytesRead;
        }

        int peeked = Math.Min(count, lookAheadIndex);
        Array.Copy(lookAheadBuffer, 0, buffer, offset, peeked);
        return peeked;
    }

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

    public override long Position
    {
        get
        {
            return underlyingStream.Position - lookAheadIndex;
        }
        set
        {
            underlyingStream.Position = value;
            lookAheadIndex = 0; // this needs to be done AFTER the call to underlyingStream.Position, as that might throw NotSupportedException, 
                                // in which case we don''t want to change the lookAhead status
        }
    }

    public override int Read(byte[] buffer, int offset, int count)
    {
        int bytesTakenFromLookAheadBuffer = 0;
        if (count > 0 && lookAheadIndex > 0)
        {
            bytesTakenFromLookAheadBuffer = Math.Min(count, lookAheadIndex);
            Array.Copy(lookAheadBuffer, 0, buffer, offset, bytesTakenFromLookAheadBuffer);
            count -= bytesTakenFromLookAheadBuffer;
            offset += bytesTakenFromLookAheadBuffer;
            lookAheadIndex -= bytesTakenFromLookAheadBuffer;
            if (lookAheadIndex > 0) // move remaining bytes in lookAheadBuffer to front
                // copying into same array should be fine, according to http://msdn.microsoft.com/en-us/library/z50k9bft(v=VS.90).aspx :
                // "If sourceArray and destinationArray overlap, this method behaves as if the original values of sourceArray were preserved
                // in a temporary location before destinationArray is overwritten."
                Array.Copy(lookAheadBuffer, lookAheadBuffer.Length - bytesTakenFromLookAheadBuffer + 1, lookAheadBuffer, 0, lookAheadIndex);
        }

        return count > 0
            ? bytesTakenFromLookAheadBuffer + underlyingStream.Read(buffer, offset, count)
            : bytesTakenFromLookAheadBuffer;
    }

    public override int ReadByte()
    {
        if (lookAheadIndex > 0)
        {
            lookAheadIndex--;
            byte firstByte = lookAheadBuffer[0];
            if (lookAheadIndex > 0) // move remaining bytes in lookAheadBuffer to front
                Array.Copy(lookAheadBuffer, 1, lookAheadBuffer, 0, lookAheadIndex);
            return firstByte;
        }
        else
        {
            return underlyingStream.ReadByte();
        }
    }

    public override long Seek(long offset, SeekOrigin origin)
    {
        long ret = underlyingStream.Seek(offset, origin);
        lookAheadIndex = 0; // this needs to be done AFTER the call to underlyingStream.Seek(), as that might throw NotSupportedException,
                            // in which case we don''t want to change the lookAhead status
        return ret;
    }

    // from here on, only simple delegations to underlyingStream

    public override bool CanSeek { get { return underlyingStream.CanSeek; } }
    public override bool CanWrite { get { return underlyingStream.CanWrite; } }
    public override bool CanTimeout { get { return underlyingStream.CanTimeout; } }
    public override int ReadTimeout { get { return underlyingStream.ReadTimeout; } set { underlyingStream.ReadTimeout = value; } }
    public override int WriteTimeout { get { return underlyingStream.WriteTimeout; } set { underlyingStream.WriteTimeout = value; } }
    public override void Flush() { underlyingStream.Flush(); }
    public override long Length { get { return underlyingStream.Length; } }
    public override void SetLength(long value) { underlyingStream.SetLength(value); }
    public override void Write(byte[] buffer, int offset, int count) { underlyingStream.Write(buffer, offset, count); }
    public override void WriteByte(byte value) { underlyingStream.WriteByte(value); }
}

Answer 3

Si no necesita recuperar realmente el byte, puede referirse a la propiedad DataAvailable .

De lo contrario, puede envolverlo con un StreamReader e invocar su método Peek .

Tenga en cuenta que ninguno de estos es particularmente confiable para leer desde un flujo de red, debido a problemas de latencia. Es posible que los datos estén disponibles (presentes en el búfer de lectura) en el mismo instante en que se eche un vistazo.

No estoy seguro de qué es lo que pretende hacer con esto, pero el método de Read en NetworkStream es una llamada de bloqueo, por lo que realmente no necesita verificar el estado, incluso si está recibiendo en trozos. Si está tratando de mantener la respuesta de la aplicación mientras lee la secuencia, debe usar un hilo o una llamada asíncrona para recibir los datos.

Edición: de acuerdo con esta publicación , StreamReader.Peek tiene errores en un NetworkStream , o al menos tiene un comportamiento no documentado, así que ten cuidado si eliges ir por esa ruta.

Actualizado - respuesta a comentarios

La noción de un "vistazo" en el flujo real en sí es en realidad imposible; es solo una transmisión, y una vez que se recibe el byte, deja de estar en la transmisión. Algunas corrientes admiten la búsqueda, por lo que técnicamente podría releer ese byte, pero NetworkStream no es uno de ellos.

Peeking solo se aplica cuando se está leyendo la secuencia en un búfer; una vez que los datos están en un búfer, es fácil echar un vistazo, ya que simplemente verifica lo que haya en la posición actual en el búfer. Es por eso que un StreamReader es capaz de hacer esto; ninguna clase de Stream generalmente tendrá su propio método Peek .

Ahora, específicamente para este problema, me pregunto si esta es realmente la respuesta correcta. Entiendo la idea de seleccionar dinámicamente un método para procesar el flujo, pero ¿ realmente necesita hacer esto en el flujo sin procesar? ¿No puede leer primero el flujo en una matriz de bytes, o incluso copiarlo en un MemoryStream y procesarlo desde ese momento?

El principal problema que veo es que si algo malo sucede cuando estás leyendo de una transmisión de red, los datos desaparecen. Pero si primero lo lee en una ubicación temporal, puede depurar esto. Puede averiguar qué eran los datos y por qué el objeto que estaba tratando de procesar los datos falló a la mitad.

En general, lo primero que quiere hacer con un NetworkStream es leerlo en un búfer local. La única razón por la que puedo pensar en no hacer esto es si está leyendo una enorme cantidad de datos, e incluso entonces, podría considerar usar el sistema de archivos como un búfer intermedio si no cabe en la memoria.

No conozco sus requisitos exactos, pero por lo que he aprendido hasta ahora, mi consejo sería: No intente procesar sus datos directamente desde NetworkStream menos que haya una razón convincente para hacerlo. Considere leer los datos en la memoria o en el disco primero, y luego procesar la copia.

Answer 4

Si tiene acceso al objeto Socket , puede probar el método Receive , pasando SocketFlags.Peek . Esto es análogo al indicador MSG_PEEK que se puede pasar a la llamada recv en BSD Sockets o Winsock.