Class: Sentiero::Reporter::Dispatcher

Inherits:
Object
  • Object
show all
Defined in:
lib/sentiero/reporter/dispatcher.rb

Overview

Delivers payloads to a transport, synchronously or via a bounded background queue. Never raises into the caller; when the async queue is full new payloads are dropped rather than blocking the host app.

Constant Summary collapse

FlushLatch =

A latch pushed through the work queue by #flush; recognized by type.

Thread::Queue

Instance Method Summary collapse

Constructor Details

#initialize(transport, async:, max_queue:) ⇒ Dispatcher

Returns a new instance of Dispatcher.



19
20
21
22
23
24
25
26
27
28
29
 
# File 'lib/sentiero/reporter/dispatcher.rb', line 19

def initialize(transport, async:, max_queue:)
  @transport = transport
  @async = async
  @dropped = Concurrent::AtomicFixnum.new(0) # incremented from concurrent enqueue callers
  @rejection_warned = false
  return unless @async

  @queue = SizedQueue.new(max_queue)
  @thread = Thread.new { run }
  @thread.name = "sentiero-reporter" if @thread.respond_to?(:name=)
end

Instance Method Details

#dropped ⇒ Object 



15
16
17
 
# File 'lib/sentiero/reporter/dispatcher.rb', line 15

def dropped
  @dropped.value
end

#enqueue(path, payload) ⇒ Object 



31
32
33
34
35
36
37
38
39
40
41
42
 
# File 'lib/sentiero/reporter/dispatcher.rb', line 31

def enqueue(path, payload)
  if @async
    begin
      @queue.push([path, payload], true) # non-block so ThreadError is raised if queue full
    rescue ThreadError
      @dropped.increment
    end
  else
    deliver([path, payload])
  end
  nil
end

#flush ⇒ Object

Blocks until every payload enqueued before this call has been delivered. The latch rides the FIFO queue, so reaching it means all prior jobs are done.



46
47
48
49
50
51
52
 
# File 'lib/sentiero/reporter/dispatcher.rb', line 46

def flush
  return unless @async
  latch = FlushLatch.new
  @queue.push(latch)
  latch.pop
  nil
end

#shutdown ⇒ Object 



54
55
56
57
58
 
# File 'lib/sentiero/reporter/dispatcher.rb', line 54

def shutdown
  return unless @async
  @queue.push(:stop)
  @thread.join(2)
end