Class 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 Class Summary
Nested ClassesModifier and TypeClassDescriptionstatic interfaceSends one message and says what became of it. -
Field Summary
FieldsModifier and TypeFieldDescriptionstatic final intThe receiver has the message; it is removed.static final intThe receiver will never take this message; it is dropped and counted ingetRejected(), and the flush moves on to the next one.static final intTry again later. -
Method Summary
Modifier and TypeMethodDescriptionAdds a message to the end of the outbox.intflush(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.longMessages discarded so far by this outbox object: rejected by the transport, out of attempts, or dropped to make room.longWhen 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 OutboxOpens an outbox, creating its table on first use.voidsetBackoff(long min, long max) Sets how long a message waits after a failed delivery:minafter the first, doubling with each further failure up tomax.voidsetCapacity(int messages, boolean dropOldest) Bounds the outbox.voidsetMaxAttempts(int attempts) Sets how many deliveries a message gets before it is treated asREJECTED.intsize()The number of messages waiting.
-
Field Details
-
DELIVERED
public static final int DELIVEREDThe receiver has the message; it is removed.- See Also:
-
RETRY
public static final int RETRYTry again later. The message stays at the head, and the flush stops so that nothing overtakes it.- See Also:
-
REJECTED
public static final int REJECTEDThe receiver will never take this message; it is dropped and counted ingetRejected(), and the flush moves on to the next one.- See Also:
-
-
Method Details
-
open
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 fromEntityManager.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:minafter the first, doubling with each further failure up tomax. The defaults are one second and five minutes.- Parameters:
min- the first delay in milliseconds, 0 or moremax- the longest delay in milliseconds, at leastmin
-
setMaxAttempts
public void setMaxAttempts(int attempts) Sets how many deliveries a message gets before it is treated asREJECTED. 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,dropOldestmakes 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 bounddropOldest- discard the oldest instead of refusing the newest
-
enqueue
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 nullpayload- 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
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
-