Class: Rox::Core::Analytics::Worker

Inherits:
Object
  • Object
show all
Includes:
Defaults, Logging, Utils
Defined in:
lib/rox/core/analytics/worker.rb

Constant Summary

Constants included from Utils

Utils::UTC_OFFSET_WITHOUT_COLON, Utils::UTC_OFFSET_WITH_COLON

Instance Method Summary collapse

Methods included from Logging

included, #logger

Methods included from Utils

#date_in_iso8601, #datetime_in_iso8601, #formatted_offset, #isoify_dates, #isoify_dates!, #seconds_to_utc_offset, #stringify_keys, #symbolize_keys, #symbolize_keys!, #time_in_iso8601, #uid

Constructor Details

#initialize(queue, device_properties) ⇒ Worker

public: Creates a new worker

The worker continuously takes messages off the queue and makes requests to the segment.io api

queue - Queue synchronized between client and worker

Parameters:



22
23
24
25
26
27
28
29
30
 
# File 'lib/rox/core/analytics/worker.rb', line 22

def initialize(queue, device_properties)
  @queue = queue
  @device_properties = device_properties
  @on_error = proc { |status, error| }
  batch_size = Defaults::MessageBatch::MAX_SIZE
  @batch = MessageBatch.new(batch_size)
  @lock = Mutex.new
  @transport = Transport.new(device_properties)
end

Instance Method Details

#is_requesting? ⇒ Boolean

public: Check whether we have outstanding requests.

Returns:

  • (Boolean) 


53
54
55
 
# File 'lib/rox/core/analytics/worker.rb', line 53

def is_requesting?
  @lock.synchronize { !@batch.empty? }
end

#run ⇒ Object

public: Continuously runs the loop to check for new events



34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
 
# File 'lib/rox/core/analytics/worker.rb', line 34

def run
  until Thread.current[:should_exit]
    return if @queue.empty?

    @lock.synchronize do
      consume_message_from_queue! until @batch.full? || @queue.empty?
    end

    res = @transport.send @batch
    @on_error.call(res.status, res.error) unless res.status == 200

    @lock.synchronize { @batch.clear }
  end
ensure
  @transport.shutdown
end