Add a non-blocking partial peek for layered streams

lib/io/stream/buffered.rb

@@ -6,6 +6,7 @@
 require_relative "generic"
 require_relative "connection_reset_error"
 
+# Provides buffered IO streams with consistent read, write, and transport semantics.
 module IO::Stream
 	# A buffered stream implementation that wraps an underlying IO object to provide efficient buffered reading and writing.
 	class Buffered < Generic
@@ -104,6 +105,11 @@ def syswrite(buffer)
 			return @io.write(buffer)
 		end
 
+		# Attempts to read data from the underlying stream without blocking.
+		def sysread_nonblock(size, buffer)
+			return @io.read_nonblock(size, buffer, exception: false)
+		end
+
 		# Reads data from the underlying stream as efficiently as possible.
 		def sysread(size, buffer)
 			# Come on Ruby, why couldn't this just return `nil`? EOF is not exceptional. Every file has one.

lib/io/stream/readable.rb

@@ -23,14 +23,15 @@ module IO::Stream
 
 	# A module providing readable stream functionality.
 	#
-	# You must implement the `sysread` method to read data from the underlying IO.
+	# You must implement the `sysread` method to read data from the underlying IO. You may implement `sysread_nonblock` to support non-blocking partial peeks.
 	module Readable
 		ASYNC_SAFE = {
 			read: :readable,
 			read_partial: :readable,
 			read_exactly: :readable,
 			read_until: :readable,
 			peek: :readable,
+			peek_partial: :readable,
 			gets: :readable,
 			getc: :readable,
 			getbyte: :readable,
@@ -246,6 +247,10 @@ def discard_until(pattern, offset = 0, limit: nil)
 		# @parameter size [Integer | Nil] The number of bytes to peek at. If nil, peek at all available data.
 		# @returns [String] The data in the buffer without consuming it.
 		def peek(size = nil)
+			if size == 0
+				return String.new(encoding: Encoding::BINARY)
+			end
+
 			if size
 				until @finished or @read_buffer.bytesize >= size
 					# Compute the amount of data we need to read from the underlying stream:
@@ -265,6 +270,41 @@ def peek(size = nil)
 			return @read_buffer
 		end
 
+		# Peek at data without consuming it, making at most one non-blocking read attempt.
+		#
+		# Any data read from the underlying stream is preserved in the read buffer. If
+		# the read would block or the stream is at EOF, this method returns `nil`.
+		#
+		# After this method returns `nil`, {readable?} indicates whether the read would
+		# block or EOF was observed.
+		#
+		# @parameter size [Integer] The maximum number of bytes to peek at.
+		# @returns [String | Nil] The immediately available data, or nil if no data can be read without blocking.
+		def peek_partial(size = @minimum_read_size)
+			if size == 0
+				return String.new(encoding: Encoding::BINARY)
+			end
+
+			if @read_buffer.empty?
+				if @finished
+					return nil
+				end
+
+				read_size = [size, @maximum_read_size].min
+
+				result = sysread_nonblock(read_size, @read_buffer)
+				case result
+				when :wait_readable, :wait_writable
+					return nil
+				when nil
+					@finished = true
+					return nil
+				end
+			end
+
+			return @read_buffer.byteslice(0, [size, @read_buffer.bytesize].min)
+		end
+
 		# Read a line from the stream, similar to IO#gets.
 		# @parameter separator [String] The line separator to search for.
 		# @parameter limit [Integer | Nil] The maximum number of bytes to read.
@@ -360,6 +400,12 @@ def close_read
 
 		private
 
+		# Attempts to read data from the underlying stream without blocking.
+		# Implementations may override this method when non-blocking reads are supported.
+		def sysread_nonblock(size, buffer)
+			return :wait_readable
+		end
+
 		# Fills the buffer from the underlying stream.
 		def fill_read_buffer(size = @minimum_read_size)
 			# Limit the read size to avoid exceeding SSIZE_MAX and to manage memory usage.

releases.md

@@ -1,5 +1,9 @@
 # Releases
 
+## Unreleased
+
+  - Add `IO::Stream::Readable#peek_partial` to peek through layered transports without blocking or consuming application data.
+
 ## v0.13.1
 
   - Set minimum Ruby verison to 3.3.6 to avoid hanging `close` issue in older Ruby versions.

test/io/stream/generic.rb

@@ -20,6 +20,13 @@
 		end
 	end
 
+	with "#peek_partial" do
+		it "should default to no immediately available data" do
+			expect(stream.peek_partial(1)).to be_nil
+			expect(stream).to be(:readable?)
+		end
+	end
+
 	with "#flush" do
 		it "should raise NotImplementedError" do
 			expect{stream.write("hello"); stream.flush}.to raise_exception(NotImplementedError)

test/io/stream/tls_readable.rb

@@ -0,0 +1,101 @@
+# frozen_string_literal: true
+
+# Released under the MIT License.
+# Copyright, 2026, by Samuel Williams.
+
+require "io/stream/buffered"
+
+require "sus/fixtures/async/reactor_context"
+require "sus/fixtures/openssl/verified_certificate_context"
+require "sus/fixtures/openssl/valid_certificate_context"
+
+describe IO::Stream::Buffered do
+	include Sus::Fixtures::Async::ReactorContext
+	include Sus::Fixtures::OpenSSL::VerifiedCertificateContext
+	include Sus::Fixtures::OpenSSL::ValidCertificateContext
+
+	before do
+		listener = TCPServer.new("localhost", 0)
+		port = listener.local_address.ip_port
+
+		@sockets = [
+			TCPSocket.new("localhost", port),
+			listener.accept,
+		]
+		listener.close
+
+		client = OpenSSL::SSL::SSLSocket.new(@sockets[0], client_context)
+		server = OpenSSL::SSL::SSLSocket.new(@sockets[1], server_context)
+
+		client.sync_close = true
+		server.sync_close = true
+
+		accept = Async do
+			server.accept
+		end
+
+		connect = Async do
+			client.connect
+		end
+
+		[accept, connect].each(&:wait)
+
+		@client = IO::Stream::Buffered.wrap(client)
+		@server = IO::Stream::Buffered.wrap(server)
+	end
+
+	after do
+		@client&.close
+		@server&.close
+		@sockets.each do |socket|
+			unless socket.closed?
+				socket.close
+			end
+		end
+	end
+
+	attr :client
+	attr :server
+
+	it "detects a TLS close notification" do
+		closing = reactor.async do
+			server.close
+		end
+
+		@sockets[0].wait_readable(1)
+
+		expect(client.peek_partial(1)).to be_nil
+		expect(client).not.to be(:readable?)
+		expect(client.peek_partial(0)).to be == ""
+		closing.wait
+	end
+
+	it "detects an abrupt TLS connection close" do
+		@sockets.last.close
+		@server = nil
+
+		@sockets.first.wait_readable(1)
+
+		expect do
+			client.peek_partial(1)
+		end.to raise_exception(OpenSSL::SSL::SSLError)
+	end
+
+	it "reports when reading an open TLS connection would block" do
+		expect(client.peek_partial(0)).to be == ""
+		expect(client.peek_partial(1)).to be_nil
+		expect(client).to be(:readable?)
+	end
+
+	it "preserves data consumed by the readability probe" do
+		server.write("Hello")
+		server.flush
+
+		@sockets[0].wait_readable(1)
+
+		expect(client.peek_partial(0)).to be == ""
+		expect(client.peek_partial(1)).to be == "H"
+		expect(client.peek(0)).to be == ""
+		expect(client.read(5)).to be == "Hello"
+	end
+end