Package org.acemq.amqp.core

Class DefaultPublisher<T>

java.lang.Object
org.acemq.amqp.core.DefaultPublisher<T>
Type Parameters:
T - payload type
All Implemented Interfaces:
AutoCloseable, Publisher<T>
public final class DefaultPublisher<T> extends Object implements Publisher<T>
Publishes to one destination, waiting for the broker to confirm every message.

Three things are non-negotiable here, because each is a way messages get lost quietly: the publish waits for a confirm, an unroutable message is an error rather than a shrug, and the envelope is stamped onto the message so the consumer can tell what it received.

Choosing a format

A publisher writes one format, chosen when it is built: 


 Publisher<Order> orders = mq.publisher("orders", "order.placed", Order.class);          // JSON
 Publisher<Order> legacy = mq.publisher("legacy", "order", Order.class).asXml();
 Publisher<byte[]> files = mq.publisher("files", "file.new", byte[].class).asBytes();
 Publisher<Order> avro   = mq.publisher("events", "order", Order.class).as(new AvroCodec(registry));

Chosen at the publisher rather than per message, and that is a decision rather than an omission. A queue whose messages are sometimes JSON and sometimes XML is a queue nobody can write a consumer against, so the useful place to decide is once, where the destination is named. It also has to be here for a duller reason: by the time send has returned there is nothing left to choose, because the message is already at the broker.

  • Method Details

    • replyingTo

      public DefaultPublisher<T> replyingTo(String replyQueue)
      Names the queue a reply to this message should be sent to.

      Used by Requester; publishing this by hand means taking on the correlation and the timeout yourself, which is what Requester exists to avoid.

      Parameters:
      replyQueue - queue the responder should answer on
      Returns:
      a publisher that asks for a reply

    • with

      public DefaultPublisher<T> with(PublishOptions options)
      The same destination, published differently.

      A new publisher rather than a mutated one, for the same reason as(Codec) is: two threads sharing a publisher must not have its behaviour changed underneath them.

      Parameters:
      options - how these messages should be published
      Returns:
      a publisher using those options; this one is left alone

    • options

      public PublishOptions options()
      Returns:
      how this publisher publishes

    • as

      public DefaultPublisher<T> as(String format)
      The same destination, written in a named format.
      Parameters:
      format - short format name, such as json or xml
      Returns:
      a publisher writing that format; this one is left alone
      Throws:
      AceMqException - if no module on the classpath provides that format

    • as

      public DefaultPublisher<T> as(Codec format)
      The same destination, written with a codec of the caller's own.

      How a format AceMQ does not ship is used: implement Codec and pass it. Avro and Protobuf belong here rather than behind a name of their own, because neither can be built without a schema and a method taking no arguments would only be able to fail.

      Parameters:
      format - the codec to write with
      Returns:
      a publisher writing that format; this one is left alone

    • asJson

      public DefaultPublisher<T> asJson()
      Returns:
      the same destination, written as JSON; what a publisher does anyway unless the connection was given a different codec

    • asXml

      public DefaultPublisher<T> asXml()
      Returns:
      the same destination, written as XML
      Throws:
      AceMqException - unless org.acemq:acemq-amqp-codec-xml is on the classpath

    • asYaml

      public DefaultPublisher<T> asYaml()
      Returns:
      the same destination, written as YAML, for messages a person will read as well as a program
      Throws:
      AceMqException - unless org.acemq:acemq-amqp-codec-yaml is on the classpath

    • asText

      public DefaultPublisher<T> asText()
      Returns:
      the same destination, written as UTF-8 text; for payloads that are already strings

    • asBytes

      public DefaultPublisher<T> asBytes()
      Returns:
      the same destination, written as raw bytes, for payloads something else has already encoded

    • codec

      public Codec codec()
      Returns:
      the format this publisher writes

    • sendAsync

      public CompletableFuture<PublishResult> sendAsync(T payload)
      Description copied from interface: Publisher
      Publishes without waiting for the broker to confirm.

      A synchronous publish costs a round trip per message, so a loop over ten thousand of them spends nearly all its time waiting. This lets the next message go out while the last is still in flight, and is typically an order of magnitude faster for bulk work.

      Nothing about the guarantees changes: the future carries the same result Publisher.send(Object) returns, and fails the same way — including when nothing was bound to receive the message. What changes is when you find out.

      A future nobody waits on is a message nobody knows the fate of. That is the failure this library exists to prevent, so somebody has to check, eventually. If you do not intend to look at the result, use Publisher.send(Object) and take the round trip. 

      
 List<CompletableFuture<PublishResult>> inFlight = new ArrayList<>();
 for (Order order : thousands) {
     inFlight.add(publisher.sendAsync(order));
 }
 CompletableFuture.allOf(inFlight.toArray(new CompletableFuture[0])).join();

      Publishing blocks once too many messages are awaiting confirmation, which is deliberate: an asynchronous publisher with no ceiling is a memory leak that looks like throughput.

      Specified by:
      sendAsync in interface Publisher<T>
      Parameters:
      payload - the payload to send
      Returns:
      a future completed with the broker's answer, or failed as Publisher.send(Object) would have thrown

    • sendAsync

      public CompletableFuture<PublishResult> sendAsync(T payload, Envelope envelope)
      Description copied from interface: Publisher
      Publishes without waiting, with an envelope of your own.
      Specified by:
      sendAsync in interface Publisher<T>
      Parameters:
      payload - the payload to send
      envelope - identity, correlation and headers to travel with it
      Returns:
      a future completed with the broker's answer

    • sendAll

      public List<PublishResult> sendAll(Collection<? extends T> payloads)
      Description copied from interface: Publisher
      Publishes a batch and waits for every confirm.

      What most bulk publishing actually wants: the throughput of pipelining with the safety of having waited. Every message goes out before any confirm is awaited, then all of them are checked together.

      Fails if any message failed, and the exception names how many succeeded — a partial batch is the normal outcome of a broker problem halfway through, and pretending otherwise would leave the caller resending messages that already arrived.

      This is not atomic. AMQP has no such thing: there is no way to publish a hundred messages such that all or none arrive, and a library that offered one would be lying.

      Specified by:
      sendAll in interface Publisher<T>
      Parameters:
      payloads - the payloads to send, in order
      Returns:
      the results, in the same order

    • send

      public PublishResult send(T payload)
      Description copied from interface: Publisher
      Publishes a message and waits for the broker to confirm it.
      Specified by:
      send in interface Publisher<T>
      Parameters:
      payload - the payload to send
      Returns:
      the confirmed result

    • send

      public PublishResult send(T payload, Envelope envelope)
      Description copied from interface: Publisher
      Publishes a message built on an explicit envelope.

      Used when correlation or causation must be carried from a message being handled, so that a whole flow can be followed in traces and logs.

      Specified by:
      send in interface Publisher<T>
      Parameters:
      payload - the payload to send
      envelope - metadata to publish it with
      Returns:
      the confirmed result

    • close

      public void close()
      Specified by:
      close in interface AutoCloseable
      Specified by:
      close in interface Publisher<T>

    • toString

      public String toString()
      Overrides:
      toString in class Object