ruby - 将 eventmachine 与 sinatra 一起使用时，如何在不关闭所有连接的情况下关闭一个 http 连接？

Question

我有以下代码，它将查询 Twitter 流 API 以获取特定字符串。当我打开两个包含不同查询的选项卡时，它们都有效。但是，当我停止其中一个选项卡时，另一个选项卡停止接收任何数据，并且服务器停止处理流。我该如何解决这个问题？

require 'sinatra'
require 'eventmachine'
require 'em-http'
require 'json'
require 'sinatra/streaming'

enable :logging, :dump_errors, :raise_errors

get '/test/:query' do
  q = params[:query]
  the = 'him'
  url = "https://stream.twitter.com/1/statuses/filter.json?track=#{q}"
  stream(:keep_open) do |out|   
    http = EM::HttpRequest.new(url)
    s = http.get :head => { 'Authorization' => [ 'USERNAME', 'PASSWORD' ] }
    out.callback do
      puts "callback"
      http.conn.close_connection
    end
    out.errback do 
      puts "errback"
      http.conn.close_connection
    end
    buffer = ""
    s.stream do |chunk|
      puts "what"
      buffer += chunk
      while line = buffer.slice!(/.+\r?\n/)
        tweet = JSON.parse(line)
        unless tweet.length == 0 or tweet['user'].nil? or out.closed?
          out << "<p><b>#{tweet['user']['screen_name']}</b>: #{tweet['text']}</p>" 
        end
      end
    end
  end
end

Answer 1

问题是，一旦您打开第二个连接，Twitter 就会关闭一个连接。尝试运行 curl <a href="https://USER:PASSWORD@stream.twitter.com/1/statuses/filter.json?track=bar" rel="noreferrer noopener nofollow">https://USER:PASSWORD@stream.twitter.com/1/statuses/filter.json?track=bar</a>同时在两个终端。

此外，在尝试找出问题所在时，我做了一个相当小的重构以提高可读性:

require 'sinatra'
require 'sinatra/streaming'
require 'eventmachine'
require 'em-http'
require 'json'

enable :logging, :dump_errors, :raise_errors
template(:tweet) { "<p><b><%= @tweet['user']['screen_name'] %></b>: <%= @tweet['text'] %></p>" }

get '/test/:query' do |q|
  stream(:keep_open) do |out|
    http = EM::HttpRequest.new("https://stream.twitter.com/1/statuses/filter.json?track=#{q}")

    EM.next_tick do
      s = http.get :head => { 'Authorization' => ENV.values_at('USERNAME', 'PASSWORD') }

      s.callback { out.close }
      out.callback { s.close }

      s.errback { out.close }
      out.errback { s.close }

      buffer = ""
      s.stream do |chunk|
        buffer << chunk
        while line = buffer.slice!(/.+\r?\n/)
          break if out.closed?
          @tweet = JSON.parse(line)
          out << erb(:tweet) if @tweet.length > 0 and @tweet['user']
        end
      end
    end
  end
end