Package org.acemq.amqp.core

Class Pipeline<T>

java.lang.Object
org.acemq.amqp.core.Pipeline<T>
Type Parameters:
T - payload the pipeline is entered with
All Implemented Interfaces:
AutoCloseable
public final class Pipeline<T> extends Object implements AutoCloseable
A chain of steps, each its own queue. 

 try (Pipeline<Order> fulfilment = mq.pipeline("fulfilment", Order.class)
         .step("validate", Order.class, new ValidateOrder())
         .step("enrich", Enriched.class, new EnrichOrder())
         .step("dispatch", Void.class, new DispatchOrder())
         .build()) {

     fulfilment.send(order);
 }

Every hop goes through the broker rather than through a method call, and that is the decision everything else follows from. Chaining handlers in one process is function composition wearing a messaging costume: a crash loses the work, a slow step blocks the chain, and scaling is all or nothing. With a queue between two steps, a crash leaves the message where it was, a slow step grows its own queue while the others carry on, and the step that needs ten consumers gets ten while its neighbour keeps one.

The cost is a network round trip and a durable write per hop. Three steps is three publishes and three confirms. Where each step is microseconds of CPU, a method call is the right tool and this is not.

Where a message is going travels with it — see RoutingSlip. Nothing here coordinates; each step reads the slip and publishes to whatever is next.

  • Method Details

    • name

      public String name()
      Returns:
      the pipeline's name, which is also its exchange

    • stepNames

      public List<String> stepNames()
      Returns:
      the step names, in order

    • queueFor

      public String queueFor(String step)
      Returns:
      the queue backing a step

    • send

      public String send(T payload)
      Sends a payload in at the first step.
      Parameters:
      payload - the message
      Returns:
      the identifier of this run, which every hop carries and which survives a dead-letter and a replay

    • send

      public String send(T payload, Envelope envelope)
      Parameters:
      payload - the message
      envelope - metadata to carry, for correlation with whatever caused this
      Returns:
      the run identifier

    • send

      public String send(Itinerary itinerary, T payload, Envelope envelope)
      Sends a payload to the first stop of an itinerary the message carries with it. 
      
 pipeline.send(Itinerary.empty()
         .then("orders-events", "order.validate", "validate")
         .then("orders-events", "order.charge", "charge"), order);

      The other way to start a run, and not the default. send(Object) writes the declared route as step names against this pipeline's own steps, which is the better shape when there is a declaration to resolve against — it stays readable in a management console and it costs one short header. This writes the whole itinerary as JSON, in the shape Go, Python and Ruby write, for when the route is decided per message and whoever consumes it may have no declaration at all.

      Nothing about the pipeline's own topology is used: the stops name their own exchanges and routing keys, and the first of them is where this goes.

      Parameters:
      itinerary - where the message is going, with at least one stop on it
      payload - the message
      envelope - metadata to carry
      Returns:
      the identifier of the message that started the run

    • send

      public String send(Itinerary itinerary, T payload)
      Parameters:
      itinerary - where the message is going
      payload - the message
      Returns:
      the identifier of the message that started the run

    • step

      public ConsumerGroup step(String step)
      The consumers running one step, so it can be scaled independently.

      The operational argument for putting queues between steps at all: a pipeline's slowest step is its throughput, and this is how that one step gets more consumers without touching the others.

      Parameters:
      step - step name
      Returns:
      its consumer group

    • entered

      public long entered()
      Returns:
      how many messages have been sent into this pipeline by this instance

    • completed

      public long completed()
      Returns:
      how many runs reached the end of the route

    • endedEarly

      public long endedEarly()
      Returns:
      how many runs were ended early by a step returning nothing

    • inFlight

      public long inFlight()
      Returns:
      how many messages every step is handling right now

    • close

      public void close()
      Specified by:
      close in interface AutoCloseable

    • toString

      public String toString()
      Overrides:
      toString in class Object

    • entryType

      public Class<T> entryType()
      Returns:
      the type the pipeline is entered with