Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions lib/io/stream/buffered.rb
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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.
Expand Down
45 changes: 44 additions & 1 deletion lib/io/stream/readable.rb
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -265,6 +266,42 @@ 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.
# A size of zero still performs the read attempt, returning an empty string if
# data was available and preserving that data in the read buffer.
#
# @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 @read_buffer.empty?
if @finished
return nil
end

read_size = [size, @maximum_read_size].min
if read_size == 0
read_size = @minimum_read_size
end

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.
Expand Down Expand Up @@ -360,6 +397,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.
Expand Down
4 changes: 4 additions & 0 deletions releases.md
Original file line number Diff line number Diff line change
@@ -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.
Expand Down
7 changes: 7 additions & 0 deletions test/io/stream/generic.rb
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
100 changes: 100 additions & 0 deletions test/io/stream/tls_readable.rb
Original file line number Diff line number Diff line change
@@ -0,0 +1,100 @@
# 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(0)).to be_nil
expect(client.peek_partial(1)).to be_nil
expect(client).not.to be(:readable?)
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_nil
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.read(5)).to be == "Hello"
end
end
Loading