class SplitIoClient::SSE::Workers::SegmentsWorker
def add_to_queue(change_number, segment_name)
def add_to_queue(change_number, segment_name) item = { change_number: change_number, segment_name: segment_name } @config.logger.debug("SegmentsWorker add to queue #{item}") @queue.push(item) end
def initialize(synchronizer, config, segments_repository)
def initialize(synchronizer, config, segments_repository) @synchronizer = synchronizer @config = config @segments_repository = segments_repository @queue = Queue.new @running = Concurrent::AtomicBoolean.new(false) end
def perform
def perform while (item = @queue.pop) segment_name = item[:segment_name] cn = item[:change_number] @config.logger.debug("SegmentsWorker change_number dequeue #{segment_name}, #{cn}") @synchronizer.fetch_segment(segment_name, cn) end end
def perform_thread
def perform_thread @config.threads[:segment_update_worker] = Thread.new do @config.logger.debug('Starting segments worker ...') if @config.debug_enabled perform end end
def start
def start if @running.value @config.logger.debug('segments worker already running.') return end @running.make_true perform_thread end
def stop
def stop unless @running.value @config.logger.debug('segments worker not running.') return end @running.make_false SplitIoClient::Helpers::ThreadHelper.stop(:segment_update_worker, @config) end