Stream Multiple Body Using Async Sinatra

stream multiple body using async sinatra

require 'rubygems'
require 'sinatra/async'
require 'thin'
require 'json'

class Test < Sinatra::Base
  register Sinatra::Async

  class JSONStream
    include EventMachine::Deferrable

    def stream(object)
      @block.call object.to_json + "\n"
    end

    def each(&block)
      @block = block
    end
  end

  aget '/process' do
    puts 'ok'
    out = JSONStream.new
    body out
    EM.next_tick do
      c = 0
      timer = EM.add_periodic_timer(0.3) do
        c += 1
        out.stream :data => ["this is part #{c}"]
        if c == 100
          timer.cancel
          out.succeed
        end
      end
    end
  end

  run!
end

See also: http://confreaks.net/videos/564-scotlandruby2011-real-time-rack

Asynchronously iterating over the response of a request using Thin and Sinatra

So in the end, I found out that the example did indeed work and I could eventually get Sinatra to stream each-able results concurrently, primarily using the EM.defer idea in Pusher and Async page. curl and Apache benchmarking confirmed that this was working.

The reason why it didn't work in the browser is because browsers limit the number of connections to the same URL. I was aware of there being a limit to concurrent connections to a single domain (also a low number), but not that (seemingly) all connections to a single URI are serialized:

http://maillist.caucho.com/pipermail/resin-interest/2009-August/003998.html

I don't know if this is configurable, I only see domain-wide configuration in Firefox, but that was the issue.

async requests using sinatra streaming API

to answer my own question:

require 'em-hiredis'
require 'sinatra/base'

class App < Sinatra::Base
  def redis
    @redis ||= EM::Hiredis.connect
  end

  get '/' do
    stream :keep_open do |out|
      redis.blpop('abcdef', 15).callback do |x|
        out << "x=#{x}"
        out.close
      end.errback do |e|
        out << "e=#{e}"
        out.close
      end
    end
  end

  run! if app_file == $0
end

sinatra async 1.0 returns unauthorized while 0.5.0 does not

Here's how I fixed it: I abandoned the vb.net native method, and used chilkat's http component. Works great. Here's the code:

Private Function GetOutputChilkatPostJson(jsonData As String) As String
    'Dim req As New Chilkat.HttpRequest()
    Dim http As New Chilkat.Http()
    Dim res As String
    Dim req_str As String
    Dim success As Boolean

    req_str = _urlString & ":" & _portString & "/execfqy"

    '  Any string unlocks the component for the 1st 30 days.
    success = http.UnlockComponent("Anything for 30-day trial")
    If (success <> True) Then
        MsgBox(http.LastErrorText)
        Exit Function
    End If

    '  This example duplicates the HTTP POST shown at
    '  http://json.org/JSONRequest.html

    '  Specifically, the request to be sent looks like this:
    '  
    'POST /request HTTP/1.1
    'Accept: application/jsonrequest
    '        Content(-Encoding) : identity()
    'Content-Length: 72
    'Content-Type: application/jsonrequest
    'Host:   json.penzance.org()

    '{"user":"doctoravatar@penzance.com","forecast":7,"t":"vlIj","zip":94089}

    '  First, remove default header fields that would be automatically
    '  sent.  (These headers are harmless, and shouldn't need to
    '  be suppressed, but just in case...)
    http.AcceptCharset = ""
    http.UserAgent = ""
    http.AcceptLanguage = ""
    http.Login = "myusername"
    http.Password = "mypassword"
    '  Suppress the Accept-Encoding header by disallowing
    '  a gzip response:
    http.AllowGzip = False

    '  If a Cookie needs to be added, it may be added by calling
    '  AddQuickHeader:
    'http.AddQuickHeader("Cookie", "JSESSIONID=1234")

    '  To use SSL/TLS, simply use "https://" in the URL.

    '  IMPORTANT: Make sure to change the URL, JSON text,
    '  and other data items to your own values.  The URL used
    '  in this example will not actually work.

    Dim resp As Chilkat.HttpResponse
    Try
        resp = http.PostJson(req_str, jsonData)
        res = resp.BodyStr
    Catch e As WebException
        Throw New NoResultsError("Request return nil results with error: " & e.Message)
        res = Nothing
    End Try
    Return res
End Function

Rack concurrency - rack.multithread, async.callback, or both?

Note: I use Thin as synonym for all web servers implementing the async Rack extension (i.e. Rainbows!, Ebb, future versions of Puma, ...)

Q1. Correct. It will wrap the response generation (aka call) in EventMachine.defer { ... }, which will cause EventMachine to push it onto its built-in thread pool.

Q2. Using async.callback in conjunction with EM.defer actually makes not too much sense, as it would basically use the thread-pool, too, ending up with a similar construct as described in Q1. Using async.callback makes sense when only using eventmachine libraries for IO. Thin will send the response to the client once env['async.callback'] is called with a normal Rack response as argument.

If the body is an EM::Deferrable, Thin will not close the connection until that deferrable succeeds. A rather well kept secret: If you want more than just long polling (i.e. keep the connection open after sending a partial response), you can also return an EM::Deferrable as body object directly without having to use throw :async or a status code of -1.

Q3. You're guessing correct. Threaded serving might improve the load on an otherwise unchanged Rack application. I see a 20% improve for simple Sinatra applications on my machine with Ruby 1.9.3, even more when running on Rubinius or JRuby, where all cores can be utilized. The second approach is useful if you write your application in an evented manner.

You can throw a lot of magic and hacks on top of Rack to have a non-evented application make use of those mechanisms (see em-synchrony or sinatra-synchrony), but that will leave you in debugging and dependency hell.

The async approach makes real sense with applications that tend to be best solved with an evented approach, like a web chat. However, I would not recommend using the threaded approach for implementing long-polling, because every polling connection will block a thread. This will leave you with either a ton of threads or connections you can't deal with. EM's thread pool has a size of 20 threads by default, limiting you to 20 waiting connections per process.

You could use a server that creates a new thread for every incoming connection, but creating threads is expensive (except on MacRuby, but I would not use MacRuby in any production app). Examples are serv and net-http-server. Ideally, what you want is an n:m mapping of requests and threads. But there's no server out there offering that.

If you want to learn more on the topic: I gave a presentation about this at Rocky Mountain Ruby (and a ton of other conferences). A video recording can be found on confreaks.

How to parse JSON request body in Sinatra just once and expose it to all routes?

Use a sinatra before handler:

before do
  request.body.rewind
  @request_payload = JSON.parse request.body.read
end

this will expose it to the current request handler. If you want it exposed to all handlers, put it in a superclass and extend that class in your handlers.

Does HTTP streaming in Rails 3.1 work with Thin?

In the release notes for the 0.5.4 release:

    == 0.5.4 Flying Mustard release
     * Don't read the full body, use direct streaming when sending response.
       See: Response#each
       As a result, the Content-Length can not be calculated anymore.
       You have to do set this in your adapter. All frameworks do it anyway.
       It improve memory usage and boost speed for low concurrency.
       Thanks to Kent Sibilev and Ezra for their help on that one.

However when I ran two thin instances (on different ports, one set as the asset host), the CSS and JS in the layout were not loaded before the sleep call in the controller action had finished executing. This was done in production mode.

I'm not 100% sure of the mechanics of it, perhaps the Thin rack-adapter needs to be setup to allow streaming.

---

Related Topics