Class: Sumologic::Metrics::Worker

Inherits:
Object
  • Object
show all
Includes:
Defaults, Utils
Defined in:
lib/sumologic/metrics/worker.rb

Instance Method Summary collapse

Methods included from Utils

stringify_keys, symbolize_keys, symbolize_keys!

Constructor Details

#initialize(queue, collector_uri, options = {}) ⇒ Worker

public: Creates a new worker

The worker continuously takes metrics off the queue and makes requests to the Sumologic api

queue - Queue synchronized between client and worker collector_uri - String of the unique collector URI options - Hash of worker options

batch_size - Fixnum of how many items to send in a batch
on_error   - Proc of what to do on an error



23
24
25
26
27
28
29
30
31
32
 
# File 'lib/sumologic/metrics/worker.rb', line 23

def initialize(queue, collector_uri, options = {})
  symbolize_keys! options
  @queue = queue
  @collector_uri = collector_uri
  @on_error = options[:on_error] || proc { |status, message| }
  batch_size = options[:batch_size] || Defaults::MetricBatch::MAX_SIZE
  @batch = MetricBatch.new(batch_size)
  @lock = Mutex.new
  @request = options[:request] || Request.new(uri: collector_uri)
end

Instance Method Details

#is_requesting? ⇒ Boolean

public: Check whether we have outstanding requests.

Returns:

  • (Boolean) 


55
56
57
 
# File 'lib/sumologic/metrics/worker.rb', line 55

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

#run ⇒ Object

public: Continuously runs the loop to check for new events



36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
 
# File 'lib/sumologic/metrics/worker.rb', line 36

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

    @lock.synchronize do
      until @batch.full? || @queue.empty?
        @batch << Metric.new(@queue.pop)
      end
    end

    res = @request.post(@batch)
    @on_error.call(res.status, res.message) unless res.status == 200

    @lock.synchronize { @batch.clear }
  end
end