Package org.acemq.amqp.core

Class PipelineBuilder<T,C>

java.lang.Object
org.acemq.amqp.core.PipelineBuilder<T,C>
Type Parameters:
T - payload the pipeline is entered with
C - payload the step declared most recently produces
public final class PipelineBuilder<T,C> extends Object
Declares the steps of a pipeline, in order, threading the types along the chain. 

 mq.pipeline("fulfilment", Order.class)
         .step("validate", Order.class, new ValidateOrder())
         .step("enrich", Enriched.class, new EnrichOrder())
             .withRetry(RetryPolicy.exponential(5, Duration.ofSeconds(2), Duration.ofMinutes(1)))
             .concurrency(10)
         .step("dispatch", Void.class, new DispatchOrder())
         .build();

The type parameter moves with each step, so a Step<Order, Enriched> cannot be put where one producing an Order belongs. That is the reason this is a builder rather than a list of names: a typo in a string is found in production, and a type error is found by the compiler.

The settings after a step apply to that step. They read as a chain because steps fail differently, and giving a pipeline one retry policy would mean giving the step that should never retry the same treatment as the one that always should.

  • Method Details

    • step

      public <O> PipelineBuilder<T,O> step(String stepName, Class<O> produces, Step<C,O> handler)
      Adds a step to the end of the chain.
      Type Parameters:
      O - produced type
      Parameters:
      stepName - how the step is named on the wire and in the routing slip
      produces - what this step's handler returns, which the next step receives
      handler - the work
      Returns:
      a builder whose next step receives O

    • describedAs

      public PipelineBuilder<T,C> describedAs(String text)
      Describes the most recently declared step, for the humans reading the logs. 
      
 .step("reserve", Reservation.class, this::reserve)
         .describedAs("hold stock for 15 minutes so payment cannot oversell")

      Separate from the step name rather than folded into it, because the name has a job: it is the routing key and the routing slip entry, so it is restricted to letters, digits, dashes and dots and cannot hold a sentence. Changing a name to make it more readable changes the wire format and orphans messages already in flight; changing a description changes a log line.

      Parameters:
      text - what this step is for, in a sentence
      Returns:
      this builder

    • withRetry

      public PipelineBuilder<T,C> withRetry(RetryPolicy policy)
      Retries the most recently declared step on a schedule, never in a sleeping handler: the engine holds a short wait and the broker's queues hold a long one.
      Parameters:
      policy - the schedule
      Returns:
      this builder

    • idempotent

      public PipelineBuilder<T,C> idempotent(IdempotencyStore store)
      Handles each message at most once at the most recently declared step.

      Worth configuring on any step whose work would matter twice. Each hop is at-least-once, so a step sees a message again whenever a publish succeeded and its acknowledgement did not.

      Parameters:
      store - where handled identifiers are remembered
      Returns:
      this builder

    • concurrency

      public PipelineBuilder<T,C> concurrency(int howMany)
      Parameters:
      howMany - consumers to run for the most recently declared step. This is how a slow step is given more capacity without touching its neighbours
      Returns:
      this builder

    • prefetch

      public PipelineBuilder<T,C> prefetch(int prefetch)
      Parameters:
      prefetch - how many messages each consumer of the most recent step may hold
      Returns:
      this builder

    • encodedAs

      public PipelineBuilder<T,C> encodedAs(Codec codec)
      Publishes the most recent step's output in a named format.

      A format change is not a step. Making it one would add a queue, a publish, a confirm and a consumer whose entire job is to deserialize and reserialize the same object — a network round trip to change nothing but the encoding.

      Parameters:
      codec - the format for this step's output
      Returns:
      this builder

    • build

      public Pipeline<T> build()
      Declares the topology and starts a consumer group per step.
      Returns:
      the running pipeline

    • toString

      public String toString()
      Overrides:
      toString in class Object