Class RedisStreamTransport

java.lang.Object

com.codename1.backend.messaging.MessageTransport

com.codename1.backend.messaging.PollingTransport

com.codename1.backend.messaging.redis.RedisStreamTransport

All Implemented Interfaces:

Closeable, AutoCloseable

public final class RedisStreamTransport
extends PollingTransport

Redis Streams: cn1.messaging.url=redis://... (or rediss://). Works with Redis 6.2 or newer, Valkey and KeyDB.

A destination is a stream and a group a consumer group, created on first use: each group gets every message, and the consumers of one group share them. A message is an entry with its payload, key and headers as fields; acknowledging it is XACK. A consumer that dies leaves its messages pending, and another consumer of the group claims them once they have been idle for cn1.messaging.lease milliseconds (60000) -- counting that as another delivery, so a message that crashes its consumer is dead-lettered like one that throws.

A failed message stays pending, and its next delivery waits in a sorted set per destination and group until it is due; only the group whose listener failed sees it again.

Streams are capped at about cn1.messaging.redis.max-length entries (100000, trimmed with MAXLEN ~), so a destination nobody reads cannot grow without bound.

Nested Class Summary

Nested Classes

Modifier and Type	Class	Description
static final class	RedisStreamTransport.Factory	Opens the transport for redis:// and rediss:// URLs; the generated wiring registers it when the configuration names one.

Nested classes/interfaces inherited from class PollingTransport

PollingTransport.Delivery

Field Summary

Fields inherited from class PollingTransport

retry

Constructor Summary

Constructors

Constructor	Description
RedisStreamTransport(RedisKeyValueStore redis, RetryPolicy retry, long leaseMillis, long maxLength)	A transport over redis, which it closes.

Method Summary

Modifier and Type	Method	Description
protected void	ack(String destination, String group, PollingTransport.Delivery delivery)	Acknowledges a handled delivery.
void	close()	Stops every subscription and closes the connection.
String	getName()	The scheme this transport was opened for: memory, db, redis, amqp, kafka.
boolean	isHealthy()	Whether the broker answers, for the health endpoint.
protected List<PollingTransport.Delivery>	poll(String destination, String group, int max, long waitMillis)	Up to max messages of destination for group, waiting up to waitMillis for the first; an empty list when none came.
protected void	redeliver(String destination, String group, PollingTransport.Delivery delivery, int attempt, long notBefore)	Has delivery delivered again, as attempt, no sooner than notBefore.
void	send(Message message)	Sends message.

Methods inherited from class PollingTransport

deadLetter, identified, notYet, stopSubscriptions, subscribe

Methods inherited from class MessageTransport

isTransactional

Methods inherited from class Object

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

Constructor Details

RedisStreamTransport

public RedisStreamTransport(RedisKeyValueStore redis,
 RetryPolicy retry,
 long leaseMillis,
 long maxLength)

A transport over redis, which it closes.

Method Details

getName

public String getName()

Description copied from class: MessageTransport

The scheme this transport was opened for: memory, db, redis, amqp, kafka.

Specified by:

getName in class MessageTransport

send

public void send(Message message)
          throws IOException

Description copied from class: MessageTransport

Sends message. With a database transport inside a transaction the message is written in that transaction and is seen only if it commits.

Specified by:

send in class MessageTransport

Throws:

IOException

poll

protected List<PollingTransport.Delivery> poll(String destination,
 String group,
 int max,
 long waitMillis)
                                        throws IOException

Description copied from class: PollingTransport

Up to max messages of destination for group, waiting up to waitMillis for the first; an empty list when none came.

Specified by:

poll in class PollingTransport

Throws:

IOException

ack

protected void ack(String destination,
 String group,
 PollingTransport.Delivery delivery)
            throws IOException

Description copied from class: PollingTransport

Acknowledges a handled delivery.

Specified by:

ack in class PollingTransport

Throws:

IOException

redeliver

protected void redeliver(String destination,
 String group,
 PollingTransport.Delivery delivery,
 int attempt,
 long notBefore)
                  throws IOException

Description copied from class: PollingTransport

Has delivery delivered again, as attempt, no sooner than notBefore. The default sends a copy carrying the attempt and the time in its headers and acknowledges the original; a consumer that polls the copy early puts it back with PollingTransport.notYet(Message, long).

Overrides:

redeliver in class PollingTransport

Throws:

IOException

isHealthy

public boolean isHealthy()

Description copied from class: MessageTransport

Whether the broker answers, for the health endpoint. Never throws.

Overrides:

isHealthy in class MessageTransport

close

public void close()

Description copied from class: MessageTransport

Stops every subscription and closes the connection. Never throws.

Specified by:

close in interface AutoCloseable

Specified by:

close in interface Closeable

Specified by:

close in class MessageTransport