zmq pipeline sample

collector.rb

#!/usr/bin/env ruby

require 'rubygems'
require 'zmq'

context = ZMQ::Context.new
sock =context.socket(ZMQ::UPSTREAM)
sock.bind("tcp://127.0.0.1:6000")

loop do
  msg = sock.recv()
  puts msg
end

job.rb

#!/usr/bin/env ruby

require 'rubygems'
require 'zmq'

context = ZMQ::Context.new
socket  = context.socket(ZMQ::DOWNSTREAM)
socket.bind("tcp://127.0.0.1:5000")

loop do
  # send jobs to downstream worker
  puts "send job #{rand(100)}"
  msg = "Job number #{rand(100)}"
  socket.send(msg)
  sleep 1
end

worker.rb

#!/usr/bin/env ruby

require 'rubygems'
require 'zmq'

worker_num = ARGV[0] || '1'
context = ZMQ::Context.new
socket_up  = context.socket(ZMQ::UPSTREAM)
socket_down  = context.socket(ZMQ::DOWNSTREAM)

socket_up.connect("tcp://127.0.0.1:5000")
socket_down.connect("tcp://127.0.0.1:6000")

puts "Worker #{worker_num} start."

loop do
  # send jobs to downstream worker
  msg = socket_up.recv();
  puts msg
  down_msg = "[worker ##{worker_num}] #{msg}"
  socket_down.send(down_msg);
end

This was a supremely useful gist for learning zmq! I was looking for just this. One thing I noticed though is that collector directly sends to all the workers, and if it idsn't running, you can't add workers. If you do, they don't et anything...