Невозможно получить ActionController :: жить, чтобы играть хорошо с ActiveSupport :: Уведомления

Question

Я искал способы запуска серверных событий с помощью ActionController :: Live без использования опроса. Похоже, что единственное документированное решение pub/sub для Live - использовать Redis, но мне бы очень хотелось использовать стек, который у меня есть. Я столкнулся с ActiveSupport :: Notifications, который, похоже, делает то, что я хочу.

Однако, когда я использую его почти на месте Redis, Live бросает IOError, и поток закрывается. Я не вижу веских причин, почему это произойдет.

Например, это якобы работает:

def stream 
    response.headers['Content-Type'] = 'text/event-stream' 
    redis = Redis.new 
    redis.subscribe('namespaced:stream') do |on| 
     response.stream.write "hello world\n" 
    end 
    rescue IOError 
     # Client Disconnected 
    ensure 
     response.stream.close 
end 

Но когда я пытаюсь это я получаю ошибку ввода-вывода:

def stream 
    response.headers['Content-Type'] = 'text/event-stream' 
    ActiveSupport::Notifications.subscribe("process_action.action_controller") do |*args| 
     response.stream.write "hello world\n" 
    end 
    rescue IOError 
     # Client Disconnected 
    ensure 
     response.stream.close 
end 

Я действительно в недоумении, почему это происходит. Ничто в журнале сервера не указывает на фактическую проблему, и подписка на уведомление работает иначе; Я могу сказать, потому что, даже если я получаю ошибку ввода-вывода, я все равно могу получить подписку, чтобы помещать строку в консоль при появлении уведомления «process_action.action_controller».

Когда уведомление не происходит, соединение работает нормально, но как только появляется уведомление (которое, я думаю, должно просто написать «привет мир» в поток), я получаю эту глупую ошибку ввода-вывода.

Кстати, я планирую использовать собственные уведомления; просто тестирование с существующими уведомлениями проще.

О, и вот копия того, что происходит в терминале, когда я посещаю маршрут, который использует метод контроллера потока:

IOError (closed stream): 
    /usr/local/lib/ruby/gems/2.0.0/gems/actionpack-4.0.2/lib/action_dispatch/http/response.rb:76:in `write' 
    /usr/local/lib/ruby/gems/2.0.0/gems/actionpack-4.0.2/lib/action_controller/metal/live.rb:47:in `write' 
    /usr/local/lib/ruby/gems/2.0.0/gems/actionpack-4.0.2/lib/action_controller/metal/live.rb:135:in `rescue in block in process' 
    /usr/local/lib/ruby/gems/2.0.0/gems/actionpack-4.0.2/lib/action_controller/metal/live.rb:145:in `block in process' 



IOError (closed stream): 
    /usr/local/lib/ruby/gems/2.0.0/gems/actionpack-4.0.2/lib/action_dispatch/http/response.rb:76:in `write' 
    /usr/local/lib/ruby/gems/2.0.0/gems/actionpack-4.0.2/lib/action_controller/metal/live.rb:47:in `write' 
    /home/benjamin/Desktop/workspace/fremote/app/controllers/remotes_controller.rb:25:in `block in show' 
    /usr/local/lib/ruby/gems/2.0.0/gems/activesupport-4.0.2/lib/active_support/notifications/fanout.rb:125:in `call' 
    /usr/local/lib/ruby/gems/2.0.0/gems/activesupport-4.0.2/lib/active_support/notifications/fanout.rb:125:in `finish' 
    /usr/local/lib/ruby/gems/2.0.0/gems/activesupport-4.0.2/lib/active_support/notifications/fanout.rb:40:in `block in finish' 
    /usr/local/lib/ruby/gems/2.0.0/gems/activesupport-4.0.2/lib/active_support/notifications/fanout.rb:40:in `each' 
    /usr/local/lib/ruby/gems/2.0.0/gems/activesupport-4.0.2/lib/active_support/notifications/fanout.rb:40:in `finish' 
    /usr/local/lib/ruby/gems/2.0.0/gems/activesupport-4.0.2/lib/active_support/notifications/instrumenter.rb:36:in `finish' 
    /usr/local/lib/ruby/gems/2.0.0/gems/activesupport-4.0.2/lib/active_support/notifications/instrumenter.rb:25:in `instrument' 
    /usr/local/lib/ruby/gems/2.0.0/gems/activesupport-4.0.2/lib/active_support/notifications.rb:159:in `instrument' 
    /usr/local/lib/ruby/gems/2.0.0/gems/actionpack-4.0.2/lib/action_controller/metal/instrumentation.rb:30:in `process_action' 
    /usr/local/lib/ruby/gems/2.0.0/gems/actionpack-4.0.2/lib/action_controller/metal/params_wrapper.rb:245:in `process_action' 
    /usr/local/lib/ruby/gems/2.0.0/gems/actionpack-4.0.2/lib/abstract_controller/base.rb:136:in `process' 
    /usr/local/lib/ruby/gems/2.0.0/gems/actionpack-4.0.2/lib/abstract_controller/rendering.rb:44:in `process' 
    /usr/local/lib/ruby/gems/2.0.0/gems/actionpack-4.0.2/lib/action_controller/metal/live.rb:132:in `block in process' 

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

EDIT: В случае, если вам нужно знать, я использую следующее:

• Rails 4.0.2
• Рубин 2.0.0p353 (2013-11-22 редакция 43784) [i686-Linux]

Answer 1

здесь хорошая статья на эту тему: http://37signals.com/svn/posts/3091-pssst-your-rails-application-has-a-secret-to-tell-you

Ну вы уверены, Redis и ActiveSupport :: Уведомления имеют один и тот же метод подписываться?

Так что это может быть то, что вызывает проблемы в рельсах:

def close 
    @response.commit! 
    @closed = true 
    end 


    def write(string) 
    raise IOError, "closed stream" if closed? 

    @response.commit! 
    @buf.push string 
    end 

Вы гарантируя, что поток закрывается. Когда вы пишете его, он передает IOError if closed?, который вы обеспечиваете. Я не уверен на 100%, что все происходит, но это похоже на самого вероятного преступника.

SO я вернулся выглядеть некоторые вещи:

в Initialize Ответа мы имеем:

self.body, self.header, self.status = body, header, status 

Вот body=

def body=(body) 
    @blank = true if body == EMPTY 

    if body.respond_to?(:to_path) 
    @stream = body 
    else 
    synchronize do 
     @stream = build_buffer self, munge_body_object(body) 
    end 
    end 
end 

который устанавливает @stream = к новому Buffer

def build_buffer(response, body) 
    Buffer.new response, body 
end 

Итак, я считаю, что ваш виновник на самом деле гарантирует вам response.stream.close. Теперь, как его называют, прежде чем писать, я не уверен. Вы делаете Redis первым, а затем тестируете Live write и используете ли они тот же ответ? Если это так, вы можете настроить ответ на закрытие, а затем попытаться использовать его для записи, что будет через IOerror.