Merge pull request #28 from blitline-dev/master

Update to work with Crystal 1.8

Author: bcardiff

src/zeromq/socket.cr

@@ -1,38 +1,40 @@
 # EventLoop support
 # Crystal Update for zmq sockets
-module Crystal::EventLoop
-  def self.create_fd_write_event(sock : ZMQ::Socket, edge_triggered : Bool = false)
-      flags = LibEvent2::EventFlags::Write
-      flags |= LibEvent2::EventFlags::Persist | LibEvent2::EventFlags::ET #we always do this as ZMQ is edge triggers
-      event = @@eb.new_event(sock.fd, flags, sock) do |_, sflags, data|
-        sock_ref = data.as(ZMQ::Socket)
-        zmq_events = sock_ref.events
-        is_writable = zmq_events & ZMQ::POLLOUT
-        if is_writable && sflags.includes?(LibEvent2::EventFlags::Write)
-          sock_ref.resume_write
-        elsif sflags.includes?(LibEvent2::EventFlags::Timeout)
-          sock_ref.resume_write(timed_out: true)
-        end
+class Crystal::LibEvent::EventLoop
+  def create_fd_write_event(sock : ZMQ::Socket, edge_triggered : Bool = false)
+    flags = LibEvent2::EventFlags::Write
+    flags |= LibEvent2::EventFlags::Persist | LibEvent2::EventFlags::ET # we always do this as ZMQ is edge triggers
+    event = event_base.new_event(sock.fd, flags, sock) do |_, sflags, data|
+      sock_ref = data.as(ZMQ::Socket)
+      zmq_events = sock_ref.events
+      is_writable = zmq_events & ZMQ::POLLOUT
+      if is_writable && sflags.includes?(LibEvent2::EventFlags::Write)
+        sock_ref.resume_write
+      elsif sflags.includes?(LibEvent2::EventFlags::Timeout)
+        sock_ref.resume_write(timed_out: true)
       end
-      event
+    end
+    event
   end
-  def self.create_fd_read_event(sock : ZMQ::Socket, edge_triggered : Bool = false)
-      flags = LibEvent2::EventFlags::Read
-      flags |= LibEvent2::EventFlags::Persist | LibEvent2::EventFlags::ET #we always do this as ZMQ is edge triggers
-      event = @@eb.new_event(sock.fd, flags, sock) do |_, sflags, data|
-        sock_ref = data.as(ZMQ::Socket)
-        zmq_events = sock_ref.events
-        is_readable = zmq_events & ZMQ::POLLIN
-        if is_readable && sflags.includes?(LibEvent2::EventFlags::Read)
-          sock_ref.resume_read
-        elsif sflags.includes?(LibEvent2::EventFlags::Timeout)
-          sock_ref.resume_read(timed_out: true)
-        end
+
+  def create_fd_read_event(sock : ZMQ::Socket, edge_triggered : Bool = false)
+    flags = LibEvent2::EventFlags::Read
+    flags |= LibEvent2::EventFlags::Persist | LibEvent2::EventFlags::ET # we always do this as ZMQ is edge triggers
+    event = event_base.new_event(sock.fd, flags, sock) do |_, sflags, data|
+      sock_ref = data.as(ZMQ::Socket)
+      zmq_events = sock_ref.events
+      is_readable = zmq_events & ZMQ::POLLIN
+      if is_readable && sflags.includes?(LibEvent2::EventFlags::Read)
+        sock_ref.resume_read
+      elsif sflags.includes?(LibEvent2::EventFlags::Timeout)
+        sock_ref.resume_read(timed_out: true)
       end
-      event
+    end
+    event
   end
-  def self.stop_loop
-    @@eb.loop_break
+
+  def stop_loop
+    event_base.loop_break
   end
 end
 
@@ -43,8 +45,8 @@ module ZMQ
     getter name : String
     getter? closed
 
-    @read_event = Crystal::ThreadLocalValue(Crystal::Event).new
-    @write_event = Crystal::ThreadLocalValue(Crystal::Event).new
+    @read_event = Crystal::ThreadLocalValue(Crystal::EventLoop::Event).new
+    @write_event = Crystal::ThreadLocalValue(Crystal::EventLoop::Event).new
 
     def self.create(context : Context, type : Int32, message_type = Message) : self
       new context, type, message_type
@@ -80,13 +82,13 @@ module ZMQ
 
     # libevent  support
     private def add_read_event(timeout = @read_timeout)
-      event = @read_event.get { Crystal::EventLoop.create_fd_read_event(self,true) }
+      event = @read_event.get { Thread.current.scheduler.@event_loop.create_fd_read_event(self, true) }
       event.add timeout
       nil
     end
 
     private def add_write_event(timeout = @write_timeout)
-      event = @write_event.get { Crystal::EventLoop.create_fd_write_event(self,true) }
+      event = @write_event.get { Thread.current.scheduler.@event_loop.create_fd_write_event(self, true) }
       event.add timeout
       nil
     end
@@ -137,11 +139,11 @@ module ZMQ
         message = @message_type.new
         rc = LibZMQ.msg_recv(message.address, @socket, flags | ZMQ::DONTWAIT)
         if rc == -1
-            if Util.errno == Errno::EAGAIN.to_i
-              wait_readable
-            else
-              raise Util.error_string
-            end
+          if Util.errno == Errno::EAGAIN.to_i
+            wait_readable
+          else
+            raise Util.error_string
+          end
         else
           return message
         end
@@ -159,7 +161,7 @@ module ZMQ
     def receive_strings(flags = 0)
       receive_messages(flags).map do |msg|
         str = msg.to_s
-        msg.close()
+        msg.close
         str
       end
     end
@@ -182,16 +184,16 @@ module ZMQ
               message = @message_type.new
               rc = LibZMQ.msg_recv(message.address, @socket, flags)
               if Util.resultcode_ok?(rc)
-                  messages << message
-                  return messages unless more_parts?
+                messages << message
+                return messages unless more_parts?
               else
                 message.close
                 messages.map(&.close)
                 return messages.clear
               end
             end
           else
-             return messages
+            return messages
           end
         end
       end
@@ -295,10 +297,12 @@ module ZMQ
     def finalize
       close
     end
+
     # file descriptor
     def fd
       get_socket_option(ZMQ::FD).as(Int32)
     end
+
     # event list
     def events
       get_socket_option(ZMQ::EVENTS).as(Int32)