When using eventmachine with sinatra, how can I close one http connection without closing them all?

I have the following code which will query the twitter streaming开发者_StackOverflow API for a certain string. When I open two tabs with different queries, they both work. However, when I stop one of the tabs, the other tab stops receiving any data, and the server stops processing the stream. How can I fix this?

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

The problem is that Twitter closes one connection once you open a second one. Try running curl https://USER:PASSWORD@stream.twitter.com/1/statuses/filter.json?track=bar in two terminals at the same time.

Also, while trying to figure out what is wrong, I did a rather small refactoring to improve readability:

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