Class: ABMeter::AsyncSubmitter

Inherits:
Object
  • Object
show all
Defined in:
lib/abmeter/async_submitter.rb

Constant Summary collapse

BATCH_SIZE =

Private internal constants for async submitter behavior

100
MAX_SUBMIT_ATTEMPTS = 
3
MAX_RETRY_QUEUE_SIZE = 
1000
DEFAULT_SHUTDOWN_TIMEOUT =

seconds — max time a graceful shutdown blocks before killing the worker

5.0

Class Attribute Summary collapse

Class Method Summary collapse

Class Attribute Details

.api_client ⇒ Object (readonly)

Returns the value of attribute api_client.



23
24
25
 
# File 'lib/abmeter/async_submitter.rb', line 23

def api_client
  @api_client
end

.flush_interval ⇒ Object (readonly)

Returns the value of attribute flush_interval.



23
24
25
 
# File 'lib/abmeter/async_submitter.rb', line 23

def flush_interval
  @flush_interval
end

.logger ⇒ Object (readonly)

Returns the value of attribute logger.



23
24
25
 
# File 'lib/abmeter/async_submitter.rb', line 23

def logger
  @logger
end

.retry_queue ⇒ Object (readonly)

Returns the value of attribute retry_queue.



23
24
25
 
# File 'lib/abmeter/async_submitter.rb', line 23

def retry_queue
  @retry_queue
end

Class Method Details

.configure(api_client:, config:) ⇒ Object 



25
26
27
28
29
 
# File 'lib/abmeter/async_submitter.rb', line 25

def configure(api_client:, config:)
  @api_client = api_client
  @flush_interval = config.flush_interval || DEFAULT_FLUSH_INTERVAL
  @logger = config.logger
end

.flush ⇒ Object 



51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
 
# File 'lib/abmeter/async_submitter.rb', line 51

def flush
  items = []

  @mutex.synchronize do
    # First, try to process retry queue
    process_retry_queue

    # Then process new items
    items << @queue.pop while !@queue.empty? && items.size < BATCH_SIZE
  end

  # Group by type and submit
  unless items.empty?
    exposures = items.select { |i| i[:type] == :exposure }.map { |i| i[:data] }
    events = items.select { |i| i[:type] == :event }.map { |i| i[:data] }

    submit_exposures(exposures) unless exposures.empty?
    submit_events(events) unless events.empty?
  end
end

.queue_event(event_slug, user_id, custom_fields) ⇒ Object 



39
40
41
42
43
44
45
46
47
48
49
 
# File 'lib/abmeter/async_submitter.rb', line 39

def queue_event(event_slug, user_id, custom_fields)
  @queue.push({
                type: :event,
                data: {
                  event_slug: event_slug,
                  user_id: user_id,
                  occurred_at: Time.now.iso8601,
                  custom_fields: custom_fields
                }
              })
end

.queue_exposure(exposure) ⇒ Object 



35
36
37
 
# File 'lib/abmeter/async_submitter.rb', line 35

def queue_exposure(exposure)
  @queue.push({ type: :exposure, data: exposure })
end

.queue_size ⇒ Object 



119
120
121
 
# File 'lib/abmeter/async_submitter.rb', line 119

def queue_size
  @queue.size
end

.reset(timeout: DEFAULT_SHUTDOWN_TIMEOUT) ⇒ Object 



103
104
105
106
107
 
# File 'lib/abmeter/async_submitter.rb', line 103

def reset(timeout: DEFAULT_SHUTDOWN_TIMEOUT)
  result = shutdown(timeout: timeout)
  clear_state
  result
end

.reset! ⇒ Object 



109
110
111
112
113
 
# File 'lib/abmeter/async_submitter.rb', line 109

def reset!
  shutdown!
  clear_state
  true
end

.shutdown(timeout: DEFAULT_SHUTDOWN_TIMEOUT) ⇒ Object

Graceful, lossless, bounded blocking. Signals the worker to stop, which triggers a final flush of the main and retry queues, then waits up to timeout seconds for that flush to finish and the worker to exit. Kills the worker (dropping anything still unflushed) if the timeout elapses. Returns true on clean shutdown, false on timeout.



77
78
79
80
81
82
83
84
85
86
87
88
89
 
# File 'lib/abmeter/async_submitter.rb', line 77

def shutdown(timeout: DEFAULT_SHUTDOWN_TIMEOUT)
  worker = @worker_thread
  return true unless worker

  request_stop
  joined = worker.join(timeout)
  if joined.nil?
    worker.kill
    worker.join
    log_error("Graceful shutdown timed out after #{timeout}s, killing worker (queued: #{@queue.size}, retry: #{@retry_queue.size})")
  end
  !joined.nil?
end

.shutdown! ⇒ Object

Immediate, lossy, non-blocking. Kills the worker and discards queued items without any network I/O.



93
94
95
96
97
98
99
100
101
 
# File 'lib/abmeter/async_submitter.rb', line 93

def shutdown!
  @worker_thread&.kill
  dropped_queue = @queue.size
  dropped_retry = @retry_queue.size
  if dropped_queue.positive? || dropped_retry.positive?
    log_error("Immediate shutdown, dropping items (dropped_queue: #{dropped_queue}, dropped_retry: #{dropped_retry})")
  end
  true
end

.start ⇒ Object 



31
32
33
 
# File 'lib/abmeter/async_submitter.rb', line 31

def start
  start_worker
end

.worker_alive? ⇒ Boolean

Returns:

  • (Boolean) 


115
116
117
 
# File 'lib/abmeter/async_submitter.rb', line 115

def worker_alive?
  @worker_thread&.alive? || false
end