Class: Trifle::Stats::Buffer

Inherits:

Object

• Object
• Trifle::Stats::Buffer

Defined in:

lib/trifle/stats/buffer.rb

Overview

rubocop:disable Metrics/ClassLength

Constant Summary

DEFAULT_DURATION =

1

DEFAULT_SIZE =

256

Class Method Summary

• .flush_all ⇒ Object
• .pending_flushes? ⇒ Boolean
• .register(buffer) ⇒ Object
• .run_pending! ⇒ Object
• .unregister(buffer) ⇒ Object

Instance Method Summary

• #flush! ⇒ Object
• #flush_from_registry! ⇒ Object
• #inc(keys:, values:, tracking_key: nil) ⇒ Object
• #initialize(driver:, duration: DEFAULT_DURATION, size: DEFAULT_SIZE, aggregate: true, async: true) ⇒ Buffer constructor

  rubocop:disable Metrics/MethodLength.

• #set(keys:, values:, tracking_key: nil) ⇒ Object
• #shutdown! ⇒ Object

Constructor Details

#initialize(driver:, duration: DEFAULT_DURATION, size: DEFAULT_SIZE, aggregate: true, async: true) ⇒ Buffer

rubocop:disable Metrics/MethodLength

# File 'lib/trifle/stats/buffer.rb', line 235

def initialize(driver:, duration: DEFAULT_DURATION, size: DEFAULT_SIZE, aggregate: true, async: true) # rubocop:disable Metrics/MethodLength
  @driver = driver
  @duration = duration.to_f
  @size = size.to_i.positive? ? size.to_i : 1
  @async = async
  @queue = BufferQueue.new(aggregate: aggregate)
  @mutex = Mutex.new
  @stopped = false
  @flush_pending = false
  @pending_condition = ConditionVariable.new
  @worker = start_worker if async && @duration.positive?
  self.class.register(self)
end

Class Method Details

.flush_all ⇒ Object

# File 'lib/trifle/stats/buffer.rb', line 222

def flush_all
  BufferRegistry.flush_all
end

.pending_flushes? ⇒ Boolean

Returns:

• (Boolean)

# File 'lib/trifle/stats/buffer.rb', line 230

def pending_flushes?
  BufferRegistry.pending?
end

.register(buffer) ⇒ Object

# File 'lib/trifle/stats/buffer.rb', line 214

def register(buffer)
  BufferRegistry.register(buffer)
end

.run_pending! ⇒ Object

# File 'lib/trifle/stats/buffer.rb', line 226

def run_pending!
  BufferRegistry.run_pending!
end

.unregister(buffer) ⇒ Object

# File 'lib/trifle/stats/buffer.rb', line 218

def unregister(buffer)
  BufferRegistry.unregister(buffer)
end

Instance Method Details

#flush! ⇒ Object

# File 'lib/trifle/stats/buffer.rb', line 257

def flush!
  actions = drain_actions(reset_pending: true)
  return if actions.nil?

  process(actions)
end

#flush_from_registry! ⇒ Object

# File 'lib/trifle/stats/buffer.rb', line 274

def flush_from_registry!
  actions = drain_pending_actions
  process(actions) if actions
end

#inc(keys:, values:, tracking_key: nil) ⇒ Object

# File 'lib/trifle/stats/buffer.rb', line 249

def inc(keys:, values:, tracking_key: nil)
  enqueue(:inc, keys: keys, values: values, tracking_key: tracking_key)
end

#set(keys:, values:, tracking_key: nil) ⇒ Object

# File 'lib/trifle/stats/buffer.rb', line 253

def set(keys:, values:, tracking_key: nil)
  enqueue(:set, keys: keys, values: values, tracking_key: tracking_key)
end

#shutdown! ⇒ Object

# File 'lib/trifle/stats/buffer.rb', line 264

def shutdown!
  return if @shutdown

  @shutdown = true
  stop_worker
  BufferRegistry.cancel_pending(self)
  flush!
  self.class.unregister(self)
end