Достъп до променлива в релсова нишка

АКТУАЛИЗИРАНА РЕДАКЦИЯ В КРАЯ:Показва работещ код. Основният модул не е модифициран, с изключение на кода за отстраняване на грешки. Забележка:Изпитах проблема, който вече отбелязах по отношение на необходимостта от прекратяване на абонамента преди прекратяване.

Кодът изглежда правилен. Бих искал да видя как го създавате.

В 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 #seconds
  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

Тестовият абонатен код, който предоставих:

# encoding: UTF-8
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

Тестовият код на издателя, който предоставих. Издател и абонат могат лесно да се комбинират, тъй като това са само тестове:

# encoding: UTF-8
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

Примерна 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