Async Requests Using Sinatra Streaming API

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

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.

How do I log asynchronous thin+sinatra+rack requests?

I eventually found that using rack-async with async_sinatra was causing problems with 404 pages, exception handling, and logging:

!! Unexpected error while processing request: undefined method `bytesize' for nil:NilClass

Instead I used the following wrapper around aroute for logging:

module Sinatra::Async
    alias :oldaroute :aroute
    def aroute verb, path, opts = {}, &block
        # Based on aroute from async_sinatra

        run_method = :"RunA#{verb} #{path} #{opts.hash}"
        define_method run_method, &block

        log_method = :"LogA#{verb} #{path} #{opts.hash}"
        define_method(log_method) { |*a|
            puts "#{request.ip} - #{status} #{verb} #{path}"
        }

        oldaroute verb, path, opts do |*a|
            oldcb = request.env['async.callback']
            request.env['async.callback'] = proc { |*args|
                async_runner(log_method, *a)
                oldcb[*args]
            }
            async_runner(run_method, *a)
        end
    end
end

This is for the same versions of async_sinatra, Thin, and Rack that I was using when I asked this question last year; newer versions may allow the use of common Rack middleware for logging.

Streaming data from Sinatra/Rack application

Neither Webrick nor Thin support streaming that way. You could try Mongrel or Unicorn. If you want to use Thin or Rainbows!, you have to hook into the event loop in order to achieve streaming:

require 'sinatra'

class Stream
  include EventMachine::Deferrable
  def initialize
    @counter = 0
  end

  def each(&block)
    if @counter > 10
      succeed
    else
      EM.next_tick do
        yield counter
        each(&block)
      end
    end
  end
end

get '/' do
  Stream.new
end

I recently wrote a EventSource implementation that way:

require 'sinatra'

class EventStream
  include EventMachine::Deferrable
  def each
    count = 0
    timer = EventMachine::PeriodicTimer.new(1) do
      yield "data: #{count += 1}\n\n"
    end
    errback { timer.cancel }
  end
end

get '/' do
  EventMachine.next_tick do
    request.env['async.callback'].call [
      200, {'Content-Type' => 'text/event-stream'},
      EventStream.new ]
  end
  [-1, {}, []]
end

If you want to use Webrick for Streaming: here is a patch.

Sinatra + Fibers + EventMachine

get '/some/route/' do
  fib = Fiber.current
  req = EM::SomeNonBlokingLib.request
  req.callback do |response|
    fib.resume(response)
  end
  req.errback do |err|
    fib.resume(err)
  end
  Fiber.yield
end

EDIT

In your case you should spawn a Fiber for each request. So. Firstly create Rack config file and add some magick:

# config.ru
BOOT_PATH = File.expand_path('../http.rb',  __FILE__)
require BOOT_PATH

class FiberSpawn
  def initialize(app)
    @app = app
  end

  def call(env)
    fib = Fiber.new do
      res = @app.call(env)
      env['async.callback'].call(res)
    end
    EM.next_tick{ fib.resume }
    throw :async
  end
end

use FiberSpawn
run Http

Then your http Sinatra application:

# http.rb
require 'sinatra'
require 'fiber'

class Http < Sinatra::Base
  get '/' do
    f = Fiber.current
    EM.add_timer(1) do
      f.resume("Hello World")
    end
    Fiber.yield
  end
end

Now you could run it under thin for example:

> thin start -R config.ru

Then if you will visit locakhost:3000 you'll see your Hello World message

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

Sinatra using a websocket client to respond to a http request

The problem was that async-sinatra was using its own threads, and so was websocket-client-simple. The solution is to use bindings and the eval function, though this is not very efficient at all. I am hoping that optimizations or better solutions are available.

require 'sinatra'
require 'sinatra/async'
require 'websocket-client-simple'

set :server, 'thin'

register Sinatra::Async

aget '/' do
    puts 'Request received'

    socketResponse('wss:// ... URL ...', binding)
end

def socketResponse(url, b)
    ws = WebSocket::Client::Simple.connect(url)

    puts 'Connected to web socket'

    ws.on :message do |message|
        puts 'Got message:' + message.to_s
    end

    ws.on :close do
        puts 'WS closed'
        EM.schedule { b.eval " body 'Closed' " }
    end

    ws.on :open do
        puts 'WS open'

        message = 'A nice message to process'
        ws.send message
        puts 'Sent: ' + message
    end
end

---

Related Topics