Class DeliveryOptions

java.lang.Object
org.mqtt4j.hato.DeliveryOptions
public final class DeliveryOptions extends Object

Where and how a push subscription runs its handler, for MqttSubscriber.subscribe(Subscribe, DeliveryOptions, java.util.function.Consumer). Instances are immutable and can be shared between subscriptions; they do not define equality.

serial() runs one delivery at a time, in receive order, on an internal thread, as MqttSubscriber.subscribe(Subscribe, java.util.function.Consumer) does. withExecutor(Executor) keeps that order but runs each handler on the application's executor. lanes(Function) runs each delivery on the executor that the application chooses for its message, its lane: deliveries on the same lane keep that order, and different lanes can run at the same time. striped(int) keeps that order for deliveries with equal keys, by default their topics, and runs different keys at the same time on a fixed number of stripes; each stripe is a lane in what follows. parallel(int) runs up to a given number of deliveries at the same time in no particular order, as one lane that runs that many at once. inline() runs one delivery at a time, in receive order, for a handler that returns at once: the client may run it as part of receiving, and the connection waits for it meanwhile. Any other handler on an internal thread may block: meanwhile only its own lane waits.

An executor that runs handlers receives at most one task per subscription and lane at a time, or, for parallel(int), as many as its parallelism. A task runs one delivery's handler and, while further deliveries wait for the same lane, asks the executor again, so the executor's queue holds no more tasks than that, and waiting deliveries count against Subscribe.bufferCapacity() as with an internal thread. While no delivery waits, the subscription holds none of the executor's threads. The executor must run each task it accepts, or throw from Executor.execute(Runnable): an exception of any type ends the delivery that the task would have run as a handler failure under FailurePolicy.STOP, whatever the subscription's policy, since a refusing executor has processed nothing. An executor that silently discards a task leaves the subscription waiting until it closes. An executor that runs tasks on the calling thread, such as a direct executor or one with a caller-runs policy, keeps the order and the capacity without deepening the stack.

A failure that stops the client, on any lane, stops the subscription's other lanes too. Failures on other lanes that were already running are logged, and the first failure remains the cause that the client reports.

The application owns the executor; the client never shuts it down. Once the subscription or the client has closed, a task that the executor still holds does nothing when it runs, so the tasks that ExecutorService.shutdownNow() returns need not run, and neither Subscription.terminated() nor MqttClient.terminated() waits for them. They do wait for a handler still running on the executor.

Acknowledgement follows the request's AckMode wherever the handler runs: AckMode.ON_RECEIVE just before the handler starts on its executor, and AckMode.AFTER_HANDLER when it returns there. PUBACK and PUBREC still follow receive order, so a slow handler holds back the acknowledgement of later deliveries, on other lanes too; see AckMode.

  • Method Details

    • serial

      public static DeliveryOptions serial()
      One delivery at a time, in receive order, on an internal thread: the behavior of MqttSubscriber.subscribe(Subscribe, java.util.function.Consumer).
      Returns:
      the options for serial delivery

    • lanes

      public static DeliveryOptions lanes(Function<MqttMessage, Executor> selector)

      Runs each delivery's handler on the executor that selector returns for its message: the delivery's lane. Deliveries on the same lane run one at a time in receive order. Lanes are told apart by identity, so return the same executor for the same lane; a new executor object for every message gives every message a lane of its own.

      The selector runs on an internal thread, once for each delivery in receive order. It receives the message rather than the delivery, so it cannot acknowledge it. An exception or a null result ends that delivery as a handler failure.

      Parameters:
      selector - the function that chooses the executor for each message
      Returns:
      the options for the lanes that `selector` chooses
      Throws:
      NullPointerException - 
      if `selector` is null

    • striped

      public static DeliveryOptions striped(int stripes)
      Runs deliveries with equal topics one at a time in receive order, and deliveries with different topics possibly at the same time, at most stripes at once. Each topic's deliveries go to one of stripes fixed stripes that its hash chooses, so different topics can share a stripe and then wait for each other. A stripe runs on an internal thread, or on the executor that withExecutor(Executor) sets.
      Parameters:
      stripes - the number of stripes
      Returns:
      the options for `stripes` stripes keyed by topic
      Throws:
      IllegalArgumentException - 
      if `stripes` is less than 1 or greater than 256

    • striped

      public static DeliveryOptions striped(int stripes, Function<MqttMessage, ?> key)
      Like striped(int), with the key that key returns for each message in place of its topic; keys are compared with equals and hashCode. The key function runs on an internal thread, once for each delivery in receive order. It receives the message rather than the delivery, so it cannot acknowledge it. An exception or a null result ends that delivery as a handler failure.
      Parameters:
      stripes - the number of stripes
      key - the function that returns the key of each message
      Returns:
      the options for `stripes` stripes keyed by `key`
      Throws:
      IllegalArgumentException - 
      if `stripes` is less than 1 or greater than 256
      NullPointerException - 
      if `key` is null

    • parallel

      public static DeliveryOptions parallel(int parallelism)
      Runs up to parallelism deliveries at the same time, in no particular order. Handlers run on internal threads, or on the executor that withExecutor(Executor) sets.
      Parameters:
      parallelism - the largest number of deliveries that run at the same time
      Returns:
      the options for up to `parallelism` deliveries at the same time
      Throws:
      IllegalArgumentException - 
      if `parallelism` is less than 1 or greater than 256

    • inline

      public static DeliveryOptions inline()

      One delivery at a time, in receive order, as serial(), for a handler that returns at once and never waits for this client. In return, the client may run the handler as part of receiving its message, rather than hand the delivery to another thread. While such a handler runs, the connection may read nothing else: neither messages for other subscriptions nor the broker's answers to publications, subscriptions, and keep-alive requests.

      Inside the handler, join() and the get methods of a future that this client returned, or of a future made from one, and the receive methods of this client's pull subscriptions throw IllegalStateException, even when they would not wait. Read a completed result with getNow, or chain the next step instead. Other waits are not detected: a handler that blocks, or that waits for anything that needs this connection, holds up every subscription of the connection, and one that runs for about the keep-alive interval can lose the connection.

      A delivery that the client cannot run as part of receiving runs on an internal thread, in the same order and under the same rules. The subscription's buffer, acknowledgement mode, and failure policy work as for serial(). withExecutor(Executor) rejects these options.

      Returns:
      the options for inline delivery

    • withExecutor

      public DeliveryOptions withExecutor(Executor executor)
      Keeps the order, stripes, or parallelism of these options, with each handler running on executor instead of an internal thread. Replaces an executor chosen earlier.
      Parameters:
      executor - the executor that runs the handlers
      Returns:
      options like these that run each handler on `executor`
      Throws:
      NullPointerException - 
      if `executor` is null
      IllegalArgumentException - 
      if these options are [#lanes(Function)], whose selector chooses the executors, or
[#inline()], whose handler runs as part of receiving