Package org.acemq.amqp.transport

Interface TransportConnection

All Superinterfaces:
AutoCloseable
public interface TransportConnection extends AutoCloseable
An open connection to one broker.

Implementations must be safe to use from several threads. Whatever multiplexing the protocol requires — AMQP 0-9-1 channels, AMQP 1.0 sessions and links — is the transport's business and is never exposed, because a caller holding a channel is a caller who breaks when the transport changes.

  • Method Details

    • declareExchange

      void declareExchange(String name, String type, boolean durable)
      Declares an exchange, or verifies that an equivalent one exists.
      Parameters:
      name - exchange name
      type - routing behaviour: direct, topic, fanout or headers
      durable - whether it survives a broker restart
      Throws:
      TransportException - if an incompatible exchange already exists

    • declareQueue

      void declareQueue(String name, QueueType type, boolean durable, Map<String,Object> arguments)
      Declares a queue, or verifies that an equivalent one exists.
      Parameters:
      name - queue name
      type - queue implementation to request
      durable - whether it survives a broker restart
      arguments - broker-specific queue arguments, may be empty
      Throws:
      TransportException - if an incompatible queue already exists. AMQP forbids changing most arguments in place, so this is reported rather than worked around.

    • bindQueue

      void bindQueue(String queue, String exchange, String routingKey)
      Binds a queue to an exchange.
      Parameters:
      queue - queue name
      exchange - exchange name
      routingKey - routing key or pattern

    • send

      ConfirmResult send(OutboundMessage message)
      Publishes one message and waits for the broker to confirm it.

      Blocking is deliberate. An asynchronous publish that nobody waits on is the most common way to lose messages, so the safe shape is the default one; batching and pipelining belong to the core, which knows when they are appropriate.

      Parameters:
      message - the message to publish
      Returns:
      what the broker said, including whether anything was bound to receive it
      Throws:
      TransportException - if the connection failed or the confirm timed out

    • subscribe

      Subscription subscribe(String queue, int prefetch, DeliveryListener listener)
      Starts consuming a queue.
      Parameters:
      queue - queue to consume
      prefetch - maximum unsettled deliveries allowed at once; must be at least 1, and is the only backpressure that exists, so it is required rather than defaulted
      listener - receives each delivery
      Returns:
      a handle that stops delivery when closed
      Throws:
      TransportException - if the queue cannot be consumed

    • subscribe

      default Subscription subscribe(String queue, int prefetch, Map<String,Object> consumerArguments, DeliveryListener listener)
      Starts consuming a queue with broker-specific consumer arguments.

      Exists for streams. A stream consumer says where in the log to start with an x-stream-offset argument, and there is nowhere else to put it: it belongs to the subscription rather than to the queue, because two consumers of one stream read from different positions.

      The default refuses rather than ignores. Dropping the arguments would leave a consumer that asked to replay a stream from the beginning quietly reading only new messages — working, plausible, and missing everything it was written to process.

      Parameters:
      queue - queue to consume
      prefetch - maximum unsettled deliveries allowed at once
      consumerArguments - broker-specific arguments for this subscription
      listener - receives each delivery
      Returns:
      a handle that stops delivery when closed
      Throws:
      TransportException - if the queue cannot be consumed, or if this transport cannot honour the arguments given

    • sendAsync

      default CompletableFuture<ConfirmResult> sendAsync(OutboundMessage message)
      Publishes one message without waiting for the broker to confirm it.

      The throughput answer to send(org.acemq.amqp.transport.OutboundMessage). A synchronous publish costs a full round trip per message, so a loop over ten thousand messages spends nearly all of its time waiting; this lets the next message go out while the last one is still being confirmed. Confirms come back out of order and in ranges, which is why the correlation is the transport's job rather than the caller's.

      The safety properties are unchanged, and that is the point: the returned future carries the same ConfirmResult the synchronous call returns, including whether anything was bound to receive the message. What changes is when the caller learns it. A future nobody waits on is a message nobody knows the fate of — which is the failure mode this whole library exists to avoid, so the result must be checked, eventually, by somebody.

      Implementations must bound the number of unconfirmed publishes and block once that limit is reached. An unbounded async publisher is a memory leak that looks like throughput.

      The default implementation publishes synchronously and hands back a completed future. It is correct and offers no speed-up whatsoever, which is the honest behaviour for a transport that cannot pipeline.

      Parameters:
      message - the message to publish
      Returns:
      a future completed with the broker's answer
      Throws:
      TransportException - if the message could not be written at all

    • receive

      default Optional<PulledMessage> receive(String queue, Duration timeout)
      Takes one message off a queue, or reports that there is none.

      A pull, where subscribe(java.lang.String, int, org.acemq.amqp.transport.DeliveryListener) is a push. It exists for the jobs that have to stop when a queue runs out — draining a dead-letter queue is the one this was written for. A subscription cannot do that: it is told when a message arrives and never told that no more are coming, so a tool built on one has to guess with an idle timeout and either stops early or hangs on an empty queue.

      Not for ordinary consuming. Pulling one message per round trip is far slower than a subscription with prefetch, and a loop calling this is a consumer that has thrown away its own backpressure.

      The delivery is unsettled: the caller must accept or reject it through the Acknowledger, exactly as a subscription must.

      Parameters:
      queue - queue to take from
      timeout - how long to wait for a message to appear
      Returns:
      the message and its acknowledger, or empty when the queue had nothing within the timeout
      Throws:
      TransportException - if the queue cannot be read

    • messageCount

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

      For reporting what a drain would move before it moves anything. It is a snapshot and nothing more: a live queue has a different depth by the time the number is read, so it belongs in a report to a person, never in a loop condition.

      Parameters:
      queue - queue to measure
      Returns:
      how many messages are waiting
      Throws:
      TransportException - if the queue cannot be measured or does not exist

    • isBlocked

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

      Worth exposing rather than only throwing, because this is the state a health check should report and an autoscaler should read. A blocked broker is not a broken one, and a service that reports itself unhealthy the moment a memory alarm fires will be restarted by its orchestrator, which helps nobody.

      Returns:
      whether publishing is currently blocked at the broker

    • blockedReason

      default Optional<String> blockedReason()
      Returns:
      the broker's stated reason for blocking, when it is blocked

    • deleteQueue

      void deleteQueue(String name)
      Removes a queue and everything in it.
      Parameters:
      name - queue to delete
      Implementation Note:
      provided for test fixtures and topology migration. Ordinary application code should not be deleting queues.

    • queueExists

      boolean queueExists(String name)
      Reports whether a queue exists, without creating or altering it.

      Separate from declaring because a planner has to be able to look before it touches anything. Declaring a queue that already exists with different arguments does not merely fail: on AMQP 0-9-1 it kills the channel, which is why a topology applied blindly at start-up fails in production rather than in review.

      Parameters:
      name - queue name
      Returns:
      true if the queue is already present

    • checkQueue

      default QueueCheck checkQueue(String name, QueueType type, boolean durable, Map<String,Object> arguments)
      Asks whether a queue that already exists would accept this declaration.

      queueExists(String) answers "is something there", which is the smaller half of the question. The half that costs an outage is "is what is there the same as what this release expects" — a queue whose x-message-ttl was changed by hand, or whose dead-letter exchange was pointed somewhere else, or that somebody created as classic when the topology says quorum. AMQP forbids changing any of those in place, so the declare fails at start-up, in production, taking the channel with it.

      The default answers QueueCheck.unsupported(). A transport that cannot inspect a queue must not claim it matches: silence read as agreement is how drift gets missed.

      Parameters:
      name - queue name
      type - queue implementation the caller wants
      durable - whether the caller wants it to survive a restart
      arguments - the arguments the caller would declare
      Returns:
      what the broker says, including QueueCheck.Result.UNSUPPORTED when it cannot be asked
      Implementation Requirements:
      must not create the queue, and must not disturb traffic on other channels. The usual implementation asks passively first and, only for a queue that is already there, attempts an equivalence declare on a channel of its own.

    • isOpen

      boolean isOpen()
      Returns:
      whether the connection is usable

    • close

      void close()
      Closes the connection, letting in-flight deliveries settle first.
      Specified by:
      close in interface AutoCloseable