Adding support for io/wait and wait_readable/writable API, which matc…

lib/celluloid/io.rb

@@ -1,5 +1,6 @@
 require "celluloid/io/version"
 
+require "io/wait"
 require "celluloid"
 require "celluloid/io/dns_resolver"
 require "celluloid/io/mailbox"
@@ -47,25 +48,42 @@ def self.copy_stream(src, dst, copy_length = nil, src_offset = nil)
       end
     end
 
-    def wait_readable(io)
+    def wait_readable(io, timeout=nil)
       io = io.to_io
       if IO.evented?
         mailbox = Thread.current[:celluloid_mailbox]
-        mailbox.reactor.wait_readable(io)
+        mailbox.reactor.wait_readable(io, timeout=nil)
       else
-        Kernel.select([io])
+        # hack because SSLSocket does not have the methods defined
+        if io.respond_to?(:wait_readable) && 
+           # TCP* and UNIXServer always throws EINVAL exception in Linux when using #wait_readable, 
+           # upstream to ruby!
+           !(io.is_a?(::TCPServer) || io.is_a?(::UNIXServer)) 
+          io.wait_readable(timeout)
+        else
+          Kernel.select([io], nil, nil, timeout)
+        end
       end
       nil
     end
     module_function :wait_readable
 
-    def wait_writable(io)
+    def wait_writable(io, timeout=nil)
       io = io.to_io
       if IO.evented?
         mailbox = Thread.current[:celluloid_mailbox]
-        mailbox.reactor.wait_writable(io)
+        mailbox.reactor.wait_writable(io, timeout=nil)
       else
-        Kernel.select([], [io])
+        # hack because SSLSocket does not have the methods defined
+        if io.respond_to?(:wait_writable) &&
+           # TCP* and UNIXServer always throws EINVAL exception in Linux when using #wait_readable, 
+           # upstream to ruby!
+           !(io.is_a?(::TCPServer) || io.is_a?(::UNIXServer)) 
+
+          io.wait_writable(timeout)
+        else
+          Kernel.select([], [io], nil, timeout)
+        end
       end
       nil
     end

lib/celluloid/io/reactor.rb

@@ -17,17 +17,17 @@ def initialize
       end
 
       # Wait for the given IO object to become readable
-      def wait_readable(io)
-        wait io, :r
+      def wait_readable(io, timeout=nil)
+        wait io, :r, timeout
       end
 
       # Wait for the given IO object to become writable
-      def wait_writable(io)
-        wait io, :w
+      def wait_writable(io, timeout=nil)
+        wait io, :w, timeout
       end
 
       # Wait for the given IO operation to complete
-      def wait(io, set)
+      def wait(io, set, timeout=nil)
         # zomg ugly type conversion :(
         unless io.is_a?(::IO) || io.is_a?(OpenSSL::SSL::SSLSocket)
           if io.respond_to? :to_io
@@ -43,7 +43,14 @@ def wait(io, set)
         monitor.value = Task.current
 
         begin
-          Task.suspend :iowait
+          # this condition is actually superflous; Timeout.timeout method signature is not respected
+          # by Celluloid:Actor#timeout; Timeout.timeout has arity 2 and can receive nil; the first one
+          # is already addressed in a PR in github
+          if timeout 
+            Thread.current[:celluloid_actor].timeout(timeout) { Task.suspend :iowait }
+          else
+            Task.suspend :iowait
+          end
         ensure
           # In all cases we want to ensure that the monitor is closed once we
           # have woken up. However, in some cases, the monitor is already

lib/celluloid/io/socket.rb

@@ -70,6 +70,18 @@ def self.try_convert(socket, convert_io = true)
         end
       end
 
+      # io/wait API, it belongs here as it is extended on all IOs
+
+      # Wait until the current object is readable
+      def wait_readable(*args)
+        Celluloid::IO.wait_readable(self, *args)
+      end
+
+      # Wait until the current object is writable
+      def wait_writable(*args)
+        Celluloid::IO.wait_writable(self, *args)
+      end
+
       class << self
         extend Forwardable
         def_delegators '::Socket', *(::Socket.methods - self.methods - [:try_convert])

lib/celluloid/io/stream.rb

@@ -27,12 +27,6 @@ def initialize(socket)
         @write_latch = Latch.new
       end
 
-      # Wait until the current object is readable
-      def wait_readable; Celluloid::IO.wait_readable(self); end
-
-      # Wait until the current object is writable
-      def wait_writable; Celluloid::IO.wait_writable(self); end
-
       # System read via the nonblocking subsystem
       def sysread(length = nil, buffer = nil)
         buffer ||= ''.force_encoding(Encoding::ASCII_8BIT)

lib/celluloid/io/tcp_server.rb

@@ -33,12 +33,12 @@ def initialize(*args)
 
       # @return [TCPSocket]
       def accept
-        Celluloid::IO.wait_readable(to_io)
         accept_nonblock
       end
 
       # @return [TCPSocket]
       def accept_nonblock
+        Celluloid::IO.wait_readable(to_io)
         Celluloid::IO::TCPSocket.new(to_io.accept_nonblock)
       end
 

lib/celluloid/io/udp_socket.rb

@@ -24,9 +24,6 @@ def initialize(*args)
         end
       end
 
-      # Wait until the socket is readable
-      def wait_readable; Celluloid::IO.wait_readable(self); end
-
       # Receives up to maxlen bytes from socket. flags is zero or more of the
       # MSG_ options. The first element of the results, mesg, is the data
       # received. The second element, sender_addrinfo, contains