ZeroMQ Dealer-Router pattern with Celluloid::ZMQ

zmq-test-dealer.rb

require 'celluloid/zmq'

Celluloid::ZMQ.init

class Client
  include Celluloid::ZMQ

  def initialize(address)
    @socket = DealerSocket.new

    begin
      @socket.connect(address)
    rescue IOError
      @socket.close
      raise
    end
  end

  def run
    loop do
      puts "Waiting for response..."
      async.handle_response(@socket.read_multipart)
    end
  end

  def write(*message)
    @socket.send(*message)
    nil
  end

  def handle_response(msg)
    puts "Received: #{msg}"
  end
end

addr = 'tcp://127.0.0.1:3435'

clients = Client.new(addr)
clients.async.run

# Fire loads of non-blocking requests at the same time over the same socket
100.times do |i|
  clients.write("#{i}", "Test #{i}")
end

sleep 2

zmq-test-req.rb

require 'celluloid/zmq'

Celluloid::ZMQ.init

class Client
  include Celluloid::ZMQ

  def initialize(address)
    @socket = ReqSocket.new

    begin
      @socket.connect(address)
    rescue IOError
      @socket.close
      raise
    end
  end

  def write(*message)
    @socket.send(*message)
    handle_response(@socket.read_multipart)
  end

  def handle_response(msg)
    puts "Received: #{msg}"
  end
end

addr = 'tcp://127.0.0.1:3435'

# Send a simple req using a single socket and wait for response
client = Client.new(addr)
client.write("Test 1")

zmq-test-router.rb

require 'celluloid/zmq'

Celluloid::ZMQ.init

class Server
  include Celluloid::ZMQ

  def initialize(address)
    @socket = RouterSocket.new

    begin
      @socket.bind(address)
    rescue IOError
      @socket.close
      raise
    end
  end

  def run
    loop do
      puts "Waiting for requests..."
      async.handle_message @socket.read_multipart
    end
  end

  def handle_message(parts)
    puts "PARTS: #{parts.inspect}"
    sleep 1
    puts "Sending response #{parts[2]}"
    @socket.send(parts[0], parts[1], "Received #{parts[2]}")
  end
end

addr = 'tcp://127.0.0.1:3435'

server = Server.new(addr)

server.run