我正在尝试使用AMQP, Websockets和Ruby构建一个简单的聊天应用程序。我知道这可能不是理解AMQP的最佳用例，但我想了解我在哪里出错了。
以下是我的amqp-server代码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' is called but it still doesn't get pushed via the ws。发送'到浏览器。
我对EventMachine和AMQP的理解是否有根本性的错误?
下面是相同代码的粘贴http://pastie.org/private/xosgb8tw1w5vuroa4w7a
当我从queue = channel.queue(status[:username], :durable => true)中删除durable => true时，我的代码似乎按预期工作
下面是我的Rails视图的一个片段，它标识用户的用户名和房间名，并将其作为消息的一部分通过Websockets发送。
虽然代码似乎工作时，我删除durable => true我不明白为什么这会影响传递的消息。请忽略的mongo部分，因为它还没有发挥任何作用。
我还想知道我的AMQP方法及其用法是否正确
<script>
$(document).ready(function(){
var username = '<%= @user.email %>';
var roomname = 'Bazingaa';
socket = new WebSocket('ws://127.0.0.1:8080/');
socket.onopen = function(msg){
console.log('connected');
socket.send(JSON.stringify({status:'status', username:username, roomname:roomname}));
}
socket.onmessage = function(msg){
$('#chat-log').append(msg.data);
}
});
</script>
<div class='block'>
<div class='content'>
<h2 class='title'><%= @room.name %></h2>
<div class='inner'>
<div id="chat-log">
</div>
<div id="chat-console">
<textarea rows="5" cols="40"></textarea>
</div>
</div>
</div>
</div>
<style>
#chat-log{
color:#000;
font-weight:bold;
margin-top:1em;
width:900px;
overflow:auto;
height:300px;
}
#chat-console{
bottom:10px;
}
textarea{
width:100%;
height:60px;
}
</style>
我认为您的问题可能是队列在调用ws.onmessage之间挂在代理上。当客户端重新连接队列时，绑定已经存在，因此不会调用ws.send()。
默认情况下，当你创建一个队列时，它和它所拥有的任何绑定都会一直挂起，直到代理重新启动，或者你显式地告诉代理删除它。
有两种方法可以改变:
- 在创建队列时添加持久标志，这将导致即使代理重新启动队列也会保留
- 添加auto_delete标志，这将导致代理在短时间内没有消费者附加到实体后自动删除实体
如果你使用rabbitmq代理来控制代理，一个简单的方法是安装管理插件，它为代理上的交换、绑定和队列提供了一个web界面。
在第一次看AMQP位似乎是好的，但我不想设置所有的依赖关系。如果您提供一个最小的示例，只有AMQP部分，我将检查它。