最新编辑:在结尾处展示了可运行的代码。主模块未经修改,除了调试代码。注意:我已经遇到了我已经提到的需要在终止之前取消订阅的问题。
代码看起来正确。我想看看你是如何实例化它的。
在config/application.rb中,你可能至少有以下内容:
require 'ws_communication'
config.middleware.use WsCommunication
然后,在您的 JavaScript 客户端中,应该有类似以下的代码:
var ws = new WebSocket(uri);
你会实例化另一个 WsCommunication 实例吗?这样会将 @clients 设置为空数组,并且可能会出现你所描述的问题。下面这样的代码是错误的:
var ws = new WsCommunication;
如果这篇文章没有帮助到您,展示客户端和config/application.rb会对我们有所帮助。
顺便说一下,我同意评论中的观点,即在任何更新时都应该通过互斥锁来保护@clients,如果不是读取也是如此。它是一个动态结构,在事件驱动系统中随时可能发生变化。redis-mutex是一个不错的选择。(希望链接正确,因为Github似乎在目前的情况下会出现500错误。)
您还可以注意到$redis.publish返回接收消息的客户端数量的整数值。
最后,您可能需要确保在终止之前取消订阅您的频道。我曾经遇到过这样的情况,由于早期订阅了相同频道而没有清理,导致每条消息发送多次,甚至很多次。由于您正在线程中订阅频道,因此您需要在同一线程中取消订阅,否则进程将只是“挂起”,等待正确的线程神奇地出现。我通过设置“取消订阅”标志然后发送消息来处理这种情况。然后,在on.message块中,我测试取消订阅标志并在那里发布取消订阅。
您提供的模块,只需要进行一些小的调试修改:
require 'faye/websocket'
require 'redis'
class WsCommunication
KEEPALIVE_TIME = 15
CHANNEL = 'vip-deck'
def initialize(app)
@app = app
@clients = []
uri = URI.parse(ENV['REDISCLOUD_URL'])
$redis = Redis.new(host: uri.host, port: uri.port, password: uri.password)
Thread.new do
redis_sub = Redis.new(host: uri.host, port: uri.port, password: uri.password)
redis_sub.subscribe(CHANNEL) do |on|
on.message do |channel, msg|
puts "Message event. Clients receiving:#{@clients.count};"
@clients.each { |ws| ws.send(msg) }
end
end
end
end
def call(env)
if Faye::WebSocket.websocket?(env)
ws = Faye::WebSocket.new(env, nil, {ping: KEEPALIVE_TIME})
ws.on :open do |event|
@clients << ws
puts "Open event. Clients open:#{@clients.count};"
end
ws.on :message do |event|
receivers = $redis.publish(CHANNEL, event.data)
puts "Message published:#{event.data}; Receivers:#{receivers};"
end
ws.on :close do |event|
@clients.delete(ws)
puts "Close event. Clients open:#{@clients.count};"
ws = nil
end
ws.rack_response
else
@app.call(env)
end
end
end
我提供的测试订阅者代码:
puts "Starting client-subscriber.rb"
$:.unshift File.expand_path '../lib', File.dirname(__FILE__)
require 'rubygems'
require 'eventmachine'
require 'websocket-client-simple'
puts "websocket-client-simple v#{WebSocket::Client::Simple::VERSION}"
url = ARGV.shift || 'ws://localhost:3000'
EM.run do
ws = WebSocket::Client::Simple.connect url
ws.on :message do |msg|
puts msg
end
ws.on :open do
puts "-- Subscriber open (#{ws.url})"
end
ws.on :close do |e|
puts "-- Subscriber close (#{e.inspect})"
exit 1
end
ws.on :error do |e|
puts "-- Subscriber error (#{e.inspect})"
end
end
我提供了测试发布者代码。发布者和订阅者可以很容易地合并,因为这只是测试。
puts "Starting client-publisher.rb"
$:.unshift File.expand_path '../lib', File.dirname(__FILE__)
require 'rubygems'
require 'eventmachine'
require 'json'
require 'websocket-client-simple'
puts "websocket-client-simple v#{WebSocket::Client::Simple::VERSION}"
url = ARGV.shift || 'ws://localhost:3000'
EM.run do
count ||= 0
timer = EventMachine.add_periodic_timer(5+rand(5)) do
count += 1
send({"MESSAGE": "COUNT:#{count};"})
end
@ws = WebSocket::Client::Simple.connect url
@ws.on :message do |msg|
puts msg
end
@ws.on :open do
puts "-- Publisher open"
end
@ws.on :close do |e|
puts "-- Publisher close (#{e.inspect})"
exit 1
end
@ws.on :error do |e|
puts "-- Publisher error (#{e.inspect})"
@ws.close
end
def self.send message
payload = message.is_a?(Hash) ? message : {payload: message}
@ws.send(payload.to_json)
end
end
一个在Rack中间件层运行所有这些的示例config.ru:
require './controllers/main'
require './middlewares/ws_communication'
use WsCommunication
run Main.new
这是主程序。我将其从我的运行版本中剥离出来,因此如果您使用它,可能需要进行微调:
%w(rubygems bundler sinatra/base json erb).each { |m| require m }
ENV['RACK_ENV'] ||= 'development'
Bundler.require
$: << File.expand_path('../', __FILE__)
$: << File.expand_path('../lib', __FILE__)
Dir["./lib/*.rb", "./lib/**/*.rb"].each { |file| require file }
env = ENV['OS'] == 'Windows_NT' ? 'development' : ENV['RACK_ENV']
class Main < Sinatra::Base
env = ENV['OS'] == 'Windows_NT' ? 'development' : ENV['RACK_ENV']
get "/" do
erb :"index.html"
end
get "/assets/js/application.js" do
content_type :js
@scheme = env == "production" ? "wss://" : "ws://"
erb :"application.js"
end
end
actioncable
,它可以解决所有的问题。 - siegy22