无法让 ActionController::Live 与 ActiveSupport::Notifications 配合得很好

Question

我一直在寻找在不使用轮询的情况下使用 ActionController::Live 触发服务器端事件的方法。似乎唯一记录在案的 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

但是当我尝试这个时，我得到一个 IO 错误：

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

我真的不知道为什么会发生这种情况。服务器日志中没有任何内容表明存在实际问题，否则通知订阅工作；我可以说出来，因为即使我收到 IO 错误，我仍然可以订阅以在“process_action.action_controller”通知发生时将字符串放入控制台。

当通知没有发生时，连接工作得很好，但是一旦有通知（我认为应该只是将“hello world”写入流），我就会收到那个愚蠢的 IO 错误。

顺便说一句，我确实计划检测自己的通知；仅使用现有通知进行测试更容易。

哦，这是我访问使用流控制器方法的路由时在终端中发生的情况的副本：

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 的新手，所以我非常感谢任何帮助。

编辑：如果您需要知道，我正在使用以下内容：

• Rails 4.0.2
• Ruby 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::Notifications 有相同的订阅方法吗？

所以这可能是导致 Rails 出现问题的原因：

  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% 确定一切都在发生，但这似乎是最有可能的罪魁祸首。

所以我回去看看一些东西：

在 Response 的初始化中我们有：

  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。现在，在调用 write 之前如何调用它，我不确定。你是先做 Redis，然后测试 Live 写，他们是否使用相同的响应？如果是这样，您可能正在将响应设置为关闭，然后尝试使用它来写入，这将通过 IOerror。