Class StreamEnvironmentBuilder

java.lang.Object

com.rabbitmq.stream.impl.StreamEnvironmentBuilder

All Implemented Interfaces:

EnvironmentBuilder

public class StreamEnvironmentBuilder
extends Object
implements EnvironmentBuilder

Nested Class Summary

Nested classes/interfaces inherited from interface EnvironmentBuilder

EnvironmentBuilder.NettyConfiguration, EnvironmentBuilder.OAuth2Configuration, EnvironmentBuilder.TlsConfiguration

Constructor Summary

Constructors

Constructor	Description
StreamEnvironmentBuilder()

Method Summary

Modifier and Type	Method	Description
EnvironmentBuilder	addressResolver(AddressResolver addressResolver)	An AddressResolver to potentially change resolved node address to connect to.
Environment	build()	Create the Environment instance.
StreamEnvironmentBuilder	chunkChecksum(ChunkChecksum chunkChecksum)	The checksum strategy used for chunk checksum.
StreamEnvironmentBuilder	clientProperties(Map<String,String> clientProperties)	Custom client properties to add to default client properties.
StreamEnvironmentBuilder	clientProperty(String key, String value)	Add a custom client properties to default client properties.
StreamEnvironmentBuilder	codec(Codec codec)	The AMQP 1.0 codec used to encode and decode AMQP 1.0 messages.
EnvironmentBuilder	compressionCodecFactory(CompressionCodecFactory compressionCodecFactory)	Compression codec factory to use for compression in sub-entry batching.
StreamEnvironmentBuilder	credentialsProvider(CredentialsProvider credentialsProvider)	The CredentialsProvider to use.
EnvironmentBuilder	forceLeaderForProducers(boolean forceLeader)	Flag to force the connection to the stream leader for producers.
EnvironmentBuilder	forceReplicaForConsumers(boolean forceReplica)	Flag to force the connection to a stream replica for consumers.
StreamEnvironmentBuilder	host(String host)	The host to connect to.
EnvironmentBuilder	id(String id)	Informational ID for this environment instance.
EnvironmentBuilder	lazyInitialization(boolean lazy)	To delay the connection opening until necessary.
StreamEnvironmentBuilder	locatorConnectionCount(int locatorCount)	Set the expected number of "locator" connections to maintain.
EnvironmentBuilder	maxConsumersByConnection(int maxConsumersByConnection)	The maximum number of consumers allocated to a single connection.
EnvironmentBuilder	maxProducersByConnection(int maxProducersByConnection)	The maximum number of producers allocated to a single connection.
EnvironmentBuilder	maxTrackingConsumersByConnection(int maxTrackingConsumersByConnection)	The maximum number of tracking consumers allocated to a single connection.
StreamEnvironmentBuilder	metricsCollector(MetricsCollector metricsCollector)	Set up a MetricsCollector.
EnvironmentBuilder.NettyConfiguration	netty()	Helper to configure netty.
EnvironmentBuilder.OAuth2Configuration	oauth2()	OAuth 2 settings.
EnvironmentBuilder	observationCollector(ObservationCollector<?> observationCollector)	Set up an ObservationCollector.
StreamEnvironmentBuilder	password(String password)	The password to use.
StreamEnvironmentBuilder	port(int port)	The port to use to connect.
EnvironmentBuilder	recoveryBackOffDelayPolicy(BackOffDelayPolicy recoveryBackOffDelayPolicy)	The BackOffDelayPolicy to use for connection recovery.
StreamEnvironmentBuilder	requestedHeartbeat(Duration requestedHeartbeat)	The heartbeat to request.
StreamEnvironmentBuilder	requestedMaxFrameSize(int requestedMaxFrameSize)	The maximum frame size to request.
EnvironmentBuilder	rpcTimeout(Duration timeout)	Timeout for RPC calls.
StreamEnvironmentBuilder	saslConfiguration(SaslConfiguration saslConfiguration)	The SASL configuration to use.
EnvironmentBuilder	scheduledExecutorService(ScheduledExecutorService scheduledExecutorService)	Set the ScheduledExecutorService used to: Schedule producers batch sending Handle connection recovery Handle topology update
EnvironmentBuilder.TlsConfiguration	tls()	Enable and configure TLS.
EnvironmentBuilder	topologyUpdateBackOffDelayPolicy(BackOffDelayPolicy topologyUpdateBackOffDelayPolicy)	The BackOffDelayPolicy to use for topology recovery.
StreamEnvironmentBuilder	uri(String uriString)	The URI of a node to connect to.
StreamEnvironmentBuilder	uris(List<String> uris)	A list of URIs of nodes of the same cluster to use to connect to.
StreamEnvironmentBuilder	username(String username)	The username to use.
StreamEnvironmentBuilder	virtualHost(String virtualHost)	The virtual host to connect to.

Methods inherited from class Object

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

Constructor Details

StreamEnvironmentBuilder

public StreamEnvironmentBuilder()

Method Details

uri

public StreamEnvironmentBuilder uri(String uriString)

Description copied from interface: EnvironmentBuilder

The URI of a node to connect to.

URI must be of the form rabbitmq-stream://guest:guest@localhost:5552/%2f.

Specified by:

uri in interface EnvironmentBuilder

Parameters:

uriString -

Returns:

this builder instance

uris

public StreamEnvironmentBuilder uris(List<String> uris)

Description copied from interface: EnvironmentBuilder

A list of URIs of nodes of the same cluster to use to connect to.

URIs must be of the form rabbitmq-stream://guest:guest@localhost:5552/%2f.

Specified by:

uris in interface EnvironmentBuilder

Parameters:

uris -

Returns:

this builder instance

host

public StreamEnvironmentBuilder host(String host)

Description copied from interface: EnvironmentBuilder

The host to connect to.

Specified by:

host in interface EnvironmentBuilder

Parameters:

host -

Returns:

this builder instance

port

public StreamEnvironmentBuilder port(int port)

Description copied from interface: EnvironmentBuilder

The port to use to connect.

Specified by:

port in interface EnvironmentBuilder

Parameters:

port -

Returns:

this builder instance

codec

public StreamEnvironmentBuilder codec(Codec codec)

Description copied from interface: EnvironmentBuilder

The AMQP 1.0 codec used to encode and decode AMQP 1.0 messages.

Specified by:

codec in interface EnvironmentBuilder

Parameters:

codec -

Returns:

this builder instance

compressionCodecFactory

public EnvironmentBuilder compressionCodecFactory(CompressionCodecFactory compressionCodecFactory)

Description copied from interface: EnvironmentBuilder

Compression codec factory to use for compression in sub-entry batching.

Specified by:

compressionCodecFactory in interface EnvironmentBuilder

Parameters:

compressionCodecFactory -

Returns:

this builder instance

See Also:

• ProducerBuilder.subEntrySize(int)
• ProducerBuilder.compression(Compression)

id

public EnvironmentBuilder id(String id)

Description copied from interface: EnvironmentBuilder

Informational ID for this environment instance.

This is currently used as a prefix for connection names. The broker does not enforce any kind of uniqueness based on this property. Default to rabbitmq-stream.

Specified by:

id in interface EnvironmentBuilder

Parameters:

id -

Returns:

this builder instance

rpcTimeout

public EnvironmentBuilder rpcTimeout(Duration timeout)

Description copied from interface: EnvironmentBuilder

Timeout for RPC calls.

Default is 10 seconds.

Specified by:

rpcTimeout in interface EnvironmentBuilder

Parameters:

timeout -

Returns:

this builder instance

saslConfiguration

public StreamEnvironmentBuilder saslConfiguration(SaslConfiguration saslConfiguration)

Description copied from interface: EnvironmentBuilder

The SASL configuration to use.

Specified by:

saslConfiguration in interface EnvironmentBuilder

Parameters:

saslConfiguration -

Returns:

this builder instance

See Also:

• EnvironmentBuilder.credentialsProvider(CredentialsProvider)

credentialsProvider

public StreamEnvironmentBuilder credentialsProvider(CredentialsProvider credentialsProvider)

Description copied from interface: EnvironmentBuilder

The CredentialsProvider to use.

Specified by:

credentialsProvider in interface EnvironmentBuilder

Parameters:

credentialsProvider -

Returns:

this builder instance

See Also:

• EnvironmentBuilder.saslConfiguration(SaslConfiguration)

username

public StreamEnvironmentBuilder username(String username)

Description copied from interface: EnvironmentBuilder

The username to use.

Specified by:

username in interface EnvironmentBuilder

Parameters:

username -

Returns:

this builder instance

password

public StreamEnvironmentBuilder password(String password)

Description copied from interface: EnvironmentBuilder

The password to use.

Specified by:

password in interface EnvironmentBuilder

Parameters:

password -

Returns:

this builder instance

virtualHost

public StreamEnvironmentBuilder virtualHost(String virtualHost)

Description copied from interface: EnvironmentBuilder

The virtual host to connect to.

Specified by:

virtualHost in interface EnvironmentBuilder

Parameters:

virtualHost -

Returns:

this builder instance

requestedHeartbeat

public StreamEnvironmentBuilder requestedHeartbeat(Duration requestedHeartbeat)

Description copied from interface: EnvironmentBuilder

The heartbeat to request.

Default is 60 seconds.

Specified by:

requestedHeartbeat in interface EnvironmentBuilder

Parameters:

requestedHeartbeat -

Returns:

this builder instance

See Also:

• Stream plugin documentation

requestedMaxFrameSize

public StreamEnvironmentBuilder requestedMaxFrameSize(int requestedMaxFrameSize)

Description copied from interface: EnvironmentBuilder

The maximum frame size to request.

Default is 1048576.

Specified by:

requestedMaxFrameSize in interface EnvironmentBuilder

Parameters:

requestedMaxFrameSize -

Returns:

this builder instance

See Also:

• Stream plugin documentation

chunkChecksum

public StreamEnvironmentBuilder chunkChecksum(ChunkChecksum chunkChecksum)

Description copied from interface: EnvironmentBuilder

The checksum strategy used for chunk checksum.

The default is CRC32 based on JDK implementation.

Specified by:

chunkChecksum in interface EnvironmentBuilder

Parameters:

chunkChecksum -

Returns:

this builder instance

clientProperties

public StreamEnvironmentBuilder clientProperties(Map<String,String> clientProperties)

Description copied from interface: EnvironmentBuilder

Custom client properties to add to default client properties.

Specified by:

clientProperties in interface EnvironmentBuilder

Parameters:

clientProperties -

Returns:

this builder instance

clientProperty

public StreamEnvironmentBuilder clientProperty(String key,
 String value)

Description copied from interface: EnvironmentBuilder

Add a custom client properties to default client properties.

Specified by:

clientProperty in interface EnvironmentBuilder

Parameters:

key -

value -

Returns:

this builder instance

metricsCollector

public StreamEnvironmentBuilder metricsCollector(MetricsCollector metricsCollector)

Description copied from interface: EnvironmentBuilder

Set up a MetricsCollector.

Specified by:

metricsCollector in interface EnvironmentBuilder

Parameters:

metricsCollector -

Returns:

this builder instance

scheduledExecutorService

public EnvironmentBuilder scheduledExecutorService(ScheduledExecutorService scheduledExecutorService)

Description copied from interface: EnvironmentBuilder

Set the ScheduledExecutorService used to:

• Schedule producers batch sending
• Handle connection recovery
• Handle topology update

Specified by:

scheduledExecutorService in interface EnvironmentBuilder

Parameters:

scheduledExecutorService - the service to use

Returns:

this builder instance

recoveryBackOffDelayPolicy

public EnvironmentBuilder recoveryBackOffDelayPolicy(BackOffDelayPolicy recoveryBackOffDelayPolicy)

Description copied from interface: EnvironmentBuilder

The BackOffDelayPolicy to use for connection recovery.

The default is a fixed delay of 5 seconds.

Specified by:

recoveryBackOffDelayPolicy in interface EnvironmentBuilder

Parameters:

recoveryBackOffDelayPolicy -

Returns:

this builder instance

topologyUpdateBackOffDelayPolicy

public EnvironmentBuilder topologyUpdateBackOffDelayPolicy(BackOffDelayPolicy topologyUpdateBackOffDelayPolicy)

Description copied from interface: EnvironmentBuilder

The BackOffDelayPolicy to use for topology recovery.

Topology recovery can kick in when streams leaders and replicas move from nodes to nodes in the cluster.

The default is a first delay of 5 seconds, then 1 second for the next attempts.

Specified by:

topologyUpdateBackOffDelayPolicy in interface EnvironmentBuilder

Parameters:

topologyUpdateBackOffDelayPolicy -

Returns:

this builder instance

addressResolver

public EnvironmentBuilder addressResolver(AddressResolver addressResolver)

Description copied from interface: EnvironmentBuilder

An AddressResolver to potentially change resolved node address to connect to.

Applications can use this abstraction to make sure connection attempts ignore metadata hints and always go to a single point like a load balancer. Consider setting EnvironmentBuilder.locatorConnectionCount(int) when using a load balancer.

The default implementation does not perform any logic, it just returns the passed-in address.

The default implementation is overridden automatically if the following conditions are met: the host to connect to is localhost, the user is guest, and no address resolver has been provided. The client will then always try to connect to localhost to facilitate local development. Just provide a pass-through address resolver to avoid this behavior, e.g.:

Environment.builder()
  .addressResolver(address -> address)
  .build();

Specified by:

addressResolver in interface EnvironmentBuilder

Parameters:

addressResolver - the address resolver

Returns:

this builder instance

See Also:

• "Connecting to Streams" blog post
• EnvironmentBuilder.locatorConnectionCount(int)

maxProducersByConnection

public EnvironmentBuilder maxProducersByConnection(int maxProducersByConnection)

Description copied from interface: EnvironmentBuilder

The maximum number of producers allocated to a single connection.

Default is 256, which is the maximum value.

The limit may not be strictly enforced in case of too many concurrent creations.

Specified by:

maxProducersByConnection in interface EnvironmentBuilder

Parameters:

maxProducersByConnection -

Returns:

this builder instance

maxTrackingConsumersByConnection

public EnvironmentBuilder maxTrackingConsumersByConnection(int maxTrackingConsumersByConnection)

Description copied from interface: EnvironmentBuilder

The maximum number of tracking consumers allocated to a single connection.

Default is 50, which is the maximum value.

The limit may not be strictly enforced in case of too many concurrent creations.

Specified by:

maxTrackingConsumersByConnection in interface EnvironmentBuilder

Parameters:

maxTrackingConsumersByConnection -

Returns:

this builder instance

maxConsumersByConnection

public EnvironmentBuilder maxConsumersByConnection(int maxConsumersByConnection)

Description copied from interface: EnvironmentBuilder

The maximum number of consumers allocated to a single connection.

Default is 256, which is the maximum value.

The limit may not be strictly enforced in case of too many concurrent creations.

Specified by:

maxConsumersByConnection in interface EnvironmentBuilder

Parameters:

maxConsumersByConnection -

Returns:

this builder instance

lazyInitialization

public EnvironmentBuilder lazyInitialization(boolean lazy)

Description copied from interface: EnvironmentBuilder

To delay the connection opening until necessary.

No connection will be open before it is necessary (for stream management or producer/consumer creation).

Default is false.

Specified by:

lazyInitialization in interface EnvironmentBuilder

Parameters:

lazy -

Returns:

this builder instance

forceReplicaForConsumers

public EnvironmentBuilder forceReplicaForConsumers(boolean forceReplica)

Description copied from interface: EnvironmentBuilder

Flag to force the connection to a stream replica for consumers.

The library will always prefer to connect to a stream replica to consume from, but it will fall back to the stream leader if no replica is available. This is the default behavior. Set this flag to true to make the library wait for a replica to become available if only the stream leader is available. This can lead to longer recovery time but help to offload a stream leader and let it deal only with write requests.

Note the library performs only 5 attempts to locate a replica before falling back to the leader when the flag is set to true.

The EnvironmentBuilder.recoveryBackOffDelayPolicy(BackOffDelayPolicy) and EnvironmentBuilder.topologyUpdateBackOffDelayPolicy(BackOffDelayPolicy) policies control the time between attempts.

Do not set this flag to true when streams have only 1 member (the leader), e.g. for local development.

Default is false.

Specified by:

forceReplicaForConsumers in interface EnvironmentBuilder

Parameters:

forceReplica - whether to force the connection to a replica or not

Returns:

this builder instance

See Also:

• EnvironmentBuilder.recoveryBackOffDelayPolicy(BackOffDelayPolicy)
• EnvironmentBuilder.topologyUpdateBackOffDelayPolicy(BackOffDelayPolicy)

forceLeaderForProducers

public EnvironmentBuilder forceLeaderForProducers(boolean forceLeader)

Description copied from interface: EnvironmentBuilder

Flag to force the connection to the stream leader for producers.

The library prefers to connect to a node that hosts a stream leader for producers (default behavior).

When using a load balancer, the library does not know in advance the node it connects to. It may have to retry to connect to the appropriate node.

It will retry until it connects to the appropriate node (flag set to true, the default). This provides the best data locality, but may require several attempts, delaying the creation or the recovery of producers. This usually suits high-throughput use cases.

The library will accept the connection to a stream replica if the flag is set to false . This will speed up the creation/recovery of producers, but at the cost of network hops between cluster nodes when publishing messages because only a stream leader accepts writes. This is usually acceptable for low-throughput use cases.

Changing the default value should only benefit systems where a load balancer sits between the client applications and the cluster nodes.

Default is true.

Specified by:

forceLeaderForProducers in interface EnvironmentBuilder

Parameters:

forceLeader - whether to force the connection to the leader or not

Returns:

this builder instance

See Also:

• EnvironmentBuilder.recoveryBackOffDelayPolicy(BackOffDelayPolicy)
• EnvironmentBuilder.topologyUpdateBackOffDelayPolicy(BackOffDelayPolicy)

tls

public EnvironmentBuilder.TlsConfiguration tls()

Description copied from interface: EnvironmentBuilder

Enable and configure TLS.

Specified by:

tls in interface EnvironmentBuilder

Returns:

the TLS configuration helper

netty

public EnvironmentBuilder.NettyConfiguration netty()

Description copied from interface: EnvironmentBuilder

Helper to configure netty.

Specified by:

netty in interface EnvironmentBuilder

Returns:

Netty configuration helper

oauth2

public EnvironmentBuilder.OAuth2Configuration oauth2()

Description copied from interface: EnvironmentBuilder

OAuth 2 settings.

Specified by:

oauth2 in interface EnvironmentBuilder

Returns:

OAuth 2 settings

See Also:

• EnvironmentBuilder.OAuth2Configuration

observationCollector

public EnvironmentBuilder observationCollector(ObservationCollector<?> observationCollector)

Description copied from interface: EnvironmentBuilder

Set up an ObservationCollector.

Specified by:

observationCollector in interface EnvironmentBuilder

Parameters:

observationCollector -

Returns:

this builder instance

locatorConnectionCount

public StreamEnvironmentBuilder locatorConnectionCount(int locatorCount)

Description copied from interface: EnvironmentBuilder

Set the expected number of "locator" connections to maintain.

Locator connections are used to perform infrastructure-related operations (e.g. looking up the topology of a stream to find an appropriate node to connect to).

It is recommended to maintain 2 to 3 locator connections. The environment uses the smaller of the number of passed-in URIs and 3 by default (see EnvironmentBuilder.uris(List)).

The number of locator connections should be explicitly set when a load balancer is used, as the environment cannot know the number of cluster nodes in this case (the only URI set is the one of the load balancer).

Specified by:

locatorConnectionCount in interface EnvironmentBuilder

Parameters:

locatorCount - number of expected locator connections

Returns:

this builder instance

See Also:

• EnvironmentBuilder.uris(List)
• EnvironmentBuilder.addressResolver(AddressResolver)

build

public Environment build()

Description copied from interface: EnvironmentBuilder

Create the Environment instance.

Specified by:

build in interface EnvironmentBuilder

Returns:

the configured environment