Class: DRbQueue::Server

Inherits:
Object
  • Object
show all
Defined in:
lib/drb_queue/server.rb

Defined Under Namespace

Classes: UnableToStart, Work

Constant Summary collapse

NotStarted = 
Class.new(StandardError)
AlreadyStarted = 
Class.new(StandardError)

Instance Method Summary collapse

Constructor Details

#initialize(configuration) ⇒ Server

Returns a new instance of Server.



34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
 
# File 'lib/drb_queue/server.rb', line 34

def initialize(configuration)
  [:logger, :error_handler, :immediate].each do |p|
    __send__("#{p}=", configuration.__send__(p))
  end

  self.queue = Queue.new
  self.running = true

  self.store = configuration.construct_persistence_store

  start_workers(configuration.num_workers)

  if store
    store.each_persisted_work do |serialized_work|
      enqueue_work(Work.unserialize(serialized_work))
    end
  end
end

Instance Method Details

#enqueue(worker, *args) ⇒ Object 



53
54
55
56
57
 
# File 'lib/drb_queue/server.rb', line 53

def enqueue(worker, *args)
  uuid.generate.tap do |id|
    enqueue_work(Work.new(worker, args))
  end
end

#ping ⇒ Object 



63
64
65
 
# File 'lib/drb_queue/server.rb', line 63

def ping
  'pong'
end

#shutdown! ⇒ Object 



67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
 
# File 'lib/drb_queue/server.rb', line 67

def shutdown!
  self.running = false

  if store
    begin
      while work = queue.pop(:dont_block)
        store.persist(work)
      end
    rescue ThreadError => e
    rescue => e
      error_handler.call(e)
    end
  elsif queue.size > 0
    logger.error("Queue is non-empty and we're shutting down...probably better to configure a persistence store\n")
  end

  workers.each(&:join)
end

#uuid ⇒ Object 



59
60
61
 
# File 'lib/drb_queue/server.rb', line 59

def uuid
  @uuid ||= UUID.new
end