Class: Datadog::CI::ITR::Coverage::Writer

Inherits:
Object
  • Object
show all
Includes:
Core::Workers::Polling, Core::Workers::Queue
Defined in:
lib/datadog/ci/itr/coverage/writer.rb

Constant Summary collapse

DEFAULT_BUFFER_MAX_SIZE = 
10_000
DEFAULT_SHUTDOWN_TIMEOUT = 
60

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(transport:, options: {}) ⇒ Writer

Returns a new instance of Writer.



25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
 
# File 'lib/datadog/ci/itr/coverage/writer.rb', line 25

def initialize(transport:, options: {})
  @transport = transport

  # Workers::Polling settings
  self.enabled = options.fetch(:enabled, true)

  # Workers::Async::Thread settings
  self.fork_policy = Core::Workers::Async::Thread::FORK_POLICY_RESTART

  # Workers::IntervalLoop settings
  self.loop_base_interval = options[:interval] if options.key?(:interval)
  self.loop_back_off_ratio = options[:back_off_ratio] if options.key?(:back_off_ratio)
  self.loop_back_off_max = options[:back_off_max] if options.key?(:back_off_max)

  @buffer_size = options.fetch(:buffer_size, DEFAULT_BUFFER_MAX_SIZE)

  self.buffer = buffer_klass.new(@buffer_size)

  @shutdown_timeout = options.fetch(:shutdown_timeout, DEFAULT_SHUTDOWN_TIMEOUT)

  @stopped = false
end

Instance Attribute Details

#transport ⇒ Object (readonly)

Returns the value of attribute transport.



20
21
22
 
# File 'lib/datadog/ci/itr/coverage/writer.rb', line 20

def transport
  @transport
end

Instance Method Details

#after_fork ⇒ Object 



90
91
92
93
94
95
 
# File 'lib/datadog/ci/itr/coverage/writer.rb', line 90

def after_fork
  # In multiprocess environments, forks will share the same buffer until its written to.
  # A.K.A. copy-on-write. We don't want forks to write events generated from another process.
  # Instead, we reset it after the fork. (Make sure any enqueue operations happen after this.)
  self.buffer = buffer_klass.new(@buffer_size)
end

#async? ⇒ Boolean

Returns:

  • (Boolean) 


86
87
88
 
# File 'lib/datadog/ci/itr/coverage/writer.rb', line 86

def async?
  true
end

#buffer_klass ⇒ Object 



97
98
99
100
101
102
103
 
# File 'lib/datadog/ci/itr/coverage/writer.rb', line 97

def buffer_klass
  if Core::Environment::Ext::RUBY_ENGINE == "ruby"
    Core::Buffer::CRuby
  else
    Core::Buffer::ThreadSafe
  end
end

#dequeue ⇒ Object 



78
79
80
 
# File 'lib/datadog/ci/itr/coverage/writer.rb', line 78

def dequeue
  buffer.pop
end

#enqueue(event) ⇒ Object 



74
75
76
 
# File 'lib/datadog/ci/itr/coverage/writer.rb', line 74

def enqueue(event)
  buffer.push(event)
end

#perform(*events) ⇒ Object 



58
59
60
61
62
63
64
 
# File 'lib/datadog/ci/itr/coverage/writer.rb', line 58

def perform(*events)
  responses = transport.send_events(events)

  loop_back_off! if responses.find(&:server_error?)

  nil
end

#stop(force_stop = false, timeout = @shutdown_timeout) ⇒ Object 



66
67
68
69
70
71
72
 
# File 'lib/datadog/ci/itr/coverage/writer.rb', line 66

def stop(force_stop = false, timeout = @shutdown_timeout)
  @stopped = true

  buffer.close if running?

  super
end

#work_pending? ⇒ Boolean

Returns:

  • (Boolean) 


82
83
84
 
# File 'lib/datadog/ci/itr/coverage/writer.rb', line 82

def work_pending?
  !buffer.empty?
end

#write(event) ⇒ Object 



48
49
50
51
52
53
54
55
56
 
# File 'lib/datadog/ci/itr/coverage/writer.rb', line 48

def write(event)
  return if @stopped

  # Start worker thread. If the process has forked, it will trigger #after_fork to
  # reconfigure the worker accordingly.
  perform

  enqueue(event)
end