Class: AceMQ::Rails::Runner

Inherits:
Object
  • Object
show all
Defined in:
lib/acemq/rails/runner.rb

Overview

The consumer process: what rake acemq:consume and acemq-consumer run.

This is deliberately not a thing the web server does. See docs/consumers.md for the argument at length; the short form is that Puma owns its workers' lifetimes and a message handler is not a request. A consumer started inside Puma is forked with the worker if preload_app! is on (and a bunny socket does not survive a fork), killed by worker_timeout if a handler takes longer than a slow request should, scaled by however many web processes the traffic wants rather than by how deep the queue is, and drained on Puma's schedule rather than on the broker's. None of those is a bug that can be fixed from in here.

What it does, in order: eager-load the application so consumer classes exist, apply the topology, subscribe every runnable consumer, and then sleep until a signal arrives — at which point it drains.

Defined Under Namespace

Modules: Passthrough

Constant Summary collapse

SIGNALS =

Signals that mean "stop": what Kubernetes, systemd, Docker and a terminal all send.

%w[INT TERM].freeze

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(config: AceMQ::Rails.config, connection: nil, logger: nil) ⇒ Runner

Returns a new instance of Runner.

Parameters:

  • config (Configuration) (defaults to: AceMQ::Rails.config)
  • connection (AceMQ::AMQP::Connection, nil) (defaults to: nil)
  • logger (#info, #warn, #error, nil) (defaults to: nil) 


44
45
46
47
48
49
50
51
 
# File 'lib/acemq/rails/runner.rb', line 44

def initialize(config: AceMQ::Rails.config, connection: nil, logger: nil)
  @config = config
  @connection = connection
  @logger = logger || AceMQ::Rails.logger
  @consumers = []
  @stopping = Thread::Queue.new
  @lock = Mutex.new
end

Instance Attribute Details

#config ⇒ Object (readonly)

Returns the value of attribute config.



53
54
55
 
# File 'lib/acemq/rails/runner.rb', line 53

def config
  @config
end

#consumers ⇒ Object (readonly)

Returns the value of attribute consumers.



53
54
55
 
# File 'lib/acemq/rails/runner.rb', line 53

def consumers
  @consumers
end

Instance Method Details

#connection ⇒ Object

The connection. Opened on first use, so a runner can be built in a test without one.



57
58
59
 
# File 'lib/acemq/rails/runner.rb', line 57

def connection
  @lock.synchronize { @connection ||= AceMQ::Rails.connection }
end

#drain(timeout: config.shutdown_timeout) ⇒ Boolean

Stops every consumer, waits for the handlers they are already running, and closes the connection.

What this finishes and what it abandons is set out with measurements in docs/lifecycle.md. In brief:

  • A handler in flight finishes, and its message is acknowledged. The drain blocks on it.
  • A delivery the broker had sent but no handler had picked up goes back to the broker unacknowledged and is redelivered.
  • A retry waiting out a short backoff is waited out in full, because the library's backoff is a sleep on the handler's own thread and is therefore in flight. A long one is waiting on a rung queue in the broker and costs this nothing. Which is which is consumer.broker_wait_threshold, thirty seconds by default, and it is the single setting that most affects how long a deploy takes.
  • A publish waiting for a confirm is waited for: the library's publish is synchronous, so a publish in progress is a thread this has to join or abandon, and the transport close below happens after.

The drain itself is the library's Connection#close, which since acemq-amqp 0.7.0 spends one deadline across every consumer rather than a fresh one on each. Until then it did not, which is why this class had a drain of its own: eight consumers each given thirty seconds is four minutes, and four minutes into a shutdown Kubernetes has long since sent SIGKILL — killing every handler mid-flight and leaving everything they held unsettled, the exact outcome draining exists to avoid. One deadline is the library's answer now, so this passes shutdown_timeout to it and keeps none of the arithmetic.

What is left here is the shape a process wants rather than the shape a library wants: a boolean and a log line, where the library raises AMQP::DrainTimeout. acemq-consumer turns that boolean into exit 0 or 75, and an exception out of a signal handler's drain would be a backtrace where an exit status belongs.

Returns:

  • (Boolean) —

    whether every handler finished in time



129
130
131
132
133
134
135
136
137
138
 
# File 'lib/acemq/rails/runner.rb', line 129

def drain(timeout: config.shutdown_timeout)
  started = monotonic
  log_info "acemq: draining #{@consumers.size} consumer(s), #{timeout}s at most"

  finished = close_connection(timeout)

  took = (monotonic - started).round(2)
  log_info "acemq: drained in #{took}s" if finished
  finished
end

#run(classes = nil) ⇒ Boolean

Boots, runs, and drains on a signal. Returns when the drain is over.

Parameters:

  • classes (Array<Class>, nil) (defaults to: nil) —

    the consumers to run; every runnable registered class by default

Returns:

  • (Boolean) —

    whether the drain finished within the deadline



66
67
68
69
70
71
72
 
# File 'lib/acemq/rails/runner.rb', line 66

def run(classes = nil)
  trap_signals
  start(classes)
  log_info "acemq: #{@consumers.size} consumer(s) running; waiting for a signal"
  @stopping.pop
  drain
end

#start(classes = nil) ⇒ Array<AceMQ::AMQP::Consumer>

Applies the topology and subscribes every consumer, without waiting.

Separated from #run so a test can start, publish, assert and stop without a signal anywhere near it.

Returns:

  • (Array<AceMQ::AMQP::Consumer>) 


80
81
82
83
84
85
86
87
 
# File 'lib/acemq/rails/runner.rb', line 80

def start(classes = nil)
  classes = Array(classes || Registry.runnable)
  refuse_an_empty_run(classes)
  complain_about_reloading
  apply_topology

  @consumers = classes.map { |klass| subscribe(klass) }
end

#stop ⇒ Object

Asks #run to come back. For a test, and for anything embedding this.



90
 
# File 'lib/acemq/rails/runner.rb', line 90

def stop = @stopping << :stop