Listeners

@AceListener consumes a queue with a method.

@Component
class OrderListener {

    @AceListener(queue = "orders.new")
    void onOrder(Order order) {
        // ...
    }
}

The annotation goes on a method of any bean. Nothing else is needed: no container factory, no registrar, no interface to implement.

What the method may take

Exactly one parameter, in one of two shapes.

The payload, decoded with the configured codec:

@AceListener(queue = "orders.new")
void onOrder(Order order) { }

The whole message, when the envelope matters:

@AceListener(queue = "orders.new")
void onOrder(Message<Order> message) {
    Order order = message.payload();
    int attempt = message.attempt();
    String correlation = message.envelope().correlationId();
    Map<String, Object> headers = message.headers();
}

Message<T> carries the payload, the envelope — id, type, correlation and causation ids, attempt count, first-seen time, origin — the raw headers, the queue it came from and the time it was received.

Two parameters is an error, and so is a raw Message without a type argument. Both fail at startup with the method in the message, because the codec needs a concrete type to decode into and guessing Object produces a LinkedHashMap that fails at the handler's first field access instead.

Per-listener settings

@AceListener(
        queue = "${orders.queue}",
        id = "orders",
        prefetch = 40,
        concurrency = 4,
        autoStartup = "${orders.consume:true}")
void onOrder(Order order) { }

Concurrency above 1 gives up ordering. Messages on one queue are handled in parallel, which is usually what a service wants and occasionally the cause of a bug that only appears under load. When order matters per key rather than globally, the library's ordered queues do that properly; this annotation does not try to.

Two listeners cannot share an id. The clash fails at startup rather than leaving whichever one lost invisible in the registry.

When a handler throws

Returning normally acknowledges the message. Throwing rejects it, and what happens then is configuration:

The ladder is the reason to prefer this library, so it is worth being precise about what it does. A retry written by hand usually looks like this:

@AceListener(queue = "orders.new")
void onOrder(Order order) throws InterruptedException {
    try {
        downstream.call(order);
    } catch (TransientFailure e) {
        Thread.sleep(2000);       // do not do this
        downstream.call(order);
    }
}

That sleep blocks the consumer's channel. With prefetch 100, up to a hundred messages are sitting in that consumer's buffer, unavailable to any other consumer, waiting for a thread that is doing nothing. Ten seconds of backoff across four consumers is forty seconds of stalled throughput and a queue that looks busy while nothing moves.

The ladder republishes instead: the message goes back to the broker with a delay and an incremented attempt count, the channel is free, and the next message is handled now. Set it up in properties:

acemq:
  listener:
    retry:
      enabled: true
      max-attempts: 5
      initial-delay: 2s
      max-delay: 30s
      jitter: 0.2

and throw:

@AceListener(queue = "orders.new")
void onOrder(Order order) {
    downstream.call(order);   // throws, the ladder handles it
}

message.attempt() tells a handler which attempt it is on, which is how a fifth attempt can do something different from the first.

There is no return value that acknowledges. The library has an ack-aware handler interface that nothing in the library currently uses. An annotation wired to an unused code path is a starter growing a feature its own library does not have, so this one does not.

Retries, dead letters and replay is the whole of this subject: where a message goes when the ladder gives up, how to get it back, and how to survive the redelivery.

What the annotation does not reach

The annotation has six attributes and there will not be many more. Three things it cannot do, each with the same answer — declare that consumer as a ConsumerGroup bean and use the library's own API:

All three are ConsumerOptions, and the registry that starts annotated listeners builds those from acemq.listener.* alone — which is a property file, and none of the three is a value a property file can hold. Idempotency has the bean form in full, and Patterns is the map of which patterns need it.

Starting and stopping

Listeners are started by AceListenerRegistry, a SmartLifecycle, after the rest of the context is built. Starting a consumer while the beans it calls are still being created is how a message arrives at a half-built service, and it happens on exactly the deployments where the queue already has a backlog.

On shutdown the registry drains every listener before closing any of them:

WARN o.a.s.b.AceListenerRegistry : listener orders still had 3 message(s) in flight after PT30S; they will be redelivered

Draining is not the same as emptying the queue. A listener stopped with a backlog leaves the backlog; what it drains is the messages already inside handlers. Everything else stays on the queue.

A queue that does not exist

A listener on a queue nobody created is caught as the application starts, not at the broker's first refusal:

@AceListener orders (bean 'orderListener', method onOrder) listens on queue 'orders.nwe', which does not exist
Declare the queue under acemq.topology.queues or with @AceListener(declare = @AceQueue), create it on the broker, or set acemq.listener.missing-queue=warn to start without these listeners.

The check runs when listeners start, which is after the declared topology has been applied and after every bean that declares a queue in its own initialisation has run. A queue under acemq.topology.queues is taken as provided for and not asked about — unless acemq.topology.apply is dry-run, which creates nothing. Every other queue is asked about passively: nothing is created, and an existing queue is not changed whatever its type or arguments. Every missing queue is named in the one failure, so three typos cost one restart.

With acemq.listener.missing-queue: warn the application starts, each missing queue is logged as a warning, and those listeners stay stopped. Start one later with registry.start("orders"), or through the actuator endpoint; both check the queue again and refuse while it is still missing.

A listener with autoStartup = "false" is not checked until it is started.

Declaring the queue

A listener can declare the queue it consumes instead of relying on something else to:

@AceListener(queue = "orders.new", declare = @AceQueue)
void onOrder(Order order) { }

@AceListener(queue = "audit.log", declare = @AceQueue(
        type = Kind.CLASSIC,                       // AceMqProperties.Topology.Queue.Kind
        arguments = "x-message-ttl=${audit.ttl:60000}",
        deadLetter = true))
void onAudit(AuditLine line) { }

@AceQueue has the options of an entry under acemq.topology.queues, with the same names and defaults, and is declared through the same topology:

An argument value that is a whole number is sent as a number, and true or false as a boolean — x-message-ttl as the string "60000" is refused by the broker. Placeholders are resolved.

It follows acemq.topology.apply. With create-only the queue is created if it is missing and left alone if it is there; with validate it is only checked, and a missing one fails the startup; with dry-run nothing is declared and the listener is checked like one that declares nothing. The declaration happens as the listener starts, before the missing-queue check, so a self-declaring listener passes it — and a listener with autoStartup = "false" declares when it is started by id.

One queue, one declaration. A queue declared both here and under acemq.topology.queues, or by two listeners, must be declared the same way — the same type and the same arguments, dead-letter ones included — or the startup fails naming both:

@AceListener audit (bean 'auditListener', method onAudit) declares queue 'audit.log' differently from acemq.topology.queues: arguments {x-message-ttl=60000} here, {x-dead-letter-exchange=acemq.dlx, x-dead-letter-routing-key=audit.log.dlq, x-message-ttl=60000} there. Declare it the same way in both, or in one place only.

The broker would refuse the second declaration with PRECONDITION_FAILED anyway, but only when it got there, and closing the channel. When the two agree, the topology's declaration is the one applied. A queue already on the broker with different settings fails the same way, through the library's drift check — AMQP cannot change a queue in place.

The retry ladder is not part of it. With acemq.listener.retry.enabled the consumer declares its own rungs, .dlq and .parked as it starts, whether or not the listener declares its queue. What deadLetter adds is the broker-side half: the source queue's x-dead-letter-exchange, which the consumer cannot set because it does not declare the source queue. Use both together for a queue whose failures must never be dropped.

The registry

Inject AceListenerRegistry to reach a running listener:

@Service
class OrderThrottle {

    private final AceListenerRegistry listeners;

    OrderThrottle(AceListenerRegistry listeners) {
        this.listeners = listeners;
    }

    void slowDown() {
        listeners.get("orders").ifPresent(group -> {
            group.scaleTo(1);
            group.prefetch(10);
        });
    }

    long acknowledged() {
        return listeners.get("orders").map(ConsumerGroup::acknowledged).orElse(0L);
    }
}

get(id) returns the library's own ConsumerGroup, which is what carries scaleTo, prefetch, drainTimeout, drain, close and the counters — acknowledged, rejected, retried, inFlight, plus queue, size and prefetch as readers. It is a library type rather than a wrapper of this starter's, because a wrapper would be one more thing to keep in step.

Two things are not on it, and are worth knowing before looking for them.

Pausing. A group has no pause of its own. mq.pauseConsuming() and mq.resumeConsuming() stop and start every consumer on the connection at once, which is the right tool for shedding load and the wrong one for quietening a single noisy listener. scaleTo(1) is the floor — scaleTo(0) is refused with "a group needs at least one consumer" — so the way to stop one listener entirely is registry.stop("id"), and registry.start("id") to resume it.

Note also that scaling down blocks: the consumers being removed are drained first, up to drainTimeout. A scaleTo that returned instantly while abandoning work in the background would be the more convenient lie.

Dead letters. Not a group counter. acemq.messages.dead.lettered.total, tagged by queue, is the metric, and mq.messageCount("orders.new.dlq") is the depth — see Dead letters.

Every listener is a group, even at concurrency one. One type means one set of counters and one shutdown path, and a group of one costs nothing.

A listener registered with autoStartup = "false" is not running; registry.start("id") starts it. registry.stop("id") stops one — draining it first, as shutdown does — and start brings it back. The acemq actuator endpoint calls the same two methods.

There is no template

The publishing side has no annotation and no AceTemplate. Inject AceMq:

mq.publisher("orders", "order.created", Order.class).send(order);

A template wrapping that would be a second API to document, a second one to keep in step with the first, and the place where a starter starts making decisions the library deliberately left to the caller — which codec, which options, whether the send is synchronous. The publisher is already small, already typed, and already documented in the library's own guide.

Publishing is the page: keeping a publisher as a bean, what send returns, batches, and what happens when the broker stops reading.