Add stream method for non-destructive reading

[agronholm]

Things to check first

• I have searched the existing issues and didn't find my feature already requested there

Feature description

Sometimes applications or protocol implementations would benefit from the ability to see if there's data ready for reading from the stream, but without actually removing said data from the buffer. For socket streams, this could be implemented using recv() with the MSG_PEEK flag. This should work on all platforms. For TLS streams, this is also doable but a tad more complex. One consideration would be that the stream might not return the same result when a destructive read is done after a non-destructive read with a yield point in between, but I think this could just be documented (and can be avoided by not sharing the stream among tasks in the first place).

Use case

This was mainly inspired by redis-py wanting to defensively check if the stream has any more data to be read. That said, I've come across other cases which I can't quite remember, where the knowledge of pending incoming data would help optimize the operation of a protocol implementation.

[11kkw]

I believe the peek functionality, which allows non-destructive data inspection, is useful and I’d like to implement it. I have a few considerations regarding this. I'm wondering which approach would be more appropriate: implementing it separately for each stream type (e.g., SocketStream, TLSStream), or integrating it into BufferedByteReceiveStream.

Personally, I think implementing it in BufferedByteReceiveStream is more appropriate, as it would provide a consistent interface and reduce code duplication. However, if you have a different opinion or suggestions for improvement regarding my reasoning, I would greatly appreciate your input.

[m-u-xyz]

Personally, I think implementing it in BufferedByteReceiveStream is more appropriate

On the one hand: +1 because it's the generic solution; adding a method that does this (or two; one to return a view of the current buffer and another to wait for N more bytes without returning them) is almost absurdly easy.

On the other hand: sometimes you want to peek into the OS buffer and then hand the raw socket off to another process. You can't do that with BufferedByteReceiveStream.

On the gripping hand you can't do that with a TLSStream either. Thus I'd implement the generic method.

[agronholm]

I can't imagine how you would implement this in a generic fashion. Socket streams need sock.recv(..., MSG_PEEK) or on asyncio, checking the buffer. TLSStream needs to shovel all available data from the backing stream into the buffer and see if there's anything to read. Each stream implementation has its specific needs that must be considered in the implementation.

[11kkw]

Following up on the earlier discussion regarding non-destructive reads — I’ve been exploring the idea of adding a peek() method to the stream classes, and I’d really appreciate your thoughts before opening a PR.

Here's a brief summary of what I’ve done:

• Added peek() support to SocketStream, UNIXSocketStream, and TLSStream
• Defined the method in the abc.SocketStream interface
• Implemented tests for each backend to ensure consistent, non-destructive behavior

Each backend required a slightly different approach:

• asyncio: reads from the internal protocol buffer
• trio / unix: uses socket.MSG_PEEK
• TLSStream: uses an internal buffer to support peeking

Would you mind taking a quick look and letting me know if this direction seems appropriate?
I'm especially interested in any concerns you might have around how peek() is integrated into the existing logic, or any edge cases I may have overlooked.

Thank you very much in advance!

[agronholm]

I had intended for such functionality to be non-blocking. That is, to return any data that is immediately available, but not waiting for more to arrive. So basically, receive_nowait() but non-destructive.

[11kkw]

I had intended for such functionality to be non-blocking. That is, to return any data that is immediately available, but not waiting for more to arrive. So basically, receive_nowait() but non-destructive.

Thank you for the clarification!

Just to make sure I understand correctly — you're suggesting that peek() should behave like a non-blocking receive_nowait(), in the sense that it should:

• Immediately return any available data
• But never wait for more to arrive

Did I get that right?

In my current implementation, peek() may wait for data if none is immediately available (especially in asyncio or TLSStream). If the expectation is to strictly avoid blocking, I can revise the logic accordingly.

Would you recommend raising an exception when there's no data available (like BlockingIOError), or would returning an empty bytes object be acceptable in that case?

I'll give it another try with the non-blocking behavior in mind and update the implementation accordingly.

Hope it's okay if I come back with another question once I have the revised version. Really appreciate your time and feedback

[agronholm]

The receive_nowait(), already present in memory object streams, is synchronous and raises WouldBlock if there is no data to return.

These two methods have so much in common that I wonder if peeking should just be a keyword argument to receive_nowait().

[11kkw]

The receive_nowait(), already present in memory object streams, is synchronous and raises WouldBlock if there is no data to return.

These two methods have so much in common that I wonder if peeking should just be a keyword argument to receive_nowait().

receive_nowait(peek=True) would definitely help reduce duplication and make the API surface cleaner.

Just to clarify — are you envisioning this as a shared method across all stream classes (not just memory streams), where the peek argument simply controls whether the data is consumed?

And to confirm the expected behavior:

• Return available data immediately
• Raise WouldBlock if no data is available
• Only consume the data if peek=False

Does that align with what you had in mind?

Happy to explore this approach further if it sounds like the right direction.

[agronholm]

The receive_nowait(), already present in memory object streams, is synchronous and raises WouldBlock if there is no data to return.
These two methods have so much in common that I wonder if peeking should just be a keyword argument to receive_nowait().

receive_nowait(peek=True) would definitely help reduce duplication and make the API surface cleaner.

Just to clarify — are you envisioning this as a shared method across all stream classes (not just memory streams), where the peek argument simply controls whether the data is consumed?

And to confirm the expected behavior:

• Return available data immediately
• Raise WouldBlock if no data is available
• Only consume the data if peek=False

Does that align with what you had in mind?

Happy to explore this approach further if it sounds like the right direction.

Yes, this sounds exactly what I wanted. Note that this is a backwards incompatible change and will require a major version bump when merged.

[agronholm]

The biggest issue here is how to handle wrapped streams: Can, for example, TLSStream perform destructive reads on the wrapped stream, while it itself still returns the same data on a successive read?