Module: AceMQ::Rails

Defined in:
lib/acemq/rails.rb,
lib/acemq/rails/health.rb,
lib/acemq/rails/runner.rb,
lib/acemq/rails/railtie.rb,
lib/acemq/rails/version.rb,
lib/acemq/rails/consumer.rb,
lib/acemq/rails/registry.rb,
lib/acemq/rails/configuration.rb,
lib/acemq/rails/topology_builder.rb

Overview

Rails integration for AceMQ.

One connection for the process, reachable from anywhere in the application; consumers written as classes and run in a process of their own; a health check that composes into whatever the application already exposes.

A note on the name

This module is called Rails and sits inside AceMQ, which means that inside any file lexically nested in module AceMQ; module Rails, a bare Rails resolves to this module and not to the framework. Every reference to the framework in this gem is therefore written ::Rails, and that is not a style choice — leaving one off gives a NoMethodError on AceMQ::Rails.application that reads like a bug in the application.

Defined Under Namespace

Modules: Health, Registry, TopologyBuilder Classes: Configuration, Consumer, Railtie, Runner

Constant Summary collapse

LOCK =

Guards the one connection. Named here rather than inside the singleton class, where a constant would belong to the singleton and not to the module — which compiles, and then cannot be reached by the name it was given.

Mutex.new
VERSION =

A version line of its own, deliberately.

This gem tracks two release trains: AceMQ's and Rails'. A Rails 8.1 that moves an autoloading hook is a release here and nothing at all in acemq-amqp, and a library release that adds a publishing method is a dependency bump here rather than a new number. Sharing a version with the library would mean one of those two facts had to be lied about.

"0.2.4"

Class Attribute Summary collapse

Class Method Summary collapse

Class Attribute Details

.config ⇒ Configuration

The configuration. Set from config/acemq.yml and config.acemq.* when a Railtie is loaded, and directly when there is no Rails — a rake task in a plain Ruby process, or a spec.

Returns:



55
56
57
 
# File 'lib/acemq/rails.rb', line 55

def config
  @config ||= Configuration.new
end

.logger ⇒ #info, ...

Where this gem logs. Rails.logger when there is one.

Returns:

  • (#info, #warn, #error, nil) 


64
65
66
67
68
 
# File 'lib/acemq/rails.rb', line 64

def logger
  return @logger if defined?(@logger) && @logger

  defined?(::Rails) && ::Rails.respond_to?(:logger) ? ::Rails.logger : nil
end

Class Method Details

.apply_topology ⇒ Object

Declares the topology. What rake acemq:topology runs, and what the consumer process does on boot.



157
158
159
160
161
 
# File 'lib/acemq/rails.rb', line 157

def apply_topology
  return nil unless TopologyBuilder.declared?(config.topology)

  connection.apply(topology)
end

.connected? ⇒ Boolean

Whether a connection has been opened. Does not open one.

Returns:

  • (Boolean) 


115
 
# File 'lib/acemq/rails.rb', line 115

def connected? = !@connection.nil?

.connection ⇒ AceMQ::AMQP::Connection

The connection this process publishes and consumes on.

One per process, opened on first use and memoised. A connection is a socket and a heartbeat; opening one per request would be the single most expensive mistake available here, and is the reason this exists rather than every caller writing Connection.open.

Opened lazily by default, which is a decision rather than laziness: a broker that is down should not stop an application from serving the pages that never touch it. config.acemq.connect_on_boot = true moves it to boot for an application that would rather fail fast.

Returns:

  • (AceMQ::AMQP::Connection) 


83
84
85
86
87
88
89
90
91
92
93
94
95
96
 
# File 'lib/acemq/rails.rb', line 83

def connection
  # Double-checked, and the outer check is not an optimisation: this is on
  # the path of every publish in every request, and taking a process-wide
  # mutex there would serialise a Puma worker's threads behind each other
  # for the whole life of the process.
  return @connection if @connection

  LOCK.synchronize do
    @connection ||= begin
      url, options = config.connection_arguments
      intercept(AceMQ::AMQP::Connection.open(url, **options))
    end
  end
end

.connection=(connection) ⇒ Object

Replaces the connection, for a test that wants a fake transport.

Parameters:

  • connection (AceMQ::AMQP::Connection, nil) 


120
121
122
 
# File 'lib/acemq/rails.rb', line 120

def connection=(connection)
  LOCK.synchronize { @connection = connection }
end

.consumers ⇒ Array<Class>

The registered consumer classes.

Returns:

  • (Array<Class>) 


166
 
# File 'lib/acemq/rails.rb', line 166

def consumers = Registry.runnable

.disconnect!(timeout: nil) ⇒ Object

Closes the connection, draining any consumers on it first.

Safe to call when nothing was ever opened, which is what makes it usable from an at_exit in a web process that may never have published.

The connection is let go of before it is closed, and not after: a close that raises — which since acemq-amqp 0.7.0 it does when the drain ran out of time — must still leave this process without a connection it thinks it has.

Parameters:

  • timeout (Numeric, nil) (defaults to: nil) —

    seconds for the whole drain, every consumer together. The library's own default when nil, which is what an at_exit wants; the consumer process passes shutdown_timeout

Raises:

  • (AceMQ::AMQP::DrainTimeout) —

    when handlers were still running when the deadline expired. The socket is shut either way



183
184
185
186
187
188
189
190
191
 
# File 'lib/acemq/rails.rb', line 183

def disconnect!(timeout: nil)
  connection = LOCK.synchronize do
    taken = @connection
    @connection = nil
    taken
  end
  timeout ? connection&.close(timeout: timeout) : connection&.close
  nil
end

.health ⇒ AceMQ::AMQP::Health::Report

What this process can say about its broker, blocked connections included. See Health.

Returns:

  • (AceMQ::AMQP::Health::Report) 


148
 
# File 'lib/acemq/rails.rb', line 148

def health = Health.of(connection)

.intercept(connection) ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Registers config.acemq.interceptors on a freshly opened connection.

Here rather than in an initializer, because an interceptor lives on a connection and an initializer that reaches for one opens the socket during boot — see AceMQ::Rails::Configuration#interceptors. Each is offered to both sides; the library keeps only the hooks the object answers to.



106
107
108
109
110
111
112
 
# File 'lib/acemq/rails.rb', line 106

def intercept(connection)
  Array(config.interceptors).each do |interceptor|
    connection.intercept_publish(interceptor)
    connection.intercept_consume(interceptor)
  end
  connection
end

.publish(payload, **options) ⇒ AceMQ::AMQP::Envelope

Publishes a message.

AceMQ::Rails.publish(order.as_json, to: "order.placed",
                   exchange: "orders-events", type: "order.placed.v2")

A thin delegation, and thin on purpose. The library's publish is already the right shape, and a wrapper that renamed its keywords would be a second API to document, a second one to keep in step, and the place where an integration starts making decisions the library deliberately left to the caller.

Returns:

  • (AceMQ::AMQP::Envelope) —

    what actually went on the wire



136
 
# File 'lib/acemq/rails.rb', line 136

def publish(payload, **options) = connection.publish(payload, **options)

.publish_all(payloads, **options) ⇒ Array<AceMQ::AMQP::Envelope>

Publishes a batch in one round trip. Added to the library in 0.6.0.

Returns:

  • (Array<AceMQ::AMQP::Envelope>) —

    in the order the payloads were given, whatever order the broker confirmed them in



142
 
# File 'lib/acemq/rails.rb', line 142

def publish_all(payloads, **options) = connection.publish_all(payloads, **options)

.topology ⇒ AceMQ::AMQP::Topology

The topology the configuration declares.

Returns:

  • (AceMQ::AMQP::Topology) 


153
 
# File 'lib/acemq/rails.rb', line 153

def topology = TopologyBuilder.build(config.topology)