Class Outbox

java.lang.Object
com.codename1.io.outbox.Outbox
public final class Outbox extends Object

A durable queue of outgoing messages, kept in the application's own database, for a device whose link to the server comes and goes.

The device records what it has to say with [#enqueue(String, byte[])] and sends it with flush(Transport) whenever the link might be up. Messages survive a restart or a power cut, leave in the order they were enqueued, and are removed only once the transport reports them delivered.

The table lives in the database of the Session the outbox was opened on, and an enqueue inside one of that session's transactions belongs to the transaction: a reading saved through the ORM and the message announcing it commit together or not at all. That is the point of keeping the queue there rather than in a file of its own.

Outbox outbox = Outbox.open(session, "readings@" + deviceId);
session.beginTransaction();
session.persist(reading);
outbox.enqueue("/v1/readings", body);
session.commitTransaction();

// Later, from a timer or when the network comes back. postToServer is the
// application's own call: Rest on a client, Web on the backend.
outbox.flush(message -> {
    int status = postToServer(message.getTopic(), message.getKey(), message.getPayload());
    return status < 300 ? Outbox.DELIVERED : status < 500 ? Outbox.REJECTED : Outbox.RETRY;
});

Delivery is at least once. A message is removed after the transport answers DELIVERED; if the answer never arrives the same message, with the same OutboxMessage.getKey(), is sent again, and the receiver drops the copy. Several devices feeding one server should put their identity in the outbox name, which is the first half of the key.

Like its session, an outbox is used from one thread at a time. The outbox works on the client and on the backend alike.

  • Nested Class Summary

    Nested Classes
    Modifier and Type
    Class
    Description
    static interface 
    Outbox.Transport
    Sends one message and says what became of it.

  • Field Summary

    Fields
    Modifier and Type
    Field
    Description
    static final int
    DELIVERED
    The receiver has the message; it is removed.
    static final int
    REJECTED
    The receiver will never take this message; it is dropped and counted in getRejected(), and the flush moves on to the next one.
    static final int
    RETRY
    Try again later.

  • Method Summary

    Modifier and Type
    Method
    Description
    OutboxMessage
    enqueue(String topic, byte[] payload)
    Adds a message to the end of the outbox.
    int
    flush(Outbox.Transport transport)
    Sends what is due, oldest first, until the outbox is empty, the head is waiting out a backoff, or a delivery has to be retried.
    long
    getRejected()
    Messages discarded so far by this outbox object: rejected by the transport, out of attempts, or dropped to make room.
    long
    nextAttemptTime()
    When the head of the outbox may next be sent, in milliseconds since the epoch, so a device can sleep until then; -1 when the outbox is empty.
    static Outbox
    open(Session session, String name)
    Opens an outbox, creating its table on first use.
    void
    setBackoff(long min, long max)
    Sets how long a message waits after a failed delivery: min after the first, doubling with each further failure up to max.
    void
    setCapacity(int messages, boolean dropOldest)
    Bounds the outbox.
    void
    setMaxAttempts(int attempts)
    Sets how many deliveries a message gets before it is treated as REJECTED.
    int
    size()
    The number of messages waiting.

    Methods inherited from class Object

    clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

  • Field Details

    • DELIVERED

      public static final int DELIVERED
      The receiver has the message; it is removed.
      See Also:

    • RETRY

      public static final int RETRY
      Try again later. The message stays at the head, and the flush stops so that nothing overtakes it.
      See Also:

    • REJECTED

      public static final int REJECTED
      The receiver will never take this message; it is dropped and counted in getRejected(), and the flush moves on to the next one.
      See Also:

  • Method Details

    • open

      public static Outbox open(Session session, String name)
      Opens an outbox, creating its table on first use. Outboxes of different names share the table and never see each other's messages.
      Parameters:
      session - a session from EntityManager.openSession()
      name - the outbox name, which also starts every message key
      Returns:
      the outbox
      Throws:
      PersistenceException - if the table cannot be created

    • setBackoff

      public void setBackoff(long min, long max)
      Sets how long a message waits after a failed delivery: min after the first, doubling with each further failure up to max. The defaults are one second and five minutes.
      Parameters:
      min - the first delay in milliseconds, 0 or more
      max - the longest delay in milliseconds, at least min

    • setMaxAttempts

      public void setMaxAttempts(int attempts)
      Sets how many deliveries a message gets before it is treated as REJECTED. 0, the default, retries forever.
      Parameters:
      attempts - the limit, or 0 for none

    • setCapacity

      public void setCapacity(int messages, boolean dropOldest)
      Bounds the outbox. When it is full, dropOldest makes room by discarding the oldest messages, which suits telemetry where the newest reading matters most; otherwise [#enqueue(String, byte[])] refuses the message. 0, the default, is unbounded.
      Parameters:
      messages - the most messages kept, or 0 for no bound
      dropOldest - discard the oldest instead of refusing the newest

    • enqueue

      public OutboxMessage enqueue(String topic, byte[] payload)
      Adds a message to the end of the outbox. Inside a transaction of the outbox's session the message commits or rolls back with it; otherwise it is committed before this returns.
      Parameters:
      topic - what the transport routes on, never null
      payload - the message body, copied
      Returns:
      the message as it was stored
      Throws:
      PersistenceException - if it cannot be stored, or the outbox is full and set to refuse

    • size

      public int size()
      The number of messages waiting.
      Returns:
      the count

    • nextAttemptTime

      public long nextAttemptTime()
      When the head of the outbox may next be sent, in milliseconds since the epoch, so a device can sleep until then; -1 when the outbox is empty.
      Returns:
      the time, or -1

    • getRejected

      public long getRejected()
      Messages discarded so far by this outbox object: rejected by the transport, out of attempts, or dropped to make room.
      Returns:
      the count since open(Session, String)

    • flush

      public int flush(Outbox.Transport transport)

      Sends what is due, oldest first, until the outbox is empty, the head is waiting out a backoff, or a delivery has to be retried. Call it whenever the link may be up; a flush that cannot send costs one query.

      Not inside a transaction: each delivery is network I/O, and each removal commits by itself so that a crash mid-flush repeats at most one message.

      Parameters:
      transport - sends one message
      Returns:
      how many messages were delivered
      Throws:
      IllegalStateException - inside a transaction of the session