Class: Chronos::Application::DeliveryPipeline

Inherits:
Object
  • Object
show all
Defined in:
lib/chronos/application/delivery_pipeline.rb

Overview

Coordinates bounded queueing, retry, backlog, circuit state, and delivery.

Examples:

pipeline.enqueue(serialized_event)
pipeline.flush(2.0)

Constant Summary collapse

STATES = 
[:accepted, :queued, :serialized, :sent, :retried, :dropped, :rejected].freeze

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(config, transport, logger = nil, options = {}) ⇒ DeliveryPipeline

Returns a new instance of DeliveryPipeline.



21
22
23
24
25
26
27
 
# File 'lib/chronos/application/delivery_pipeline.rb', line 21

def initialize(config, transport, logger = nil, options = {})
  @config = config
  @transport = transport
  @logger = logger || Internal::SafeLogger.new(config.logger)
  initialize_resilience(options)
  initialize_storage(options)
end

Instance Attribute Details

#remote_configuration ⇒ Object (readonly)

Returns the value of attribute remote_configuration.



19
20
21
 
# File 'lib/chronos/application/delivery_pipeline.rb', line 19

def remote_configuration
  @remote_configuration
end

Instance Method Details

#capture_allowed?(event_type, fingerprint = nil) ⇒ Boolean

Returns:

  • (Boolean) 


63
64
65
66
67
68
 
# File 'lib/chronos/application/delivery_pipeline.rb', line 63

def capture_allowed?(event_type, fingerprint = nil)
  @remote_configuration.capture?(event_type, fingerprint)
rescue StandardError => error
  diagnose(error)
  false
end

#close(timeout) ⇒ Object 



85
86
87
88
89
90
91
92
93
94
95
 
# File 'lib/chronos/application/delivery_pipeline.rb', line 85

def close(timeout)
  return true if closed?

  @state_mutex.synchronize { @closed = true }
  flushed = @worker_pool.close(timeout)
  @transport.close
  flushed
rescue StandardError => error
  diagnose(error)
  false
end

#deliver_sync(event) ⇒ Object 



44
45
46
47
48
49
50
51
52
 
# File 'lib/chronos/application/delivery_pipeline.rb', line 44

def deliver_sync(event)
  record_pre_delivery(event)
  result = deliver_with_backlog(event)
  result && result.success?
rescue StandardError => error
  diagnose(error)
  store_in_backlog(event)
  false
end

#diagnostics ⇒ Object 



97
98
99
100
101
102
103
104
105
 
# File 'lib/chronos/application/delivery_pipeline.rb', line 97

def diagnostics
  {
    :states => state_counts,
    :queue => @queue.stats,
    :backlog => @backlog.stats,
    :circuit => @circuit_breaker.state,
    :remote_configuration => @remote_configuration.to_h
  }
end

#enqueue(event) ⇒ Object 



29
30
31
32
33
34
35
36
37
38
39
40
41
42
 
# File 'lib/chronos/application/delivery_pipeline.rb', line 29

def enqueue(event)
  record_pre_delivery(event)
  if @worker_pool.enqueue(event)
    transition(:queued)
    true
  else
    transition(:dropped)
    false
  end
rescue StandardError => error
  diagnose(error)
  transition(:dropped)
  false
end

#event_enabled?(event_type) ⇒ Boolean

Returns:

  • (Boolean) 


70
71
72
73
74
75
 
# File 'lib/chronos/application/delivery_pipeline.rb', line 70

def event_enabled?(event_type)
  @remote_configuration.delivery_enabled?(event_type)
rescue StandardError => error
  diagnose(error)
  false
end

#flush(timeout) ⇒ Object 



81
82
83
 
# File 'lib/chronos/application/delivery_pipeline.rb', line 81

def flush(timeout)
  @worker_pool.flush(timeout)
end

#max_payload_size ⇒ Object 



77
78
79
 
# File 'lib/chronos/application/delivery_pipeline.rb', line 77

def max_payload_size
  @remote_configuration.max_payload_size
end

#send_event(event) ⇒ Object

WorkerPool delivery entry point. Events were counted when accepted into the queue.



55
56
57
58
59
60
61
 
# File 'lib/chronos/application/delivery_pipeline.rb', line 55

def send_event(event)
  deliver_with_backlog(event)
rescue StandardError => error
  diagnose(error)
  store_in_backlog(event)
  Ports::TransportResult.new(:network_error, :error => error.class.name)
end