Rewrite of wait_readable to use the Cellulid::ZMQ mailbox reactor

lib/celluloid/zmq.rb

@@ -10,6 +10,9 @@
 module Celluloid
   # Actors which run alongside 0MQ sockets
   module ZMQ
+    class NotZmqActorError < StandardError; end
+    class NotZmqSocketError < StandardError; end
+
     class << self
       attr_writer :context
 
@@ -31,9 +34,40 @@ def terminate
       end
     end
 
-    extend Forwardable
+    def wait_readable(socket)
+      if !socket.is_a?(::ZMQ::Socket)
+        throw NotZmqSocketError
+      end
+      actor = Thread.current[:celluloid_actor]
+      if actor && actor.mailbox.is_a?(Celluloid::ZMQ::Mailbox)
+        actor.mailbox.reactor.wait_readable(socket)
+      else
+        throw NotZmqActorError
+      end
+      nil
+    end
+    module_function :wait_readable
+
+    def wait_writable(socket)
+      actor = Thread.current[:celluloid_actor]
+      if actor && actor.mailbox.is_a?(Celluloid::ZMQ::Mailbox)
+        actor.mailbox.reactor.wait_writable(socket)
+      else
+        Kernel.select([], [io])
+      end
+      nil
+    end
+    module_function :wait_writable
+
+    # Does the 0MQ socket support evented operation?
+    def evented?
+      actor = Thread.current[:celluloid_actor]
+      return unless actor
+
+      mailbox = actor.mailbox
+      mailbox.is_a?(Celluloid::IO::Mailbox) && mailbox.reactor.is_a?(Celluloid::ZMQ::Reactor)
+    end
+    module_function :evented?
 
-    # Wait for the given IO object to become readable/writeable
-    def_delegators 'current_actor.mailbox.reactor', :wait_readable, :wait_writeable
   end
 end

lib/celluloid/zmq/sockets.rb

@@ -38,15 +38,6 @@ def close
         @socket.close
       end
 
-      # Does the 0MQ socket support evented operation?
-      def evented?
-        actor = Thread.current[:celluloid_actor]
-        return unless actor
-
-        mailbox = actor.mailbox
-        mailbox.is_a?(Celluloid::IO::Mailbox) && mailbox.reactor.is_a?(Celluloid::ZMQ::Reactor)
-      end
-
       # Hide ffi-rzmq internals
       alias_method :inspect, :to_s
     end
@@ -68,7 +59,7 @@ def connect(addr)
 
       # Read a message from the socket
       def read(buffer = '')
-        Celluloid.current_actor.wait_readable(@socket) if evented?
+        Celluloid::ZMQ.wait_readable(@socket) if Celluloid::ZMQ.evented?
 
         unless ::ZMQ::Util.resultcode_ok? @socket.recv_string buffer
           raise IOError, "error receiving ZMQ string: #{::ZMQ::Util.error_string}"