Package org.acemq.amqp.core

Class AceMq

java.lang.Object
org.acemq.amqp.core.AceMq
All Implemented Interfaces:
AutoCloseable
public final class AceMq extends Object implements AutoCloseable
The entry point: one connection to one broker, and the publishers and consumers built on it. 

 try (AceMq mq = AceMq.connect("amqp://localhost")) {
     mq.declareExchange("orders", "topic");
     mq.declareQueue("orders.new");
     mq.bind("orders.new", "orders", "order.placed");

     Publisher<Order> publisher = mq.publisher("orders", "order.placed", Order.class);
     publisher.send(new Order("o-1", 42.00));

     mq.consume("orders.new", Order.class, message -> process(message.payload()));
 }

Payloads are JSON unless the publisher is told otherwise, and a consumer reads whatever format arrives. Neither side has to say anything about serialisation for the common case; DefaultPublisher.asXml() and its neighbours are there for the rest.

Instances are thread safe and long lived. Closing one closes every consumer and publisher created from it, so a try-with-resources block cannot leak a subscription.

  • Method Details

    • connect

      public static AceMq connect(String url)
      Connects to a broker, choosing the transport from the URL scheme.
      Parameters:
      url - broker URL, for example amqp://localhost:5672
      Returns:
      an open connection
      Throws:
      AceMqException - if no transport handles the scheme, or the broker cannot be reached

    • connect

      public static AceMq connect(String url, @Nullable Telemetry telemetry)
      Connects to a broker with an explicit telemetry sink.
      Parameters:
      url - broker URL
      telemetry - where to report
      Returns:
      an open connection

    • connect

      public static AceMq connect(ConnectionConfig config)
      Connects to a broker with explicit settings.
      Parameters:
      config - connection settings
      Returns:
      an open connection

    • connect

      public static AceMq connect(ConnectionConfig config, @Nullable Telemetry telemetry)
      Connects with an explicit telemetry sink.

      Preferable to auto-detection wherever the answer is known. Auto-detection reaches for Micrometer's global registry and OpenTelemetry's global instance, both of which are process-wide state; passing a sink lets two connections report separately and lets a test observe only its own measurements.

      Parameters:
      config - connection settings
      telemetry - where to report, or null to detect what is on the classpath
      Returns:
      an open connection

    • connect

      public static AceMq connect(String url, Codec codec)
      Connects to a broker, reading and writing payloads with the given codec.
      Parameters:
      url - broker URL
      codec - how payloads become bytes and back
      Returns:
      an open connection

    • connect

      public static AceMq connect(String url, @Nullable Telemetry telemetry, Codec codec)
      Connects to a broker with an explicit telemetry sink and codec.
      Parameters:
      url - broker URL
      telemetry - where to report, or null to detect what is on the classpath
      codec - how payloads become bytes and back
      Returns:
      an open connection

    • connect

      public static AceMq connect(ConnectionConfig config, @Nullable Telemetry telemetry, @Nullable Codec codec)
      Connects with an explicit telemetry sink and codec.

      Passing a codec fixes the format in both directions. Without one, publishers write JSON and consumers read every format on the classpath, which is what most applications want and what makes a format migration two ordinary releases. An application that has named a codec has said it wants one format, and should not find its consumers quietly accepting others.

      Parameters:
      config - connection settings
      telemetry - where to report, or null to detect what is on the classpath
      codec - the single format to read and write, or null for the defaults
      Returns:
      an open connection

    • capabilities

      public Set<Capability> capabilities()
      Reports what the connected broker supports.

      Worth logging at startup. When a capability is missing the engine either applies a documented alternative or refuses to start, so knowing the answer explains behaviour that would otherwise look arbitrary.

      Returns:
      the capabilities of this broker

    • supports

      public boolean supports(Capability capability)
      Returns:
      whether the broker supports a capability

    • telemetry

      public Telemetry telemetry()
      Returns:
      where this connection reports what it is doing; a no-op sink when neither Micrometer nor OpenTelemetry is on the classpath

    • drainConsumers

      public boolean drainConsumers(Duration timeout)
      Stops every consumer taking new work and waits for what is in hand.

      The consuming half of a cutover. Publishers are untouched, deliberately: a service being taken out of rotation still has requests to finish, and those requests still need to publish. Stop consuming first, stop publishing second, and the order is what makes the cutover clean.

      Stream readers are closed rather than drained, because a stream is resumable by construction: the next reader starts from the offset this one recorded.

      Parameters:
      timeout - how long to wait for handlers still running
      Returns:
      whether everything finished in time. False means something is still running, and the caller has a decision to make rather than a guarantee

    • pauseConsuming

      public void pauseConsuming()
      Stops every consumer taking new work, reversibly.

      Unlike drainConsumers(Duration) this does not wait and can be undone. For a maintenance window, or for holding a service still while something downstream catches up.

    • resumeConsuming

      public void resumeConsuming()
      Starts every paused consumer again.

    • pausePublishing

      public void pausePublishing()
      Refuses further publishes on this connection.

      The publishing half of a cutover, and the last thing to do before shutting a service down. A publish attempted while paused throws PublishingPausedException without sending anything, so there is never a half-published message to reason about.

      Publishes already in flight are not interrupted. They have been sent, and the broker will confirm them or not on its own schedule.

    • resumePublishing

      public void resumePublishing()
      Allows publishing again.

    • isPublishingPaused

      public boolean isPublishingPaused()
      Returns:
      whether publishing is currently refused

    • isConsumingPaused

      public boolean isConsumingPaused()
      Returns:
      whether every consumer on this connection is paused; false when there are none

    • inFlight

      public long inFlight()
      Returns:
      how many messages every consumer on this connection is handling right now

    • transportName

      public String transportName()
      Returns:
      the transport's short name, such as rabbitmq

    • isBlocked

      public boolean isBlocked()
      Whether the broker is currently refusing publishes.

      RabbitMQ blocks publishing connections when it runs low on memory or disk. Publishing while blocked throws ConnectionBlockedException at once, or after waiting up to blockedTimeout when one is set.

      Useful in a readiness probe, and deliberately not in a liveness one. A blocked broker is under pressure, not broken, and a service that reports itself dead every time an alarm fires will be restarted by its orchestrator while the broker recovers on its own.

      Returns:
      whether the broker is refusing publishes right now

    • blockedReason

      public Optional<String> blockedReason()
      Returns:
      the broker's own explanation for blocking, when it is blocked. RabbitMQ says low on memory or low on disk, which is the difference between an operator adding consumers and one adding disk

    • declareExchange

      public AceMq declareExchange(String name, String type)
      Declares a durable exchange.
      Parameters:
      name - exchange name
      type - direct, topic, fanout or headers
      Returns:
      this instance, for chaining

    • declareQueue

      public AceMq declareQueue(String name)
      Declares a durable quorum queue.

      Quorum is the default because a queue that survives losing its node is what almost everyone wants and almost nobody remembers to ask for.

      Parameters:
      name - queue name
      Returns:
      this instance, for chaining

    • declareQueue

      public AceMq declareQueue(String name, QueueType type, Map<String,Object> arguments)
      Declares a queue of a specific type.
      Parameters:
      name - queue name
      type - queue implementation
      arguments - broker-specific arguments
      Returns:
      this instance, for chaining

    • bind

      public AceMq bind(String queue, String exchange, String routingKey)
      Binds a queue to an exchange.
      Parameters:
      queue - queue name
      exchange - exchange name
      routingKey - routing key or pattern
      Returns:
      this instance, for chaining

    • publisher

      public <T> DefaultPublisher<T> publisher(String exchange, String routingKey)
      Creates a publisher for one destination.
      Type Parameters:
      T - payload type
      Parameters:
      exchange - target exchange, or the empty string to publish straight to a queue
      routingKey - routing key, or the queue name when publishing without an exchange
      Returns:
      a publisher, closed automatically when this instance closes

    • publisher

      public <T> DefaultPublisher<T> publisher(String exchange, String routingKey, Class<T> payloadType)
      Creates a publisher for one destination and one payload type.

      Preferable to the two-argument form, which infers the payload type from whatever the result is assigned to and so will happily produce a Publisher<Anything>. Naming the type makes the mistake a compile error instead of a message no consumer can read.

      Writes JSON. Call DefaultPublisher.asXml(), DefaultPublisher.asText() or DefaultPublisher.as(Codec) on the result for anything else.

      Type Parameters:
      T - payload type
      Parameters:
      exchange - target exchange, or the empty string to publish straight to a queue
      routingKey - routing key, or the queue name when publishing without an exchange
      payloadType - type this publisher sends
      Returns:
      a publisher, closed automatically when this instance closes

    • publisher

      public <T> DefaultPublisher<T> publisher(String exchange, String routingKey, Class<T> payloadType, PublishOptions options)
      Creates a publisher that publishes on the caller's terms rather than the safe defaults.

      The defaults — written to disk, unroutable treated as a failure — are what most messages want. This is for the ones that do not: telemetry that may be dropped, events with a shelf life, fan-outs nobody is required to be listening to.

      Type Parameters:
      T - payload type
      Parameters:
      exchange - target exchange, or the empty string to publish straight to a queue
      routingKey - routing key, or the queue name when publishing without an exchange
      payloadType - type this publisher sends
      options - how these messages should be published
      Returns:
      a publisher, closed automatically when this instance closes

    • intercept

      public AceMq intercept(PublishInterceptor interceptor)
      Runs something around every publish on this connection.

      For what every message in an organisation needs and no library can guess: a tenant identifier, an authorisation token, a schema version, a size limit. Without this they get copied into every call site, where one of them is always missing. 

      
 mq.intercept((PublishInterceptor) context ->
         context.withEnvelope(context.envelope().toBuilder()
                 .header("tenant", TenantContext.current())
                 .build()));

      Applies to publishers already created as well as later ones, so ordering against start-up code is not a trap. Register during start-up all the same: a message already on its way will not see an interceptor added mid-flight.

      Parameters:
      interceptor - what to run; throwing from it refuses the publish
      Returns:
      this instance, for chaining

    • intercept

      public AceMq intercept(ConsumeInterceptor interceptor)
      Runs something around every handler on this connection.

      For the work that surrounds handling rather than being part of it: log context, tenant adoption, opening and closing a unit of work.

      Throwing from beforeHandle fails the delivery, which means it is retried and eventually dead-lettered. That is the honest outcome for a refused message; acknowledging one nothing processed is not.

      Parameters:
      interceptor - what to run around each handler
      Returns:
      this instance, for chaining

    • messageCount

      public long messageCount(String queue)
      Counts the messages waiting in a queue.

      A snapshot, and useful as one: queue depth is the number an operator wants before deciding whether to replay, scale up, or leave well alone. It is not a control value — a live queue is a different depth by the time this returns, so a loop written around it is a loop written around a number that has already changed.

      Parameters:
      queue - queue to measure
      Returns:
      how many messages are waiting

    • replay

      public Replay replay(String queue)
      Moves failed messages back to the queue they failed in.

      The other half of dead-lettering. Messages that exhausted their retries are in <queue>.dlq and messages that could not be decoded are in <queue>.parked; both are safe there and neither comes back on its own. This is how they do, once the bug is fixed or the downstream service is up. 

      
 Replay replay = mq.replay("orders.new");
 long waiting = replay.pending();     // look first
 replay.replay(50);                   // then move a bounded batch
      Parameters:
      queue - the queue that was being consumed, not the dead-letter queue itself
      Returns:
      a replay reading <queue>.dlq; call Replay.parked() for the other one

    • declareStream

      public AceMq declareStream(String name, @Nullable Duration maxAge, @Nullable Long maxLengthBytes)
      Declares a stream: an append-only log that keeps messages until retention removes them.

      Not a queue with different settings. A stream is read without being emptied, every consumer holds its own position, and nothing is ever dead-lettered — see StreamConsumer for what that rules out.

      Retention is not optional in practice. A stream with neither a size nor an age limit grows until the disk is full, and the broker will not stop it.

      Parameters:
      name - stream name
      maxAge - how long a message is kept, or null for no age limit
      maxLengthBytes - how large the stream may grow, or null for no size limit
      Returns:
      this instance, for chaining
      Throws:
      AceMqException - if the broker does not support streams

    • declareStream

      public AceMq declareStream(String name, @Nullable Duration maxAge, @Nullable Long maxLengthBytes, @Nullable Long segmentBytes)
      Declares a stream, saying how large each file on disk gets.

      A stream is stored as a sequence of segment files, and retention happens a whole segment at a time: nothing is discarded until an entire segment can be. That makes the segment size the granularity of every other retention setting — a stream told to keep an hour, in segments large enough to hold a day, keeps a day.

      Absent unless asked for. No default is invented here, because the broker has one and it is tuned for the broker's own storage rather than for any particular stream. Naming a size that merely repeated the broker's would freeze the broker's choice at whatever it was on the day this was written, and a declaration carrying an argument the first one did not is a PRECONDITION_FAILED for whoever declares the same stream second. Go, Python and Ruby all leave it out the same way.

      Parameters:
      name - stream name
      maxAge - how long a message is kept, or null for no age limit
      maxLengthBytes - how large the stream may grow, or null for no size limit
      segmentBytes - how large each segment file gets, or null to leave it to the broker
      Returns:
      this instance, for chaining
      Throws:
      AceMqException - if the broker does not support streams

    • stream

      public <T> StreamReader<T> stream(String queue, Class<T> payloadType)
      Reads a stream, starting where the returned reader is told to. 
      
 mq.stream("orders.log", OrderPlaced.class)
         .fromFirst()
         .consume(message -> projection.apply(message.payload()));

      Nothing is consumed until consume is called.

      Type Parameters:
      T - payload type
      Parameters:
      queue - stream to read
      payloadType - type to decode payloads into
      Returns:
      a reader awaiting a starting position and a handler
      Throws:
      AceMqException - if the broker does not support streams

    • consume

      public <T> MessageConsumer consume(String queue, Class<T> payloadType, MessageHandler<T> handler)
      Starts consuming a queue.
      Type Parameters:
      T - payload type
      Parameters:
      queue - queue to consume
      payloadType - type to decode payloads into
      handler - called for each message; returning normally acknowledges it
      Returns:
      a handle that stops consumption when closed

    • pipeline

      public <T> PipelineBuilder<T,T> pipeline(String name, Class<T> entryType)
      A chain of steps, each with its own queue. 
      
 try (Pipeline<Order> fulfilment = 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()) {

     fulfilment.send(order);
 }

      Every hop goes through the broker, so a crash leaves the message where it was, a slow step grows its own queue rather than blocking the chain, and the step needing ten consumers gets ten while its neighbour keeps one. The cost is a round trip and a durable write per hop.

      Where a message is going travels with it, so nothing here coordinates.

      Type Parameters:
      T - entry type
      Parameters:
      name - pipeline name; also the exchange, with one queue per step beneath it
      entryType - type the pipeline is entered with
      Returns:
      a builder; nothing is declared until build() is called

    • ordered

      public <T> OrderedQueueBuilder<T> ordered(String name, Class<T> payloadType)
      A queue where messages sharing a key are handled in the order they were sent. 
      
 try (OrderedQueue<Order> orders = mq.ordered("orders", Order.class)
         .partitions(8)
         .keyedBy(Order::customerId)
         .declare()) {

     orders.send(order);
     orders.consume(message -> ledger.post(message.payload()));
 }

      The key decides a partition, each partition is a queue, and each queue has exactly one consumer. That gives ordering within a key and parallelism across keys, which is almost always the ordering anyone actually wanted: two orders for one customer are sequenced, two orders for different customers never needed to be.

      The retry ladder is not available here, because republishing a failed message to come back later is what breaks a sequence. See OrderedQueue.OnFailure for the three things that can happen instead, all of which preserve order and none of which are free.

      Type Parameters:
      T - payload type
      Parameters:
      name - logical name; also the exchange, with one queue per partition beneath it
      payloadType - type to publish and decode
      Returns:
      a builder; nothing is declared until declare() is called

    • consumeGroup

      public <T> ConsumerGroupBuilder<T> consumeGroup(String queue, Class<T> payloadType, MessageHandler<T> handler)
      Consumes a queue with several consumers, resizable while the application runs. 
      
 ConsumerGroup orders = mq.consumeGroup("orders.new", Order.class, handler)
         .concurrency(4)
         .prefetch(50)
         .start();

 orders.scaleTo(8);

      Prefetch and concurrency are the two numbers that decide how fast a queue drains, and both are usually guessed once and frozen in a properties file because changing either has meant a redeploy. Neither has to be.

      Type Parameters:
      T - payload type
      Parameters:
      queue - queue to consume
      payloadType - type to decode payloads into
      handler - called for each message
      Returns:
      a builder; nothing consumes until start() is called

    • consume

      public <T> MessageConsumer consume(String queue, Class<T> payloadType, ConsumerOptions options, MessageHandler<T> handler)
      Starts consuming a queue with explicit options.
      Type Parameters:
      T - payload type
      Parameters:
      queue - queue to consume
      payloadType - type to decode payloads into
      options - prefetch and failure behaviour
      handler - called for each message
      Returns:
      a handle that stops consumption when closed

    • deleteQueue

      public void deleteQueue(String name)
      Removes a queue and its contents.
      Parameters:
      name - queue to delete
      Implementation Note:
      intended for test fixtures and topology migration rather than application code.

    • requester

      public Requester requester()
      Opens a requester for asking questions and waiting for answers.

      Before reaching for this: request/reply over a broker is synchronous calling in asynchronous clothes. Where two services can speak HTTP or gRPC, they should — those have timeouts, load balancing and circuit breakers that this will not match. Where the callee is reachable only on the broker, this is the right tool and doing it by hand means reply queues, correlation ids and a timeout somebody forgets.

      One requester per connection. It owns a reply queue, so building one per call creates and destroys a queue for every question.

      Returns:
      a requester; close it to release its reply queue

    • respond

      public <Q, A> Responder respond(String queue, Class<Q> requestType, Function<Q,A> handler)
      Answers requests arriving on a queue. 
      
 try (Responder responder = mq.respond("pricing", Quote.class, quote -> price(quote))) {
     // serving until closed
 }

      The reply goes to whatever the request asked for and carries its correlation id back; the handler is a plain function and never sees either.

      Type Parameters:
      Q - request type
      A - response type
      Parameters:
      queue - queue requests arrive on
      requestType - type to decode requests into
      handler - computes the answer
      Returns:
      a responder; close it to stop serving

    • respond

      public <Q, A> Responder respond(String queue, Class<Q> requestType, ConsumerOptions options, Function<Q,A> handler)
      As respond(String, Class, java.util.function.Function), with consumer options.

      Concurrency is the one worth setting: a responder handles one request at a time by default, and every caller behind a slow one is blocked waiting.

      Type Parameters:
      Q - request type
      A - response type
      Parameters:
      queue - queue requests arrive on
      requestType - type to decode requests into
      options - prefetch, concurrency, codec and failure policy
      handler - computes the answer
      Returns:
      a responder; close it to stop serving

    • topology

      public TopologyPlanner topology()
      Plans and applies a declared topology.

      Preferable to the individual declare methods for anything beyond a single queue: it works out what would change before changing it, and can be asked to report without acting at all.

      Returns:
      a planner bound to this connection

    • isOpen

      public boolean isOpen()
      Returns:
      whether the underlying connection is usable

    • close

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

    • close

      public void close(Duration budget)
      Closes the connection, giving everything still working at most budget in total.

      The budget is shared, and that is the whole point of the parameter. Consumer groups drain before they close, and a connection that handed each group its own full timeout would take as long as the number of groups multiplied by it — a figure nobody chose, quietly larger than whatever grace period the caller is shutting down inside.

      Consumers that are not in a group are cancelled without waiting, as they always have been; what they hold is redelivered.

      Parameters:
      budget - the whole time this close may spend waiting for work in progress