Class: Kilden::EventQueue Private

Inherits:
Object
  • Object
show all
Defined in:
lib/kilden/event_queue.rb

Overview

This class is part of a private API. You should avoid using this class if possible, as it may be removed or be changed in the future.

Bounded in-memory queue (spec contract 7): at capacity the NEW event is dropped, never the old ones, and the drop is counted. Wakes the worker when flush_at is reached.

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(max_size:, flush_at:) ⇒ EventQueue

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Returns a new instance of EventQueue.



11
12
13
14
15
16
17
18
19
 
# File 'lib/kilden/event_queue.rb', line 11

def initialize(max_size:, flush_at:)
  @max_size = max_size
  @flush_at = flush_at
  @items = []
  @dropped_count = 0
  @mutex = Mutex.new
  @signal = ConditionVariable.new
  @closed = false
end

Instance Attribute Details

#dropped_count ⇒ Object (readonly)

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future. 



9
10
11
 
# File 'lib/kilden/event_queue.rb', line 9

def dropped_count
  @dropped_count
end

Instance Method Details

#close ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future. 



56
57
58
59
60
61
 
# File 'lib/kilden/event_queue.rb', line 56

def close
  @mutex.synchronize do
    @closed = true
    @signal.broadcast
  end
end

#closed? ⇒ Boolean

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Returns:

  • (Boolean) 


63
64
65
 
# File 'lib/kilden/event_queue.rb', line 63

def closed?
  @mutex.synchronize { @closed }
end

#drain ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Everything queued at the moment of the call (for flush/shutdown).



44
45
46
 
# File 'lib/kilden/event_queue.rb', line 44

def drain
  @mutex.synchronize { @items.slice!(0, @items.size) }
end

#empty? ⇒ Boolean

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Returns:

  • (Boolean) 


52
53
54
 
# File 'lib/kilden/event_queue.rb', line 52

def empty?
  size.zero?
end

#push(event) ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Returns false when the event was dropped because the queue is full.



22
23
24
25
26
27
28
29
30
31
32
 
# File 'lib/kilden/event_queue.rb', line 22

def push(event)
  @mutex.synchronize do
    if @items.size >= @max_size
      @dropped_count += 1
      return false
    end
    @items << event
    @signal.signal if @items.size >= @flush_at
    true
  end
end

#reset! ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Fork recovery (contract 9): the child discards the inherited queue — those events belong to the parent; sending them twice would duplicate.



69
70
71
72
73
74
75
76
 
# File 'lib/kilden/event_queue.rb', line 69

def reset!
  @mutex.synchronize do
    discarded = @items.size
    @items.clear
    @closed = false
    discarded
  end
end

#size ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future. 



48
49
50
 
# File 'lib/kilden/event_queue.rb', line 48

def size
  @mutex.synchronize { @items.size }
end

#wait_batch(interval, max: 1000) ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Blocks until flush_at is reached, interval elapses, or close; then pops up to max events. Returns [] on a quiet interval tick.



36
37
38
39
40
41
 
# File 'lib/kilden/event_queue.rb', line 36

def wait_batch(interval, max: 1000)
  @mutex.synchronize do
    @signal.wait(@mutex, interval) if @items.size < @flush_at && !@closed
    @items.shift(max)
  end
end