Sha256: 7f7b514164d3bb5be4117dfa1b5277c4d2f8c6212686855c57f45258b599b359
Contents?: true
Size: 1.34 KB
Versions: 17
Compression:
Stored size: 1.34 KB
Contents
#!/usr/bin/env ruby
$: << File.expand_path(File.join(File.dirname(__FILE__), '..', '..'))
require 'beanstalk-client'
require 'flapjack/transports/result'
module Flapjack
module Transport
class Beanstalkd
def initialize(options={})
@options = options
@config = OpenStruct.new(options)
@log = @config.log
unless @config.host && @config.port && @config.queue_name
raise ArgumentError, "You need to specify a beanstalkd host, port, and queue name to connect to."
end
connect
end
def connect
begin
@queue = Beanstalk::Pool.new(["#{@config.host}:#{@config.port}"], @config.queue_name)
rescue Beanstalk::NotConnected => e
@log.error("Couldn't connect to the '#{@config.queue_name}' Beanstalk queue. Retrying in 5 seconds.")
sleep 5
retry
end
end
def next
begin
job = @queue.reserve # blocks
result = YAML::load(job.body)
Flapjack::Transport::Result.new(:job => job, :result => result)
rescue Beanstalk::NotConnected
@log.error("The '#{@config.queue_name}' Beanstalk queue went away. Waiting 5 seconds, then retrying.")
sleep 5
retry
end
end
def delete(result)
result.job.delete
end
end
end
end
Version data entries
17 entries across 17 versions & 1 rubygems