em deferrable faraday adapter

em_deferrable.rb

module Faraday
  class Adapter
    class EMHttpRequest < Faraday::Adapter
      self.supports_parallel_requests = true

      def self.setup_parallel_manager(options = {})
        EMParallelManager.new
      end

      class EMParallelManager
        def run; true; end
      end

      begin
        require 'em'
        require 'em-http'
      rescue LoadError, NameError => e
        self.load_error = e
      end

      class Header
        include Net::HTTPHeader
        def initialize response
          @header = {}
          response.response_header.each do |key, value|
            case key
            when "CONTENT_TYPE"; self.content_type = value
            when "CONTENT_LENGTH"; self.content_length = value
            else; self[key] = value
            end
          end
        end
      end

      def call(env)
        super

        http_async = EventMachine::HttpRequest.new(URI::parse(env[:url].to_s))

        options = setup_request_options(env)

        http = http_async.setup_request(env[:method].to_s.downcase.to_sym, options)

        http.callback do
          env.update({
            :body   => http.response,
            :status => http.response_header.status.to_i,
            :response_headers => Header.new(http)
          })
          env[:response].finish(env)
        end

        http.errback do
          raise Error::ConnectionFailed.new(Errno::ECONNREFUSED)
        end

        @app.call(env)
      end

      def setup_request_options(env)
        options = { :head => env[:request_headers] }

        options[:ssl] = env[:ssl] if env[:ssl]

        if env[:body]
          if env[:body].respond_to? :read
            options[:body] = env[:body].read
          else
            options[:body] = env[:body]
          end
        end

        if req = env[:request]
          if proxy = req[:proxy]
            uri = Addressable::URI.parse(proxy[:uri])
            options[:proxy] = {
              :host => uri.host,
              :port => uri.port
            }
            if proxy[:username] && proxy[:password]
              options[:proxy][:authorization] = [proxy[:username], proxy[:password]]
            end
          end

          # only one timeout currently supported by em http request
          if req[:timeout] or req[:open_timeout]
            options[:timeout] = [req[:timeout] || 0, req[:open_timeout] || 0].max
          end
        end

        options
      end

    end
  end
end

test_run.rb

require 'eventmachine'
require 'em-http'

require 'faraday'
require 'faraday/adapter/em_http_request'

parallel_manager = Faraday::Adapter::EMHttpRequest.setup_parallel_manager

c = Faraday.new(:url => 'http://google.com', :parallel => parallel_manager) do |builder|
  builder.adapter :EMHttpRequest
end

EM.run do
  c.get('/')
  puts 'wait for it all to run'
  EM.add_timer(3) { EM.stop }
end

Hmm. Haven't had a chance to actually run the code, but looking at your example.. try taking out sleep(3) and EM.stop - I don't think that will do what you want it to do. Namely, sleep(3) will block the reactor for 3 seconds, and immediately thereafter exit. Without the EM.stop your reactor will run forever.. but your requests should fire and finish. To exit cleanly, you should in fact be specifying a callback within the request to indicate that the reactor should be stopped once the request is complete.

Hey Ilya,

Ahhhh, once I remove the sleep and EM.stop it runs the queries I want. So there is no way to 'sleep' for x many seconds and let other em fibers/code finish executing? My next step is to benchmark this against other Faraday adapters, looking at both the fire and forget time and the full execution time.

Thanks for all the help!

Josh

If you want to stop the reactor after a certain amount of time passes, the way to do that is: EM.add_timer(3) { EM.stop }

In practice though, you still want to add some logic to make sure everyone finishes their requests before you terminate the reactor. Take a look at Multi that ships with em-http, its designed to do exactly this.. It'll dispatch all the requests in parallel, but you can then define a callback which will fire once all requests are done.

Hey Ilya,

Thanks for all the help, the above script works now, and it does the fire and forget strategy I was aiming for, mimicking what the Pusher gem does in its trigger_async method (or is it async_trigger).

I'll have a look over Multi, thanks for the tip.

If you think my adapter is, well, stupid or poorly thought out, any and all advice, comments and criticism is much appreciated.

No, no.. Given what you're going for (fire and forget), I think it makes sense.

My only worry is that if at some point you will want to synchronize to see what happens to all the requests (or wait for them to finish), then I think you'll have to introduce some other deferrable into the mix (which is multi is all about - and its very simple). Take a look at that code and you'll see what I mean.

Glad you got it working!