AMQP динамически создает подписку на очереди

Я пытаюсь создать простое чат-приложение с использованием AMQP, Websockets и Ruby. Я понимаю, что это может быть не лучший вариант использования AMQP, но я хотел бы понять, в чем я ошибаюсь.

Ниже приведен мой код amqp-сервера

require 'rubygems'
require 'amqp'
require 'mongo'
require 'em-websocket'
require 'json'

class MessageParser
  # message format => "room:harry_potter, nickname:siddharth, room:members"
  def self.parse(message)
    parsed_message = JSON.parse(message)

    response = {}
    if parsed_message['status'] == 'status'
      response[:status] = 'STATUS'
      response[:username] = parsed_message['username']
      response[:roomname] = parsed_message['roomname']
    elsif parsed_message['status'] == 'message'
      response[:status]   = 'MESSAGE'
      response[:message]  = parsed_message['message']
      response[:roomname] = parsed_message['roomname'].split().join('_')
    end

    response
  end
end

class MongoManager
  def self.establish_connection(database)
    @db ||= Mongo::Connection.new('localhost', 27017).db(database)
    @db.collection('rooms')

    @db
  end  
end


@sockets = []
EventMachine.run do
  connection = AMQP.connect(:host => '127.0.0.1')
  channel = AMQP::Channel.new(connection)

  puts "Connected to AMQP broker. #{AMQP::VERSION} "

  mongo = MongoManager.establish_connection("trackertalk_development")

  EventMachine::WebSocket.start(:host => '127.0.0.1', :port => 8080) do |ws|
    socket_detail = {:socket => ws}
    ws.onopen do 
      @sockets << socket_detail

    end

    ws.onmessage do |message|

      status  = MessageParser.parse(message)         
      exchange = channel.fanout(status[:roomname].split().join('_'))   

      if status[:status] == 'STATUS'               
         queue = channel.queue(status[:username], :durable => true)

        unless queue.subscribed? 
         puts "--------- SUBSCRIBED --------------"
         queue.bind(exchange).subscribe do |payload|
            puts "PAYLOAD :  #{payload}"
            ws.send(payload)
          end 
        else
          puts "----ALREADY SUBSCRIBED"
        end                  

        # only after 0.8.0rc14
        #queue = channel.queue(status[:username], :durable => true)      
        #AMQP::Consumer.new(channel, queue)        

      elsif status[:status] == 'MESSAGE'
        puts "********************* Message- published ******************************"
        exchange.publish(status[:message)  
      end                  
    end

    ws.onclose do 
      @sockets.delete ws
    end
  end    
end

. Я использую статус, чтобы указать, является ли входящее сообщение сообщением для текущего чата или для сообщения о состоянии, требующего от меня выполнения таких рутинных операций, как подписка на очередь.

Проблема, с которой я сталкиваюсь, заключается в том, что когда я отправляю сообщение типа socket.send (JSON.stringify ({status: 'message', message: 'test', roomname: 'Harry Potter'}))

Вызывается exchange.publish ', но он все равно не "не отправляются через ws.send` в браузер.

Что-то в корне неверно в моем понимании EventMachine и AMQP?

Вот пример того же кода http://pastie.org/private/xosgb8tw1w5vuroa4w7a

Мой код, кажется, работает должным образом, когда я удаляю Durable => true из queue = channel.queue (status [: username],: Durable => true)

Ниже приведен фрагмент моего представления Rails, который идентифицирует имя пользователя и имя комнаты и отправляет его как часть сообщения через Websockets. .

Хотя код, кажется, работает, когда я удаляю Durable => true , я не понимаю, почему это влияет на доставляемое сообщение. Пожалуйста, не обращайте внимания на часть mongo, поскольку она пока не играет никакой роли.

Я также хотел бы знать, верен ли мой подход к AMQP и его использованию

  

    
<%= @room.name %>