Class: Snare::Queue

Inherits:
Object
  • Object
show all
Defined in:
lib/snare/queue.rb

Overview

A small, dependency-free, thread-safe batching queue — mirrors packages/sdk/src/queue.ts. Flushes on flush_size events queued, or flush_interval seconds elapsed since the OLDEST queued event (the timer is armed once, when the queue transitions from empty to non-empty, and cancelled whenever the queue empties out).

Unlike the JS SDK's single-threaded browser environment, Ruby apps (Rails, Sidekiq workers) are commonly multi-threaded, so every mutation of the internal array is guarded by a Mutex.

Instance Method Summary collapse

Constructor Details

#initialize(flush_size:, flush_interval:, &on_flush) ⇒ Queue

Returns a new instance of Queue.



14
15
16
17
18
19
20
21
 
# File 'lib/snare/queue.rb', line 14

def initialize(flush_size:, flush_interval:, &on_flush)
  @flush_size = flush_size
  @flush_interval = flush_interval
  @on_flush = on_flush
  @mutex = Mutex.new
  @queue = []
  @timer_thread = nil
end

Instance Method Details

#drain ⇒ Object

Clears the queue and cancels the pending timer WITHOUT calling on_flush — for callers that want to deliver the drained events themselves via a different path.



43
44
45
 
# File 'lib/snare/queue.rb', line 43

def drain
  @mutex.synchronize { take_and_clear }
end

#enqueue(event) ⇒ Object 



23
24
25
26
27
28
29
30
31
 
# File 'lib/snare/queue.rb', line 23

def enqueue(event)
  batch = nil
  @mutex.synchronize do
    @queue << event
    start_timer if @queue.length == 1
    batch = take_and_clear if @queue.length >= @flush_size
  end
  @on_flush.call(batch) if batch
end

#flush ⇒ Object

Forces a flush now (calls on_flush) if anything is queued. A no-op on an empty queue.



35
36
37
38
 
# File 'lib/snare/queue.rb', line 35

def flush
  batch = @mutex.synchronize { take_and_clear if @queue.any? }
  @on_flush.call(batch) if batch
end