Class: Checkend::Worker

Inherits:
Object
  • Object
show all
Defined in:
lib/checkend/worker.rb

Overview

Worker handles async sending of notices via a background thread.

It maintains a queue of notices and sends them in the background, implementing throttling on errors and graceful shutdown.

Constant Summary collapse

SHUTDOWN = 
Object.new.freeze
FLUSH = 
Object.new.freeze
BASE_THROTTLE =

Exponential backoff base for throttling

1.05
MAX_THROTTLE = 
100

Instance Method Summary collapse

Constructor Details

#initialize(config) ⇒ Worker

Returns a new instance of Worker.



17
18
19
20
21
22
23
24
25
 
# File 'lib/checkend/worker.rb', line 17

def initialize(config)
  @config = config
  @queue = Queue.new
  @mutex = Mutex.new
  @shutdown = false
  @throttle = 0
  @client = Client.new(config)
  @thread = start_thread
end

Instance Method Details

#flush(timeout: nil) ⇒ Object

Flush the queue, blocking until all current notices are sent

Parameters:

  • timeout (Integer) (defaults to: nil) —

    seconds to wait



58
59
60
61
62
63
64
65
66
 
# File 'lib/checkend/worker.rb', line 58

def flush(timeout: nil)
  timeout ||= @config.timeout

  cv = ConditionVariable.new
  @mutex.synchronize do
    @queue.push(cv)
    cv.wait(@mutex, timeout)
  end
end

#push(notice) ⇒ Boolean

Push a notice onto the queue for async sending

Parameters:

  • notice (Notice) —

    the notice to send

Returns:

  • (Boolean) —

    true if queued, false if rejected



31
32
33
34
35
36
37
 
# File 'lib/checkend/worker.rb', line 31

def push(notice)
  return false if @shutdown
  return false if @queue.size >= @config.max_queue_size

  @queue.push(notice)
  true
end

#queue_size ⇒ Integer

Get the current queue size

Returns:

  • (Integer) 


78
79
80
 
# File 'lib/checkend/worker.rb', line 78

def queue_size
  @queue.size
end

#running? ⇒ Boolean

Check if the worker is running

Returns:

  • (Boolean) 


71
72
73
 
# File 'lib/checkend/worker.rb', line 71

def running?
  @thread&.alive? && !@shutdown
end

#shutdown(timeout: nil) ⇒ Object

Shutdown the worker, waiting for pending notices

Parameters:

  • timeout (Integer) (defaults to: nil) —

    seconds to wait (default from config)



42
43
44
45
46
47
48
49
50
51
52
53
 
# File 'lib/checkend/worker.rb', line 42

def shutdown(timeout: nil)
  timeout ||= @config.shutdown_timeout

  @mutex.synchronize do
    return if @shutdown

    @shutdown = true
    @queue.push(SHUTDOWN)
  end

  @thread&.join(timeout)
end