diff --git a/lib/io/stream/buffered.rb b/lib/io/stream/buffered.rb index 645dadd..9da02e4 100644 --- a/lib/io/stream/buffered.rb +++ b/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. diff --git a/lib/io/stream/readable.rb b/lib/io/stream/readable.rb index a0792ac..e3c1162 100644 --- a/lib/io/stream/readable.rb +++ b/lib/io/stream/readable.rb @@ -23,7 +23,7 @@ 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, @@ -31,6 +31,7 @@ module 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. diff --git a/releases.md b/releases.md index 29f7849..1abb706 100644 --- a/releases.md +++ b/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. diff --git a/test/io/stream/generic.rb b/test/io/stream/generic.rb index 1aa7c72..ce8ea0a 100644 --- a/test/io/stream/generic.rb +++ b/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) diff --git a/test/io/stream/tls_readable.rb b/test/io/stream/tls_readable.rb new file mode 100644 index 0000000..9e99451 --- /dev/null +++ b/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