Add jobs for asynchronous processing

Juggler.throw(:method, params)

Add handlers, with optional concurrency, inside an EM loop

EM.run {
  Juggler.juggle(:method, 10) do |deferrable, params|
    # Succeed the deferrable when the job is done
  end
}

For example

Juggler.juggle(:download, 10) do |df, params|
  http = EM::Protocols::HttpClient.request({
    :host => params[:host], 
    :port => 80, 
    :request => params[:path]
  })
  http.callback do |response|
    puts "Got response status #{response[:status]} for #{a}"
    df.success
  end
end

The job is considered to have failed in the following cases:

• If the block raises an error
• If the block fails the passed deferrable
• If the job timeout is exceeded. In this case the passed deferrable will be failed by juggler. If you need to clean up any state in this case (for example you might want to cancel a HTTP request) then you should bind to df.errback.

Stopping

Juggler doesn't automatically add signal handlers any more. A sensible default stop strategy is:

# Set the maximum time to wait for jobs to finish, defaults to 2s
Juggler.shutdown_grace_timeout = 10

%w{INT TERM}.each do |sig|
  Signal.trap(sig) {
    Juggler.stop
  }
end

For more control see Juggler::Runner.stop, or you can stop runners individually.

Beanstalkd connection handling

If the beanstalkd connection is dropped it will be re-established automatically by em-jack (even after the disconnect event). If the connection cannot be re-established after 5 (by default) attempts, a callback is fired. By default a warning is logged, but this can be changed

Juggler.on_disconnect {
  # Do something
}

Customising behaviour

Customising the back-off for failed jobs

By default, juggler will backoff jobs which failed exponentially using an exponent of 1.3, up to a maximum delay of 1 day, at which point the job will be buried. It's possible to customise this behaviour:

Juggler.backoff_function = lambda { |job_runner, job_stats|
  # job_stats is a hash with string keys, as returned by beanstalkd's
  # stats-job command. Particularly useful stats in this context are:
  #
  # job_stats["age"] - the time since the put command that created this job
  # job_stats["delay"] - the previous amount of time delayed

  new_delay = ([1, job_stats["delay"] * 2].max).ceil
  if job_stats["age"] > 300
    job_runner.delete
    # Or you could bury the job
    # job_runner.bury
  else
    job_runner.release(new_delay)
  end
}

Important points to note

• If your deferrable code raises errors, this will not be handled by juggler.