Class: Beanstalker::Queue

Inherits:
Object
  • Object
show all
Defined in:
lib/beanstalker/queue.rb

Constant Summary collapse

DEFAULT_PRI = 
512
DEFAULT_FUZZ = 
0
DEFAULT_DELAY = 
0
DEFAULT_TTR = 
120
DEFAULT_TUBE = 
'default'
SUBMIT_OPTS = 
[:pri, :fuzz, :delay, :ttr, :tube]

Class Attribute Summary collapse

Class Method Summary collapse

Class Attribute Details

.after_put ⇒ Object

Returns the value of attribute after_put.



32
33
34
 
# File 'lib/beanstalker/queue.rb', line 32

def after_put
  @after_put
end

.queue ⇒ Object

Returns the value of attribute queue.



32
33
34
 
# File 'lib/beanstalker/queue.rb', line 32

def queue
  @queue
end

Class Method Details

.gen(obj, selector, args) ⇒ Object

Be careful not to pass in a selector that’s not valid ruby source.



94
95
96
 
# File 'lib/beanstalker/queue.rb', line 94

def gen(obj, selector, args)
  obj.rrepr + '.' + selector.to_s + '(' + gen_args(args) + ')'
end

.gen_args(args) ⇒ Object 



98
99
100
 
# File 'lib/beanstalker/queue.rb', line 98

def gen_args(args)
  args.rrepr[1...-1]
end

.pkg(code, opts, obj, sel) ⇒ Object 



84
85
86
87
88
89
90
91
 
# File 'lib/beanstalker/queue.rb', line 84

def pkg(code, opts, obj, sel)
  opts.merge(:kind => 'rails_beanstalker',
             :data => {
               :code => code,
               :class => obj.is_a?(Class) ? obj.name : obj.class.to_s,
               :method => sel.to_s
             })
end

.put!(obj, pri = DEFAULT_PRI, delay = DEFAULT_DELAY, ttr = DEFAULT_TTR, tube = DEFAULT_TUBE) ⇒ Object 



49
50
51
52
53
54
55
56
57
58
 
# File 'lib/beanstalker/queue.rb', line 49

def put!(obj, pri=DEFAULT_PRI, delay=DEFAULT_DELAY, ttr=DEFAULT_TTR,
         tube=DEFAULT_TUBE)
  return sync_run(obj) if pri == :direct || !queue
  queue.connect()
  queue.use(tube)
  info = [queue.yput(obj, pri, delay, ttr), queue.last_server]
  f = Beanstalker::Queue.after_put
  f.call(*info) if f
  return info
end

.put_call!(obj, sel, opts, args = []) ⇒ Object 



62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
 
# File 'lib/beanstalker/queue.rb', line 62

def put_call!(obj, sel, opts, args=[])
  pri = opts.fetch(:pri, DEFAULT_PRI)
  fuzz = opts.fetch(:fuzz, DEFAULT_FUZZ)
  delay = opts.fetch(:delay, DEFAULT_DELAY)
  ttr = opts.fetch(:ttr, DEFAULT_TTR)
  tube = opts.fetch(:tube, DEFAULT_TUBE)
  worker_opts = opts.reject{|k,v| SUBMIT_OPTS.include?(k)}
  interpolator = opts.fetch(:interpolator, nil)

  pri = pri + rand(fuzz + 1) if pri != :direct

  if interpolator
    code = packed = interpolator
  else
    code = gen(obj, sel, args)
    packed = pkg(code, worker_opts, obj, sel)
  end

  RAILS_DEFAULT_LOGGER.info("put #{pri} #{code} to #{tube} with ttr #{ttr}")
  put!(packed, pri, delay, ttr, tube)
end

.sync_run(obj) ⇒ Object

This runs jobs synchronously; it’s used when no queue is configured.



41
42
43
44
45
46
47
 
# File 'lib/beanstalker/queue.rb', line 41

def sync_run(obj)
  body = YAML.dump(obj)
  job = Beanstalk::Job.new(Beanstalker::FakeConn.new(), 0, body)
  sync_worker.dispatch(job)
  sync_worker.do_all_work()
  return 0, '0.0.0.0'
end

.sync_worker ⇒ Object

This is a fake worker instance for running jobs synchronously.



35
36
37
38
 
# File 'lib/beanstalker/queue.rb', line 35

def sync_worker()
  require 'beanstalker/worker'
  @sync_worker ||= Beanstalker::Worker.new(binding)
end