Module: EndPointBlank::Writers::DelayedWriter

Included in:
ExceptionWriter, LogWriter, RequestWriter, ResponseWriter
Defined in:
lib/end_point_blank/writers/delayed_writer.rb

Overview

Drains a background send queue with a fixed pool of worker threads.

The queue is bounded (see MAX_QUEUE_SIZE) so that during an intake outage - when the worker threads can't drain as fast as the host application enqueues payloads - memory usage stays capped instead of growing without bound toward an OOM. When the queue is full, the oldest item is dropped to make room for the newest one, and a warning is logged, throttled so a sustained outage does not itself become a logging flood.

Constant Summary collapse

MAX_QUEUE_SIZE = 
1000
WARN_THROTTLE_SECONDS = 
30
DEFAULT_WORKER_COUNT =

Fallback thread count used when Configuration#worker_count is unset, preserving the previously-hardcoded pool size.

2

Instance Method Summary collapse

Instance Method Details

#direct_writer ⇒ Object 



23
24
25
 
# File 'lib/end_point_blank/writers/delayed_writer.rb', line 23

def direct_writer
  @direct_writer ||= DirectWriter.new(url)
end

#enqueue(list) ⇒ Object 



69
70
71
72
 
# File 'lib/end_point_blank/writers/delayed_writer.rb', line 69

def enqueue(list)
  items = list.is_a?(Array) ? list : [list]
  items.each { |payload| enqueue_one(payload) }
end

#log_warning(message) ⇒ Object

Logs a drop warning. Overridable/stubbable in tests; routed through the pluggable EndPointBlank.logger seam (this lib is fire-and-forget and must never raise).



77
78
79
 
# File 'lib/end_point_blank/writers/delayed_writer.rb', line 77

def log_warning(message)
  EndPointBlank.logger.warn(message)
end

#pop_additional ⇒ Object 



63
64
65
66
67
 
# File 'lib/end_point_blank/writers/delayed_writer.rb', line 63

def pop_additional
  queue.pop(true)
rescue ThreadError
  nil
end

#queue ⇒ Object 



27
28
29
 
# File 'lib/end_point_blank/writers/delayed_writer.rb', line 27

def queue
  @queue ||= Queue.new
end

#start_threads ⇒ Object 



35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
 
# File 'lib/end_point_blank/writers/delayed_writer.rb', line 35

def start_threads
  @threads = []

  worker_count.times do
    @threads << Thread.new do
      loop do
        payload = queue.pop
        payloads = [payload]
        while (payload = pop_additional)
          payloads << payload
        end

        payloads.compact!
        while payloads.any?
          list = payloads[0..5]
          response = direct_writer.write(list)
          if response.status < 299
            on_success(response) if respond_to?(:on_success)
          elsif respond_to?(:on_failure)
            on_failure(response)
          end
          payloads -= list
        end
      end
    end
  end
end

#worker_count ⇒ Object 



31
32
33
 
# File 'lib/end_point_blank/writers/delayed_writer.rb', line 31

def worker_count
  EndPointBlank::Configuration.instance.worker_count || DEFAULT_WORKER_COUNT
end