Class: AceMQ::AMQP::Patterns::Pipeline

Inherits:
Object
  • Object
show all
Defined in:
lib/acemq/amqp/patterns/routing_slip.rb

Overview

A route declared once, that messages travel by name.

orders = Patterns::Pipeline.new("orders", %w[validate charge ship])
mq.apply(orders.topology)

mq.consume(orders.queue_for("charge"), &orders.follow(mq) do |message|
charge(message.payload)
end)

orders.start(mq, { "order_id" => "A-1" })

The other half of RoutingSlip, and the half Java has. A slip carries its destinations; a pipeline knows them, and the message carries only the step names and how far along it is. That is three short headers instead of a JSON document, and a route that reads in a management console without decoding anything — at the price of a route that is the same for every message and has to be declared on both ends.

The naming is Java's, exactly. The exchange is the pipeline's name and is direct, the routing key is the step name, and the queue behind a step is pipeline.step. A Ruby consumer that got any of those wrong would be listening where no Java service publishes, so they are not arranged for Ruby's convenience.

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(name, steps) ⇒ Pipeline

Returns a new instance of Pipeline.

Parameters:

  • name (String) —

    the pipeline, which is also its exchange

  • steps (Array<String>) —

    the step names, in order

Raises:

  • (ArgumentError) 


344
345
346
347
348
349
350
351
 
# File 'lib/acemq/amqp/patterns/routing_slip.rb', line 344

def initialize(name, steps)
  @name = name.to_s
  @steps = Array(steps).map(&:to_s)
  raise ArgumentError, "a pipeline needs a name" if @name.empty?
  raise ArgumentError, "pipeline #{@name} needs at least one step" if @steps.empty?

  freeze
end

Instance Attribute Details

#name ⇒ Object (readonly)

Returns the value of attribute name.



340
341
342
 
# File 'lib/acemq/amqp/patterns/routing_slip.rb', line 340

def name
  @name
end

#steps ⇒ Object (readonly)

Returns the value of attribute steps.



340
341
342
 
# File 'lib/acemq/amqp/patterns/routing_slip.rb', line 340

def steps
  @steps
end

Instance Method Details

#exchange ⇒ Object

The exchange every step of this pipeline is published to.



354
 
# File 'lib/acemq/amqp/patterns/routing_slip.rb', line 354

def exchange = @name

#follow(connection) ⇒ Object

AceMQ::AMQP::Patterns.follow_slip bound to this pipeline.



395
 
# File 'lib/acemq/amqp/patterns/routing_slip.rb', line 395

def follow(connection, &) = Patterns.follow_slip(connection, pipeline: self, &)

#queue_for(step) ⇒ Object

The queue behind a step.

Raises:

  • (ArgumentError) —

    when the pipeline has no such step, which is worth catching here: a consumer subscribed to a queue nothing publishes to simply never receives anything.



361
362
363
364
365
366
367
368
 
# File 'lib/acemq/amqp/patterns/routing_slip.rb', line 361

def queue_for(step)
  unless @steps.include?(step.to_s)
    raise ArgumentError,
          "pipeline #{@name} has no step called '#{step}'. Its steps are " \
          "#{@steps.join(", ")}."
  end
  "#{@name}.#{step}"
end

#slip ⇒ Object

A fresh slip at the start of this route, in the declared form.



384
385
386
387
 
# File 'lib/acemq/amqp/patterns/routing_slip.rb', line 384

def slip
  RoutingSlip.new(steps: @steps.map { |step| step_for(step) }, form: RoutingSlip::ROUTE,
                  pipeline: self)
end

#start(connection, payload, **fields) ⇒ Envelope

Sends a payload into the first step.

Returns:



392
 
# File 'lib/acemq/amqp/patterns/routing_slip.rb', line 392

def start(connection, payload, **fields) = slip.start(connection, payload, **fields)

#step_named(step) ⇒ Object

One of this pipeline's steps, as a Step that knows where to publish.

Raises:

  • (ArgumentError) —

    when the pipeline has no step by that name



400
401
402
403
 
# File 'lib/acemq/amqp/patterns/routing_slip.rb', line 400

def step_named(step)
  queue_for(step)
  step_for(step.to_s)
end

#to_s ⇒ Object 



405
 
# File 'lib/acemq/amqp/patterns/routing_slip.rb', line 405

def to_s = "Pipeline[#{@name}: #{@steps.join(" -> ")}]"

#topology ⇒ Topology

The exchange, the queues and the bindings this pipeline needs.

A Topology to apply rather than declarations made here, so it composes with whatever else a service declares and is applied once at start-up like everything else.

Returns:



377
378
379
380
381
 
# File 'lib/acemq/amqp/patterns/routing_slip.rb', line 377

def topology
  @steps.reduce(Topology.new.exchange(@name, "direct")) do |built, step|
    built.queue(queue_for(step)).binding(queue_for(step), @name, step)
  end
end