Class: Shikibu::Outbox::Relayer

Inherits:
Object
  • Object
show all
Defined in:
lib/shikibu/outbox/relayer.rb

Overview

Background relayer for publishing outbox events to external message brokers.

The relayer polls the database for pending events and publishes them as CloudEvents to a configured HTTP endpoint. It implements exponential backoff for retries and graceful shutdown.

Examples:

relayer = Shikibu::Outbox::Relayer.new(
  storage: storage,
  broker_url: 'http://broker-ingress.default.svc.cluster.local'
)
relayer.start
# ... later ...
relayer.stop

Constant Summary collapse

DEFAULT_POLL_INTERVAL = 
1.0
DEFAULT_MAX_RETRIES = 
3
DEFAULT_BATCH_SIZE = 
10
MAX_BACKOFF = 
30.0
HTTP_OPEN_TIMEOUT = 
10
HTTP_READ_TIMEOUT = 
30

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(storage:, broker_url:, wake_event: nil, poll_interval: DEFAULT_POLL_INTERVAL, max_retries: DEFAULT_MAX_RETRIES, batch_size: DEFAULT_BATCH_SIZE, max_age_hours: nil) ⇒ Relayer 



33
34
35
36
37
38
39
40
41
42
43
44
 
# File 'lib/shikibu/outbox/relayer.rb', line 33

def initialize(storage:, broker_url:, wake_event: nil, poll_interval: DEFAULT_POLL_INTERVAL,
               max_retries: DEFAULT_MAX_RETRIES, batch_size: DEFAULT_BATCH_SIZE, max_age_hours: nil)
  @storage = storage
  @broker_url = URI.parse(broker_url)
  @wake_event = wake_event
  @poll_interval = poll_interval
  @max_retries = max_retries
  @batch_size = batch_size
  @max_age_hours = max_age_hours
  @running = false
  @thread = nil
end

Instance Attribute Details

#batch_size ⇒ Object (readonly)

Returns the value of attribute batch_size.



31
32
33
 
# File 'lib/shikibu/outbox/relayer.rb', line 31

def batch_size
  @batch_size
end

#broker_url ⇒ Object (readonly)

Returns the value of attribute broker_url.



31
32
33
 
# File 'lib/shikibu/outbox/relayer.rb', line 31

def broker_url
  @broker_url
end

#max_age_hours ⇒ Object (readonly)

Returns the value of attribute max_age_hours.



31
32
33
 
# File 'lib/shikibu/outbox/relayer.rb', line 31

def max_age_hours
  @max_age_hours
end

#max_retries ⇒ Object (readonly)

Returns the value of attribute max_retries.



31
32
33
 
# File 'lib/shikibu/outbox/relayer.rb', line 31

def max_retries
  @max_retries
end

#poll_interval ⇒ Object (readonly)

Returns the value of attribute poll_interval.



31
32
33
 
# File 'lib/shikibu/outbox/relayer.rb', line 31

def poll_interval
  @poll_interval
end

#storage ⇒ Object (readonly)

Returns the value of attribute storage.



31
32
33
 
# File 'lib/shikibu/outbox/relayer.rb', line 31

def storage
  @storage
end

Instance Method Details

#running? ⇒ Boolean 



63
64
65
 
# File 'lib/shikibu/outbox/relayer.rb', line 63

def running?
  @running
end

#start ⇒ Object 



46
47
48
49
50
51
52
 
# File 'lib/shikibu/outbox/relayer.rb', line 46

def start
  return if @running

  @running = true
  @thread = Thread.new { poll_loop }
  log_info("started (broker=#{@broker_url}, poll_interval=#{@poll_interval}s)")
end

#stop ⇒ Object 



54
55
56
57
58
59
60
61
 
# File 'lib/shikibu/outbox/relayer.rb', line 54

def stop
  return unless @running

  @running = false
  @wake_event&.signal # Wake up if waiting
  @thread&.join(5)
  log_info('stopped')
end