Client and backend. This class is shared: the same code is compiled into your app and into your server, so everything on this page works in both.
public final class Outbox
- Object
- Outbox
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 types
interface Outbox.Transport | Sends one message and says what became of it. |
Fields
public static final int DELIVERED = 0 | The receiver has the message; it is removed. |
public static final int RETRY = 1 | Try again later. |
public static final int REJECTED = 2 | The receiver will never take this message; it is dropped and counted in getRejected(), and the flush moves on to the next one. |
Methods
public static Outbox open(Session session, String name) | Opens an outbox, creating its table on first use. |
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. |
public void setMaxAttempts(int attempts) | Sets how many deliveries a message gets before it is treated as REJECTED. |
public void setCapacity(int messages, boolean dropOldest) | Bounds the outbox. |
public OutboxMessage enqueue(String topic, byte[] payload) | Adds a message to the end of the outbox. |
public int size() | The number of messages waiting. |
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. |
public long getRejected() | Messages discarded so far by this outbox object: rejected by the transport, out of attempts, or dropped to make room. |
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. |
Inherited methods
Field details
DELIVERED
public static final int DELIVERED = 0RETRY
public static final int RETRY = 1REJECTED
public static final int REJECTED = 2getRejected(), and the flush moves on to the next one.Method details
open
public static Outbox open(Session session, String name)Parameters
sessionSession- a session from
EntityManager.openSession() nameString- the outbox name, which also starts every message key
Returns
Throws
PersistenceException- if the table cannot be created
setBackoff
public void setBackoff(long min, long max)min after the
first, doubling with each further failure up to max. The defaults are
one second and five minutes.Parameters
minlong- the first delay in milliseconds, 0 or more
maxlong- the longest delay in milliseconds, at least
min
setMaxAttempts
public void setMaxAttempts(int attempts)REJECTED. 0, the default, retries forever.Parameters
attemptsint- the limit, or 0 for none
setCapacity
public void setCapacity(int messages, boolean dropOldest)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
messagesint- the most messages kept, or 0 for no bound
dropOldestboolean- discard the oldest instead of refusing the newest
enqueue
public OutboxMessage enqueue(String topic, byte[] payload)Parameters
topicString- what the transport routes on, never null
payloadbyte[]- the message body, copied
Returns
Throws
PersistenceException- if it cannot be stored, or the outbox is full and set to refuse
size
public int size()Returns
nextAttemptTime
public long nextAttemptTime()Returns
getRejected
public long getRejected()Returns
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
transportOutbox.Transport- sends one message
Returns
Throws
IllegalStateException- inside a transaction of the session