8000 `Async::IO::Stream` can handle native IO. (#82) · socketry/async-io@af9164d · GitHub
[go: up one dir, main page]

Skip to content
This repository was archived by the owner on Aug 29, 2024. It is now read-only.

Commit af9164d

Browse files
authored
Async::IO::Stream can handle native IO. (#82)
1 parent 1c4f93b commit af9164d

File tree

2 files changed

+41
-2
lines changed

2 files changed

+41
-2
lines changed

lib/async/io/stream.rb

Lines changed: 21 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -197,6 +197,10 @@ def connected?
197197
@io.connected?
198198
end
199199

200+
def readable?
201+
@io.readable?
202+
end
203+
200204
def closed?
201205
@io.closed?
202206
end
@@ -246,6 +250,21 @@ def eof!
246250

247251
private
248252

253+
def sysread(size, buffer)
254+
while true
255+
result = @io.read_nonblock(size, buffer, exception: false)
256+
257+
case result
258+
when :wait_readable
259+
@io.wait_readable
260+
when :wait_writable
261+
@io.wait_writable
262+
else
263+
return result
264+
end
265+
end
266+
end
267+
249268
# Fills the buffer from the underlying stream.
250269
def fill_read_buffer(size = @block_size)
251270
# We impose a limit because the underlying `read` system call can fail if we request too much data in one go.
@@ -257,12 +276,12 @@ def fill_read_buffer(size = @block_size)
257276
flush
258277

259278
if @read_buffer.empty?
260-
if @io.read_nonblock(size, @read_buffer, exception: false)
279+
if sysread(size, @read_buffer)
261280
# Console.logger.debug(self, name: "read") {@read_buffer.inspect}
262281
return true
263282
end
264283
else
265-
if chunk = @io.read_nonblock(size, @input_buffer, exception: false)
284+
if chunk = sysread(size, @input_buffer)
266285
@read_buffer << chunk
267286
# Console.logger.debug(self, name: "read") {@read_buffer.inspect}
268287

spec/async/io/stream_spec.rb

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,26 @@
2323
end
2424
end
2525

26+
context "native I/O", if: RUBY_VERSION >= "3.1" do
27+
let(:sockets) do
28+
@sockets = ::Socket.pair(::Socket::AF_UNIX, ::Socket::SOCK_STREAM)
29+
end
30+
31+
after do
32+
@sockets.each(&:close)
33+
end
34+
35+
let(:io) {sockets.first}
36+
subject {described_class.new(sockets.last)}
37+
38+
it "can read data" do
39+
io.write("Hello World")
40+
io.close_write
41+
42+
expect(subject.read).to be == "Hello World"
43+
end
44+
end
45+
2646
context "socket I/O" do
2747
let(:sockets) do
2848
@sockets = Async::IO::Socket.pair(Socket::AF_UNIX, Socket::SOCK_STREAM)

0 commit comments

Comments
 (0)
0