Contents

#!/usr/bin/env ruby
# encoding: utf-8

require "bundler"
Bundler.setup

$:.unshift(File.expand_path("../../../lib", __FILE__))

require 'amqp'


AMQP.start do |session|
  ch             = AMQP::Channel.new(session)
  exchange       = ch.fanout("amq.fanout")
  reply_exchange = ch.default_exchange

  ch.queue("", :exclusive => true) do |q1|
    q1.bind(exchange).subscribe do |header, payload|
      v = rand

      puts "Replying to #{header.message_id} with #{v}"
      reply_exchange.publish(v, :routing_key => header.reply_to, :message_id => header.message_id)
    end
  end
  ch.queue("", :exclusive => true) do |q2|
    q2.bind(exchange).subscribe do |header, payload|
      v = rand

      puts "Replying to #{header.message_id} with #{v}"
      reply_exchange.publish(v, :routing_key => header.reply_to, :message_id => header.message_id)
    end
  end
  ch.queue("", :exclusive => true) do |q3|
    q3.bind(exchange).subscribe do |header, payload|
      v = rand

      puts "Replying to #{header.message_id} with #{v}"
      reply_exchange.publish(v, :routing_key => header.reply_to, :message_id => header.message_id)
    end
  end


  requests = Hash.new

  EventMachine.add_timer(0.5) do
    ch.queue("", :exlusive => true) do |q|
      q.subscribe do |header, payload|
        requests[header.message_id].push(payload)

        puts "Got a reply for #{header.message_id}"

        if requests[header.message_id].size == 3
          puts "Gathered all 3 responses: #{requests[header.message_id].join(', ')}"
          requests[header.message_id].clear
        end
      end


      message_id           = "__message #{rand}__"
      requests[message_id] = Array.new

      exchange.publish("a request", :reply_to => q.name, :message_id => message_id)
    end
  end


  EventMachine.add_timer(2) do
    session.close { EventMachine.stop }
  end
end

Version data entries

59 entries across 59 versions & 1 rubygems

Version	Path
amqp-1.8.0	examples/publishing/simplistic_scatter_gather.rb
amqp-1.7.0	examples/publishing/simplistic_scatter_gather.rb
amqp-1.6.0	examples/publishing/simplistic_scatter_gather.rb
amqp-1.5.3	examples/publishing/simplistic_scatter_gather.rb
amqp-1.5.2	examples/publishing/simplistic_scatter_gather.rb
amqp-1.5.1	examples/publishing/simplistic_scatter_gather.rb
amqp-1.5.0	examples/publishing/simplistic_scatter_gather.rb
amqp-1.4.2	examples/publishing/simplistic_scatter_gather.rb
amqp-1.4.1	examples/publishing/simplistic_scatter_gather.rb
amqp-1.4.0	examples/publishing/simplistic_scatter_gather.rb
amqp-1.3.0	examples/publishing/simplistic_scatter_gather.rb
amqp-1.2.1	examples/publishing/simplistic_scatter_gather.rb
amqp-1.2.0	examples/publishing/simplistic_scatter_gather.rb
amqp-1.1.8	examples/publishing/simplistic_scatter_gather.rb
amqp-1.1.7	examples/publishing/simplistic_scatter_gather.rb
amqp-1.1.6	examples/publishing/simplistic_scatter_gather.rb
amqp-1.1.5	examples/publishing/simplistic_scatter_gather.rb
amqp-1.1.4	examples/publishing/simplistic_scatter_gather.rb
amqp-1.1.3	examples/publishing/simplistic_scatter_gather.rb
amqp-1.0.4	examples/publishing/simplistic_scatter_gather.rb