reactor.rb

require 'bundler/setup'
require 'zmq'
require 'delegate'

module SM
  module Reactor
    Context = ZMQ::Context.new

    class DispatchQueue < DelegateClass(Array)
      def initialize
        @queue = []
        super(@queue)
      end

      def add_block(&block)
        @queue.push(block)
      end

      def run_next
        block = @queue.shift
        if block 
          block.call
        else
          raise 
        end
      end

      def exhaust
        loop { run_next }
      rescue
      end
    end

    class Client
      def initialize(address="inproc://sm-reactor")
        @socket = Context.socket(:PUSH)
        @address = address
        connect
      end

      def connect
        @socket.connect(@address)
      end

      def send(msg)
        @socket.send(msg)
      end

      def shutdown
        @socket.close
      end
    end

    class Server
      CHARS = ('A'..'Z').to_a

      DISPATCH_TABLE = { 
        :generate_character => Proc.new { CHARS[rand(CHARS.length).to_i] },
        :generate_int       => Proc.new { 1 },
        :generate_float     => Proc.new { 1.0 }
      }

      attr_reader :dispatch_queue

      def initialize(dispatch_queue, address="inproc://sm-reactor")
        @dispatch_queue = dispatch_queue
        @socket = Context.socket(:PULL)
        @address = address
        bind
      end

      def bind
        @socket.bind(@address)
      end

      def react
        msg = @socket.recv_nonblock

        if msg 
          if DISPATCH_TABLE.has_key?(msg.to_sym)
            @dispatch_queue.add_block(&DISPATCH_TABLE[msg.to_sym])
          end
        end
      end

      def shutdown
        @socket.close
      end
    end
  end
end

sm_react_server.rb

require 'reactor'
dq = SM::Reactor::DispatchQueue.new
server = SM::Reactor::Server.new(dq)
client = SM::Reactor::Client.new

trap(:INT) { client.shutdown; server.shutdown; exit } 

messages = %w[generate_float generate_int generate_character]

loop do
  server.react
  client.send messages[rand(messages.length).to_i]
  puts dq.run_next if dq.count > 0
end

client.shutdown
server.shutdown

Leak fixed: methodmissing/rbczmq@bf2c914