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.



21
22
23
 
# File 'lib/abmeter/async_submitter.rb', line 21

def api_client
  @api_client
end

.flush_interval ⇒ Object (readonly)

Returns the value of attribute flush_interval.



21
22
23
 
# File 'lib/abmeter/async_submitter.rb', line 21

def flush_interval
  @flush_interval
end

.logger ⇒ Object (readonly)

Returns the value of attribute logger.



21
22
23
 
# File 'lib/abmeter/async_submitter.rb', line 21

def logger
  @logger
end

.retry_queue ⇒ Object (readonly)

Returns the value of attribute retry_queue.



21
22
23
 
# File 'lib/abmeter/async_submitter.rb', line 21

def retry_queue
  @retry_queue
end

Class Method Details

.configure(api_client:, config:) ⇒ Object 



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

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

.flush ⇒ Object 



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

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 



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

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 



33
34
35
 
# File 'lib/abmeter/async_submitter.rb', line 33

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

.queue_size ⇒ Object 



117
118
119
 
# File 'lib/abmeter/async_submitter.rb', line 117

def queue_size
  @queue.size
end

.reset(timeout: DEFAULT_SHUTDOWN_TIMEOUT) ⇒ Object 



101
102
103
104
105
 
# File 'lib/abmeter/async_submitter.rb', line 101

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

.reset! ⇒ Object 



107
108
109
110
111
 
# File 'lib/abmeter/async_submitter.rb', line 107

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.



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

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.



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

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 



29
30
31
 
# File 'lib/abmeter/async_submitter.rb', line 29

def start
  start_worker
end

.worker_alive? ⇒ Boolean

Returns:

  • (Boolean) 


113
114
115
 
# File 'lib/abmeter/async_submitter.rb', line 113

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