|
3 | 3 | require 'set' |
4 | 4 |
|
5 | 5 | module Flagsmith |
| 6 | + # Started event processors, stopped together at exit. A stopped processor leaves, so it is not kept in memory. |
| 7 | + module RunningEventProcessors |
| 8 | + @processors = Set.new |
| 9 | + @mutex = Mutex.new |
| 10 | + at_exit { to_a.each(&:stop) } |
| 11 | + |
| 12 | + def self.add(processor) |
| 13 | + @mutex.synchronize { @processors.add(processor) } |
| 14 | + end |
| 15 | + |
| 16 | + def self.delete(processor) |
| 17 | + @mutex.synchronize { @processors.delete(processor) } |
| 18 | + end |
| 19 | + |
| 20 | + def self.to_a |
| 21 | + @mutex.synchronize { @processors.to_a } |
| 22 | + end |
| 23 | + end |
| 24 | + |
6 | 25 | # Buffers tracking and exposure events, and sends them to the Flagsmith events API in batches. |
7 | 26 | # |
8 | 27 | # Events are sent every +flush_interval_seconds+, as soon as +max_buffer_items+ are buffered, |
@@ -58,12 +77,13 @@ def start |
58 | 77 | return if @interval |
59 | 78 |
|
60 | 79 | @interval = set_interval(@flush_interval_seconds) { flush } |
61 | | - at_exit { stop } |
| 80 | + RunningEventProcessors.add(self) |
62 | 81 | end |
63 | 82 |
|
64 | 83 | def stop |
65 | 84 | @flush_mutex.synchronize { clear_interval(@interval) } if @interval # never interrupts a batch being sent |
66 | 85 | @interval = nil |
| 86 | + RunningEventProcessors.delete(self) |
67 | 87 | flush |
68 | 88 | end |
69 | 89 |
|
@@ -122,7 +142,7 @@ def duplicate_exposure?(event) |
122 | 142 | return false unless event[:event] == FLAG_EXPOSURE_EVENT |
123 | 143 | return false if @buffered_exposures.add?(event.values_at(:feature_name, :identifier, :value)) |
124 | 144 |
|
125 | | - @logger.debug "Skipping duplicate #{FLAG_EXPOSURE_EVENT} event for feature #{event[:feature_name]}, identifier #{event[:identifier]}" |
| 145 | + @logger.debug "Skipping duplicate #{FLAG_EXPOSURE_EVENT} event for feature #{event[:feature_name]}" |
126 | 146 | true |
127 | 147 | end |
128 | 148 |
|
|
0 commit comments