Quarkus Reactive Messaging Nats Jetstream
This extension allow usage of NATS JetStream inside a Quarkus App, in JVM and Native mode.
The extension implements a new connector type quarkus-jetstream in SmallRye Reactive Messaging that will use the NATS client.
The client and configuration were significantly reworked after version 3.38.0. If you are using 3.38.0
or earlier, see the
3.38.0 documentation
instead of this page. Upgrading from 3.38.0? Follow the Migration Guide.
|
Installation
If you want to use this extension, you need to add the io.quarkiverse.reactivemessaging.nats-jetstream:quarkus-reactive-messaging-nats-jetstream extension first to your build file.
For instance, with Maven, add the following dependency to your POM file:
<dependency>
<groupId>io.quarkiverse.reactivemessaging.nats-jetstream</groupId>
<artifactId>quarkus-messaging-nats-jetstream</artifactId>
<version>4.0.0.Beta1</version>
</dependency>The connector is built on the quarkus-nats-jetstream-client extension, which is pulled in transitively. To use the reactive NATS JetStream Client without Reactive Messaging, depend on the client extension alone:
<dependency>
<groupId>io.quarkiverse.reactivemessaging.nats-jetstream</groupId>
<artifactId>quarkus-nats-jetstream-client</artifactId>
<version>4.0.0.Beta1</version>
</dependency>Then configure your application by adding the NATS JetStream connector type:
# Inbound mp.messaging.incoming.[channel-name].connector=quarkus-jetstream # Outbound mp.messaging.outgoing.[channel-name].connector=quarkus-jetstream
Receiving messages from NATS JetStream
Let’s imagine you have a NATS JetStream broker running, and accessible using the localhost:4242 address. Configure your application to receive NATS messages on the data channel from the stream named: test and the subject named: data as follows:
quarkus.messaging.nats.connection.servers=nats://localhost:4242 quarkus.messaging.nats.connection.username=guest quarkus.messaging.nats.connection.password=guest # The stream and its consumer are created if they don't already exist on NATS quarkus.messaging.nats.streams.test.name=test quarkus.messaging.nats.streams.test.subjects=data quarkus.messaging.nats.consumers.data-consumer.stream=test quarkus.messaging.nats.consumers.data-consumer.name=data-consumer quarkus.messaging.nats.consumers.data-consumer.filter-subject=data quarkus.messaging.nats.consumers.data-consumer.durable=true mp.messaging.incoming.data.connector=quarkus-jetstream mp.messaging.incoming.data.stream=test mp.messaging.incoming.data.consumer=data-consumer
Then, your application receives Message<Data>. You can consumes the payload directly:
package inbound;
import org.eclipse.microprofile.reactive.messaging.Incoming;
import jakarta.enterprise.context.ApplicationScoped;
@ApplicationScoped
public class DataConsumer {
@Incoming("data")
public void consume(Data data) {
// process your data.
}
}
If you want more examples, please take a look at the tests of the extension.
Request/Reply (RequestReply)
Instead of plain pub/sub, you can use request/reply semantics: inject RequestReply<Req, Rep> on an outgoing channel and call request(). Each call publishes the payload on the channel’s subject and returns a Uni that completes when the matching reply arrives, or fails with TimeoutException if no reply arrives within reply.timeout.
The stream MUST cover both the request subject and the reply subject. If the reply subject is not covered by the stream, replies can never be persisted or delivered and every request fails with a timeout (NATS happily accepts creating the consumer either way). If the stream does not exist at all, the first request fails with ConsumerManagementException; the next request re-subscribes automatically.
|
Requestor
Configure an outgoing channel for the requests:
mp.messaging.outgoing.orders.connector=quarkus-jetstream mp.messaging.outgoing.orders.stream=orders mp.messaging.outgoing.orders.subject=orders # Optional - defaults to '<subject>.replies' mp.messaging.outgoing.orders.reply.subject=orders.replies # Optional - default 5000 ms mp.messaging.outgoing.orders.reply.timeout=10000
Then inject the requestor and send requests:
package outbound;
import jakarta.enterprise.context.ApplicationScoped;
import jakarta.inject.Inject;
import org.eclipse.microprofile.reactive.messaging.Channel;
import io.quarkiverse.reactive.messaging.nats.jetstream.connector.reply.RequestReply;
import io.smallrye.mutiny.Uni;
@ApplicationScoped
public class OrderService {
@Inject
@Channel("orders")
RequestReply<String, String> requestor;
public Uni<String> placeOrder(String order) {
return requestor.request(order);
}
}
Replier
The replier is a plain @Incoming/@Outgoing method. The connector reads the reply subject advertised by the requestor (the message.reply-subject header) and routes the returned payload there automatically, echoing the correlation id (message.correlation-id) so concurrent requestors never receive each other’s replies:
package inbound;
import jakarta.enterprise.context.ApplicationScoped;
import org.eclipse.microprofile.reactive.messaging.Incoming;
import org.eclipse.microprofile.reactive.messaging.Outgoing;
@ApplicationScoped
public class OrderReplier {
@Incoming("orders-in")
@Outgoing("orders-out")
public String reply(String order) {
return "accepted:" + order;
}
}
Bind the channels to the same stream:
mp.messaging.incoming.orders-in.connector=quarkus-jetstream mp.messaging.incoming.orders-in.stream=orders mp.messaging.incoming.orders-in.consumer=orders-replier mp.messaging.outgoing.orders-out.connector=quarkus-jetstream mp.messaging.outgoing.orders-out.stream=orders mp.messaging.outgoing.orders-out.subject=orders
Requestor channel configuration
| Property Name | Description | Type | Default Value |
|---|---|---|---|
mp.messaging.outgoing.[channel-name].reply.subject |
The subject on which replies are expected |
String |
|
mp.messaging.outgoing.[channel-name].reply.timeout |
How long to wait for a reply in milliseconds |
Long |
5000 |
mp.messaging.outgoing.[channel-name].reply.inactive-threshold |
How long NATS keeps an idle request/reply consumer before reclaiming it, in milliseconds |
Long |
60000 |
mp.messaging.outgoing.[channel-name].reply.correlation-id.handler |
The |
String |
uuid |
mp.messaging.outgoing.[channel-name].reply.failure.handler |
The |
String |
The correlation id travels in the NATS message header message.correlation-id; the requestor advertises its reply subject via the message.reply-subject header. Both names are fixed for this version.
Scaling model and failure semantics
-
Every requestor instance creates its own non-durable consumer on the shared reply subject (
deliver_policy=new,max_deliver=1) and receives every reply; replies are demultiplexed by correlation id and non-matches are acknowledged and discarded. This trades a little extra network traffic for zero instance identity: requestors and repliers can scale independently. -
Idle consumers are reclaimed by NATS after
reply.inactive-threshold, so no consumer leaks accumulate. -
If publishing the request fails, the returned
Unifails withPublishException. -
If a reply payload should be treated as an error, provide a CDI bean implementing
ReplyFailureHandler(selected per channel viareply.failure.handler) and return the failure from it; the caller’sUnifails with that error. -
To observe in-flight requests, use
RequestReply.getPendingReplies(); pending entries are removed on completion, failure, timeout and shutdown.
Injecting the JetStream Client
Each configured datasource (see the "Datasources" section below) is exposed as an @ApplicationScoped CDI bean of
type io.quarkiverse.reactive.messaging.nats.jetstream.client.Client. The extension creates and registers these
beans itself - there is nothing to produce or configure - and closes them automatically on application shutdown.
You can inject a Client wherever you need direct access to the NATS JetStream API, for example to reach the
Key/Value or Object store (see below) or to publish/subscribe outside of a declarative channel.
For the default (unnamed) datasource, inject it directly with a plain, unqualified Client:
@Inject Client client;
For an additional, named datasource configured under quarkus.messaging.nats.data-sources.[datasource-name],
qualify the injection point with @Identifier("[datasource-name]"):
import io.smallrye.common.annotation.Identifier;
@Inject
@Identifier("orders-db")
Client client;
If the datasource to use is only known at runtime, inject @Any Instance<Client> clients and resolve it by name
yourself, for instance with SmallRye’s CDIUtils.getInstanceById(clients, datasourceName).
Using NATS Key/Value store
If you want to use the Key/Value store in NATS, then configure your application as follows:
quarkus.messaging.nats.connection.servers=nats://localhost:4242 quarkus.messaging.nats.connection.username=guest quarkus.messaging.nats.connection.password=guest # Creates a "test" bucket stored in memory. This part is optional. If you already have created a bucket on NATS # then this configuration is redundant. quarkus.messaging.nats.key-values.test.bucket-name=test quarkus.messaging.nats.key-values.test.storage-type=Memory quarkus.messaging.nats.key-values.test.description=Test bucket
Then your application can use the Key/Value store like this, through Client.keyValue(bucketName):
@Path("/key-value")
@RequestScoped
class KeyValueStoreResource {
private final Client client;
@Inject
public KeyValueStoreResource(Client client) {
this.client = client;
}
@GET
@Path("{key}")
public Uni<byte[]> getValue(@PathParam("key") String key) {
return client.keyValue("test").get(key).onItem().transform(entry -> entry.value().orElse(null));
}
@PUT
@Path("{key}")
@Consumes("application/octet-stream")
public Uni<Void> putValue(@PathParam("key") String key, byte[] value) {
return client.keyValue("test").put(key, value).replaceWithVoid();
}
@DELETE
@Path("{key}")
public Uni<Void> deleteValue(@PathParam("key") String key) {
return client.keyValue("test").delete(key);
}
}
Per-entry TTL
Buckets configured with limit-marker-ttl accept per-entry TTLs (requires NATS server 2.11+). create and purge
have overloads taking a Duration:
client.keyValue("cache").create(key, value, Duration.ofMinutes(5));
client.keyValue("cache").purge(key, Duration.ofMinutes(1));The server only supports whole-second TTLs, so the duration is rounded up to the next second.
Using NATS Object Store
If you want to use the Object store in NATS, then configure your application as follows:
quarkus.messaging.nats.connection.servers=nats://localhost:4242 quarkus.messaging.nats.connection.username=guest quarkus.messaging.nats.connection.password=guest # Creates a "test" object store. This part is optional. If you already have created the bucket on NATS # then this configuration is redundant. quarkus.messaging.nats.object-stores.test.bucket-name=test quarkus.messaging.nats.object-stores.test.description=Test object store
Then your application can use the Object store like this, through Client.objectStore(bucketName):
@Path("/object-store")
@RequestScoped
class ObjectStoreResource {
private final Client client;
@Inject
public ObjectStoreResource(Client client) {
this.client = client;
}
@GET
@Path("{name}")
public Uni<byte[]> getObject(@PathParam("name") String name) {
return client.objectStore("test").get(name).onItem().transform(ObjectEntry::data);
}
@PUT
@Path("{name}")
@Consumes("application/octet-stream")
public Uni<Void> putObject(@PathParam("name") String name, byte[] data) {
return client.objectStore("test").put(name, data).replaceWithVoid();
}
@DELETE
@Path("{name}")
public Uni<Void> deleteObject(@PathParam("name") String name) {
return client.objectStore("test").delete(name).replaceWithVoid();
}
}
Serializing Java Objects
By default, Java objects are serialized using java.io.* streams. If either the quarkus-jackson or
quarkus-jsonb dependency is present, objects are serialized as JSON instead.
To use a custom serializer, set the build-time property quarkus.messaging.nats.serializer to the
fully-qualified class name of a class that implements
io.quarkiverse.reactive.messaging.nats.jetstream.client.message.Serializer. The default value is
io.quarkiverse.reactive.messaging.nats.jetstream.client.message.JacksonSerializer.
quarkus.messaging.nats.serializer=com.example.MySerializerOpen Telemetry Tracing
To use Open Telemetry tracing, add the quarkus-opentelemetry dependency. Spans follow the
OpenTelemetry messaging semantic conventions
as implemented by the OpenTelemetry instrumentation API bundled with Quarkus, so they look like the spans of the
other Quarkus messaging extensions.
| Span | Kind | Created when |
|---|---|---|
|
|
A message is published. The span covers the publish until JetStream acknowledged it, and records the failure when publishing fails. |
|
|
A message is delivered to an incoming channel (push or pull consumer). The span is current while the application processes the message, so spans created meanwhile (for example publishing a reply) are its children. It ends when the message is acknowledged, or with the reason as error when it is not acknowledged. |
|
|
A message is pulled on demand with |
The trace context of the publisher is propagated in the message headers. It is the parent of the process and
receive spans, and is also added to them as a span link.
Besides the standard messaging.* attributes (messaging.system is nats, messaging.destination.name is the
subject, messaging.message.id the message id and messaging.message.conversation_id the request/reply correlation
id), spans carry NATS JetStream specific attributes:
-
messaging.nats.stream.name -
messaging.nats.consumer.name,messaging.nats.stream.sequence,messaging.nats.consumer.sequenceandmessaging.nats.message.delivered_counton received messages
The message payload is not recorded, since it may contain sensitive data and can be arbitrarily large. It can be
recorded as the messaging.nats.message.payload attribute with:
quarkus.messaging.nats.trace-payload=trueConfiguration
The stream, consumer, key-value store and object store configurations are all optional. Each declared entry is created on NATS if it does not already exist (addIfAbsent); if the configuration for a given name is omitted entirely, it is assumed to already exist on NATS.
| Property Name | Description | Type | Default Value |
|---|---|---|---|
quarkus.messaging.nats.connection.servers |
A comma-separated list of URI’s nats://{host}:{port} to use for establishing the initial connection to the NATS cluster. |
String |
nats://localhost:4222 |
quarkus.messaging.nats.connection.username |
The username to connect to the NATS server |
String |
|
quarkus.messaging.nats.connection.password |
The password to connect to the NATS server |
String |
|
quarkus.messaging.nats.connection.token |
The token to connect to the NATS server |
String |
|
quarkus.messaging.nats.connection.credential-path |
The path to the credentials file to connect to the NATS server |
String |
|
quarkus.messaging.nats.connection.connection-timeout |
The connection timeout |
Duration |
|
quarkus.messaging.nats.connection.error-listener |
The classname for the error listener |
String |
|
quarkus.messaging.nats.connection.buffer-size |
The size in bytes to make buffers for connections |
Integer |
|
quarkus.messaging.nats.connection.tls-algorithm |
The TLS algorithm |
String |
|
quarkus.messaging.nats.connection.connection-attempts |
The maximum number of attempts to attempt to re-connect to NATS |
Integer |
-1 (unlimited) |
quarkus.messaging.nats.connection.connection-backoff |
Back-off delay between to attempt to re-connect to NATS |
Duration |
1s |
quarkus.messaging.nats.connection.tls-configuration-name |
The name of the TLS configuration (bucket) used for client authentication in the TLS registry |
String |
|
quarkus.messaging.nats.connection.inbox-prefix |
Retrieves the optional inbox prefix used for communication. The inbox prefix allows customization of the subscriber’s inbox name. |
String |
Datasources
The connection, streams, consumers, key-value stores and object stores configured directly under
quarkus.messaging.nats. (as shown in the table above, and in the tables below) belong to the *default
datasource. For symmetry with named datasources, the same properties can also be set under
quarkus.messaging.nats.default.* (for example quarkus.messaging.nats.default.connection.servers) - both
spellings are aliases for exactly the same configuration, so use whichever reads better in your application.
Additional, named datasources - each with their own connection, streams, consumers, key-value stores and object
stores - are configured under quarkus.messaging.nats.data-sources.[datasource-name], using the same property
shape as the default datasource:
quarkus.messaging.nats.data-sources.orders-db.connection.servers=nats://orders-nats:4222 quarkus.messaging.nats.data-sources.orders-db.streams.orders.name=orders quarkus.messaging.nats.data-sources.orders-db.streams.orders.subjects=orders.*
Every configured datasource - default and named - gets its own Client CDI bean; see "Injecting the JetStream
Client`" above. A channel picks which datasource it uses via its `datasource attribute (see "Channel
configuration" below), and defaults to the default datasource when unset.
Each entry in the quarkus.messaging.nats.streams map is created (if absent) with [stream-name] as its map key; name must be set explicitly and is conventionally the same value as [stream-name].
| Property Name | Description | Type | Default Value |
|---|---|---|---|
quarkus.messaging.nats.streams.[stream-name].name |
The name of the stream |
String |
|
quarkus.messaging.nats.streams.[stream-name].retention-policy |
Declares the retention policy for the stream |
Limits, Interest or WorkQueue |
Interest |
quarkus.messaging.nats.streams.[stream-name].compression |
The compression used for stream data |
None or S2 |
None |
quarkus.messaging.nats.streams.[stream-name].storage-type |
The storage type for stream data |
File or Memory |
File |
quarkus.messaging.nats.streams.[stream-name].discard-policy |
The discard policy for this stream |
New or Old |
Old |
quarkus.messaging.nats.streams.[stream-name].description |
Description of the stream |
String |
|
quarkus.messaging.nats.streams.[stream-name].max-consumers |
The maximum number of consumers for this stream |
Long |
|
quarkus.messaging.nats.streams.[stream-name].max-messages |
The maximum number of messages for this stream |
Long |
|
quarkus.messaging.nats.streams.[stream-name].max-messages-per-subject |
The maximum number of messages per subject for this stream |
Long |
|
quarkus.messaging.nats.streams.[stream-name].max-bytes |
The maximum number of bytes for this stream |
Long |
|
quarkus.messaging.nats.streams.[stream-name].max-age |
The maximum message age for this stream |
Duration |
|
quarkus.messaging.nats.streams.[stream-name].maximum-message-size |
The maximum message size for this stream |
Integer |
|
quarkus.messaging.nats.streams.[stream-name].replicas |
The number of replicas a message must be stored on |
Integer |
1 |
quarkus.messaging.nats.streams.[stream-name].no-ack |
Whether acknowledgements are disabled for this stream |
Boolean |
true |
quarkus.messaging.nats.streams.[stream-name].template-owner |
The template json for this stream |
String |
|
quarkus.messaging.nats.streams.[stream-name].duplicate-window |
The duplicate checking window for the stream. |
Duration |
|
quarkus.messaging.nats.streams.[stream-name].subjects |
A comma separated list of the subject names bound to the stream |
String |
|
quarkus.messaging.nats.streams.[stream-name].placement |
Placement directives to consider when placing replicas of this stream, as a nested group ( |
Group |
|
quarkus.messaging.nats.streams.[stream-name].republish |
Republish configuration, as a nested group ( |
Group |
|
quarkus.messaging.nats.streams.[stream-name].subject-transform |
Subject transform applied to the stream, as a nested group ( |
Group |
|
quarkus.messaging.nats.streams.[stream-name].consumer-limits |
Default limits applied to consumers of this stream, as a nested group ( |
Group |
|
quarkus.messaging.nats.streams.[stream-name].mirror |
Configures this stream to mirror another stream, as a nested group ( |
Group |
|
quarkus.messaging.nats.streams.[stream-name].sources.[n] |
Sources this stream from one or more other streams; each entry has the same shape as |
List of Group |
|
quarkus.messaging.nats.streams.[stream-name].sealed |
Whether the stream is sealed (no further changes to messages allowed) |
Boolean |
false |
quarkus.messaging.nats.streams.[stream-name].allow-rollup |
Whether the stream allows rollup |
Boolean |
false |
quarkus.messaging.nats.streams.[stream-name].allow-direct |
Whether the stream allows direct message access |
Boolean |
false |
quarkus.messaging.nats.streams.[stream-name].mirror-direct |
Whether the stream allows higher performance and unified direct access for mirrors as well |
Boolean |
false |
quarkus.messaging.nats.streams.[stream-name].deny-delete |
Whether message delete is denied for the stream |
Boolean |
false |
quarkus.messaging.nats.streams.[stream-name].deny-purge |
Whether purge is denied for the stream |
Boolean |
false |
quarkus.messaging.nats.streams.[stream-name].discard-new-per-subject |
Whether the discard-new policy with max messages per subject is applied per subject |
Boolean |
false |
quarkus.messaging.nats.streams.[stream-name].first-sequence |
The first sequence used in the stream |
Long |
1 |
quarkus.messaging.nats.streams.[stream-name].subject-delete-marker-ttl |
The subject delete marker TTL |
Duration |
|
quarkus.messaging.nats.streams.[stream-name].allow-message-ttl |
Whether per-message TTL is allowed |
Boolean |
false |
quarkus.messaging.nats.streams.[stream-name].allow-message-schedules |
Whether message schedules are allowed |
Boolean |
false |
quarkus.messaging.nats.streams.[stream-name].allow-message-counter |
Whether the stream is a counter stream |
Boolean |
false |
quarkus.messaging.nats.streams.[stream-name].allow-atomic-publish |
Whether atomic (batch) publish is allowed |
Boolean |
false |
quarkus.messaging.nats.streams.[stream-name].allow-batched |
Whether batched publish is allowed |
Boolean |
false |
quarkus.messaging.nats.streams.[stream-name].persist-mode |
The persist mode used by the server when storing messages |
Default or Async |
Default |
quarkus.messaging.nats.streams.[stream-name].metadata.[key] |
Application-defined metadata for the stream |
Map<String, String> |
Consumer configuration
Unlike streams, consumers are configured as a single, flat map under quarkus.messaging.nats.consumers, each entry naming the stream it is bound to; a channel then references a pre-configured consumer by name via mp.messaging.incoming.[channel-name].consumer. Consumers are pull-based when only pull-options (or nothing) is set; push-options is used by consumers created outside of a regular incoming channel (for example the internal request/reply reply consumer).
| Property Name | Description | Type | Default Value |
|---|---|---|---|
quarkus.messaging.nats.consumers.[consumer-name].stream |
The name of the stream this consumer is bound to |
String |
|
quarkus.messaging.nats.consumers.[consumer-name].name |
The name of the consumer |
String |
|
quarkus.messaging.nats.consumers.[consumer-name].durable |
Set to true if the consumer should be durable |
Boolean |
false |
quarkus.messaging.nats.consumers.[consumer-name].description |
Description of the consumer |
String |
|
quarkus.messaging.nats.consumers.[consumer-name].filter-subject |
A subject that overlaps with the subjects bound to the stream to filter delivery to subscribers. Cannot be used together with |
String |
|
quarkus.messaging.nats.consumers.[consumer-name].filter-subjects |
A comma separated list of subjects that overlap with the subjects bound to the stream to filter delivery to subscribers. Cannot be used together with |
String |
|
quarkus.messaging.nats.consumers.[consumer-name].deliver-policy |
The point in the stream to receive messages from |
All, Last, New, ByStartSequence, ByStartTime or LastPerSubject |
All |
quarkus.messaging.nats.consumers.[consumer-name].start-sequence |
Used with the |
Long |
0 |
quarkus.messaging.nats.consumers.[consumer-name].start-time |
Used with the |
ZonedDateTime |
|
quarkus.messaging.nats.consumers.[consumer-name].acknowledge-wait |
The duration the server will wait for an ack for any individual message once it has been delivered to a consumer. If an ack is not received in time, the message will be redelivered |
Duration |
|
quarkus.messaging.nats.consumers.[consumer-name].inactive-threshold |
Duration that instructs the server to clean up the consumer once inactive for that long |
Duration |
|
quarkus.messaging.nats.consumers.[consumer-name].max-acknowledge-pending |
The maximum number of messages, without an acknowledgement, that can be outstanding |
Long |
|
quarkus.messaging.nats.consumers.[consumer-name].max-deliver |
The maximum number of times a specific message delivery will be attempted |
Long |
|
quarkus.messaging.nats.consumers.[consumer-name].backoff |
A comma separated list of durations controlling the re-delivery of messages on acknowledgement timeout |
List of Duration |
|
quarkus.messaging.nats.consumers.[consumer-name].replay-policy |
Whether messages are replayed at the original publish rate ( |
Instant or Original |
Instant |
quarkus.messaging.nats.consumers.[consumer-name].replicas |
The number of replicas for the consumer’s state. Defaults to inheriting the number of replicas from the stream |
Integer |
|
quarkus.messaging.nats.consumers.[consumer-name].memory-storage |
If set, forces the consumer state to be kept in memory rather than inherit the storage type of the stream |
Boolean |
false |
quarkus.messaging.nats.consumers.[consumer-name].sample-frequency |
The percentage of acknowledgements to sample for observability, e.g. |
String |
|
quarkus.messaging.nats.consumers.[consumer-name].headers-only |
Delivers only the headers of messages, not the bodies |
Boolean |
false |
quarkus.messaging.nats.consumers.[consumer-name].pause-until |
The time until the consumer is paused |
ZonedDateTime |
|
quarkus.messaging.nats.consumers.[consumer-name].metadata.[key] |
Application-defined metadata for the consumer |
Map<String, String> |
|
quarkus.messaging.nats.consumers.[consumer-name].pull-options.max-waiting |
The number of pulls that can be outstanding on this pull consumer |
Long |
|
quarkus.messaging.nats.consumers.[consumer-name].pull-options.max-expires |
The max expire time the server allows on pull requests |
Duration |
|
quarkus.messaging.nats.consumers.[consumer-name].pull-options.max-batch |
The maximum batch size a single pull request can make |
Long |
|
quarkus.messaging.nats.consumers.[consumer-name].pull-options.max-bytes |
The maximum total bytes that can be requested in a given batch |
Long |
|
quarkus.messaging.nats.consumers.[consumer-name].push-options.deliver-subject |
The subject messages are pushed to |
String |
|
quarkus.messaging.nats.consumers.[consumer-name].push-options.deliver-group |
The optional deliver (queue) group to join |
String |
|
quarkus.messaging.nats.consumers.[consumer-name].push-options.flow-control |
Enables per-subscription flow control using a sliding-window protocol |
Boolean |
false |
quarkus.messaging.nats.consumers.[consumer-name].push-options.idle-heartbeat |
If set, the server regularly sends a status message while there are no new messages to send, so the client knows the service is still up |
Duration |
|
quarkus.messaging.nats.consumers.[consumer-name].push-options.rate-limit |
Used to throttle the delivery of messages to the consumer, in bits per second |
Long |
At the channel level, mp.messaging.incoming.[channel-name] always binds to the consumer by name as a pull consumer (see the "Channel configuration" section below); push-options only applies to consumers created directly through the Client API, such as the request/reply reply consumer.
Channel configuration
| Property Name | Description | Type | Default Value |
|---|---|---|---|
mp.messaging.outgoing.[channel-name].stream |
The stream to publish messages to |
String |
|
mp.messaging.outgoing.[channel-name].subject |
The subject to publish messages to |
String |
|
mp.messaging.outgoing.[channel-name].retry-backoff |
The retry backoff in milliseconds for retrying failed publishes |
Long |
10000 |
mp.messaging.outgoing.[channel-name].datasource |
The name of the datasource (as configured under |
String |
default |
| Property Name | Description | Type | Default Value |
|---|---|---|---|
mp.messaging.incoming.[channel-name].stream |
The stream to consume messages from |
String |
|
mp.messaging.incoming.[channel-name].consumer |
The name of the pre-configured consumer to bind to (see the "Consumer configuration" section above) |
String |
|
mp.messaging.incoming.[channel-name].batch-size |
The number of messages pulled per fetch request |
Integer |
100 |
mp.messaging.incoming.[channel-name].timeout |
The timeout in milliseconds for pulling messages |
Long |
1000 |
mp.messaging.incoming.[channel-name].retry-backoff |
The retry backoff in milliseconds for retrying the subscription after a failure |
Long |
10000 |
mp.messaging.incoming.[channel-name].payload-type |
The class name of the payload type |
String |
|
mp.messaging.incoming.[channel-name].datasource |
The name of the datasource (as configured under |
String |
default |
Key/Value Store configuration
Each entry in the quarkus.messaging.nats.key-values map is created (if absent) with [bucket-name] as its map key; bucket-name must be set explicitly and is conventionally the same value as [bucket-name].
| Property Name | Description | Type | Default Value |
|---|---|---|---|
quarkus.messaging.nats.key-values.[bucket-name].bucket-name |
The name of the bucket |
String |
|
quarkus.messaging.nats.key-values.[bucket-name].description |
Description of the Key/Value store |
String |
|
quarkus.messaging.nats.key-values.[bucket-name].storage-type |
The storage type |
File or Memory |
File |
quarkus.messaging.nats.key-values.[bucket-name].max-bucket-size |
The maximum number of bytes for this bucket |
Long |
|
quarkus.messaging.nats.key-values.[bucket-name].maximum-value-size |
The maximum size for an individual value in the bucket |
Integer |
|
quarkus.messaging.nats.key-values.[bucket-name].max-history-per-key |
The maximum amount of history for any one key, from 1 to 64 inclusive. Includes the current value |
Integer |
64 |
quarkus.messaging.nats.key-values.[bucket-name].ttl |
The maximum age for a value in this bucket |
Duration |
|
quarkus.messaging.nats.key-values.[bucket-name].limit-marker-ttl |
The limit marker TTL |
Duration |
|
quarkus.messaging.nats.key-values.[bucket-name].replicas |
The number of replicas for this bucket |
Integer |
1 |
quarkus.messaging.nats.key-values.[bucket-name].compression |
Whether to use compression |
Boolean |
false |
quarkus.messaging.nats.key-values.[bucket-name].republish |
Republish configuration, as a nested group ( |
Group |
|
quarkus.messaging.nats.key-values.[bucket-name].placement |
Placement directives to consider when placing replicas of this bucket, as a nested group ( |
Group |
|
quarkus.messaging.nats.key-values.[bucket-name].mirror |
Configures this bucket to mirror another bucket/stream; same shape as the stream’s |
Group |
|
quarkus.messaging.nats.key-values.[bucket-name].sources.[n] |
Sources this bucket from one or more other buckets/streams; same shape as the stream’s |
List of Group |
|
quarkus.messaging.nats.key-values.[bucket-name].metadata.[key] |
Application-defined metadata for the bucket |
Map<String, String> |
Object Store configuration
Each entry in the quarkus.messaging.nats.object-stores map is created (if absent) with [bucket-name] as its map key; bucket-name must be set explicitly and is conventionally the same value as [bucket-name].
| Property Name | Description | Type | Default Value |
|---|---|---|---|
quarkus.messaging.nats.object-stores.[bucket-name].bucket-name |
The name of the bucket |
String |
|
quarkus.messaging.nats.object-stores.[bucket-name].description |
Description of the object store |
String |
|
quarkus.messaging.nats.object-stores.[bucket-name].storage-type |
The storage type |
File or Memory |
File |
quarkus.messaging.nats.object-stores.[bucket-name].max-bucket-size |
The maximum number of bytes for this bucket |
Long |
|
quarkus.messaging.nats.object-stores.[bucket-name].ttl |
The maximum age for a value in this bucket |
Duration |
|
quarkus.messaging.nats.object-stores.[bucket-name].replicas |
The number of replicas for this bucket |
Integer |
1 |
quarkus.messaging.nats.object-stores.[bucket-name].compression |
Whether to use compression |
Boolean |
false |
quarkus.messaging.nats.object-stores.[bucket-name].placement |
Placement directives to consider when placing replicas of this bucket, as a nested group ( |
Group |
|
quarkus.messaging.nats.object-stores.[bucket-name].metadata.[key] |
Application-defined metadata for the bucket |
Map<String, String> |
Dev service configuration
| Property Name | Description | Type | Default Value |
|---|---|---|---|
quarkus.messaging.nats.devservices.port |
Fixed port the dev service will listen to |
Integer |
|
quarkus.messaging.nats.devservices.image-name |
The image to use |
String |
nats:2.11 |
quarkus.messaging.nats.devservices.shared |
Indicates if the NATS JetStream broker managed by Quarkus Dev Services is shared |
Boolean |
true |
quarkus.messaging.nats.devservices.service-name |
This property is used when you need multiple shared NATS JetStream brokers. |
String |
nats |
quarkus.messaging.nats.devservices.enabled |
If Dev Services for NATS JetStream has been explicitly enabled or disabled |
Boolean |
true |
quarkus.messaging.nats.devservices.tls-configuration.certificate-file |
The absolute path to the PEM certificate file |
String |
|
quarkus.messaging.nats.devservices.tls-configuration.key-file |
The absolute path to the PEM key file |
String |
Configuration reference
The complete reference of the client extension’s quarkus.messaging.nats.* configuration properties, generated from
the extension sources.
Configuration property fixed at build time - All other configuration properties are overridable at runtime
Configuration property |
Type |
Default |
|---|---|---|
The fully-qualified class name of the serializer implementation used for message serialization and deserialization. The class must implement Environment variable: |
string |
|
If Dev Services for NATS JetStream has been explicitly enabled or disabled. Dev Services are generally enabled by default, unless there is an existing consumerConfiguration present. Environment variable: |
boolean |
|
Optional fixed port the dev service will listen to. If not defined, the port will be chosen randomly. Environment variable: |
int |
|
The image to use. Note that only NATS.io images are supported. Check NATS to find the available versions. Environment variable: |
string |
|
Indicates if the NATS JetStream broker managed by Quarkus Dev Services is shared. When shared, Quarkus looks for running containers using label-based service discovery. If a matching container is found, it is used, and so a second one is not started. Otherwise, Dev Services for NATS JetStream starts a new container. The discovery uses the Container sharing is only used in dev mode. Environment variable: |
boolean |
|
The value of the This property is used when you need multiple shared NATS JetStream brokers. Environment variable: |
string |
|
Retrieves the username to be used for authenticating with the NATS JetStream broker. If not explicitly configured, the default value is "guest". Environment variable: |
string |
|
Returns the password to be used for authenticating with the NATS JetStream broker. If not explicitly configured, the default value is "guest". Environment variable: |
string |
|
The absolute path to the PEM certificate file. Environment variable: |
string |
required |
The absolut path to the PEM key file. Environment variable: |
string |
required |
A list of URI’s nats://{host}:{port} to use for establishing the initial connection to the NATS cluster. Environment variable: |
list of string |
required |
The username to connect to the NATS server Environment variable: |
string |
|
The password to connect to the NATS server Environment variable: |
string |
|
The token to connect to the NATS server Environment variable: |
string |
|
The connection timeout Environment variable: |
||
The maximum number of maximumReconnects to attempt to re-connect to NATS. -1 means unlimited. Environment variable: |
int |
|
The size in bytes to make buffers for connections Environment variable: |
int |
|
The tls algorithm. Default is Environment variable: |
string |
|
The name of the TLS configuration (bucket) used for client authentication in the TLS registry. Environment variable: |
string |
|
The path to the credentials file for creating an AuthHandler Environment variable: |
string |
|
Retrieves the optional inbox prefix used for communication. The inbox prefix allows customization of the subscriber’s inbox name. Environment variable: |
string |
|
Whether the message payload is recorded as the Environment variable: |
boolean |
|
Retrieves the name as a non-null string. Environment variable: |
string |
required |
Gets the retention policy for this stream configuration. Environment variable: |
|
|
Gets the compression option for this stream configuration. Environment variable: |
|
|
Gets the storage type for this stream configuration. Environment variable: |
|
|
Gets the discard policy for this stream configuration. Environment variable: |
|
|
Gets the description of this stream configuration. Environment variable: |
string |
|
Gets the maximum number of consumers for this stream configuration. Environment variable: |
long |
|
Gets the maximum messages for this stream configuration. Environment variable: |
long |
|
Gets the maximum messages for this stream configuration. Environment variable: |
long |
|
Gets the maximum number of bytes for this stream configuration. Environment variable: |
long |
|
Gets the maximum message age for this stream configuration. Environment variable: |
||
Gets the maximum message size for this stream configuration. Environment variable: |
int |
|
Gets the number of replicas for this stream configuration. Environment variable: |
int |
|
Gets whether acknowledgements are required in this stream configuration. Environment variable: |
boolean |
|
Gets the template json for this stream configuration. Environment variable: |
string |
|
Gets the duplicate checking window stream configuration. Duration.ZERO means duplicate checking is not enabled. Environment variable: |
||
Gets the subjects for this stream configuration. Environment variable: |
list of string |
required |
The desired cluster name to place the stream. Environment variable: |
string |
|
Tags required on servers hosting this stream Environment variable: |
list of string |
required |
Get source, the Published subject matching filter Environment variable: |
string |
required |
Get destination, the RePublish Subject template Environment variable: |
string |
required |
Get headersOnly, Whether to RePublish only headers (no body) Environment variable: |
boolean |
|
Get source, the subject matching filter Environment variable: |
string |
required |
Get destination, the SubjectTransform Subject template Environment variable: |
string |
required |
Maximum value for max_ack_pending for consumers of this stream. Acts as a default when consumers do not set this value. Environment variable: |
long |
|
Maximum value for inactive_threshold for consumers of this stream. Acts as a default when consumers do not set this value. Environment variable: |
||
Get the name of the source. Same as getName() Environment variable: |
string |
required |
Get the name of the source. Same as getSourceName() Environment variable: |
string |
required |
Get the configured start sequence Environment variable: |
long |
|
Get the configured start time Environment variable: |
||
Get the configured filter subject Environment variable: |
string |
|
The subject prefix that imports the other account Environment variable: |
string |
|
The delivery subject to use for the push consumer. Environment variable: |
string |
|
Get source, the subject matching filter Environment variable: |
string |
required |
Get destination, the SubjectTransform Subject template Environment variable: |
string |
required |
The durable consumer name used for sourcing. Environment variable: |
string |
required |
The subject to deliver messages to. Environment variable: |
string |
required |
Get the name of the source. Same as getName() Environment variable: |
string |
required |
Get the name of the source. Same as getSourceName() Environment variable: |
string |
required |
Get the configured start sequence Environment variable: |
long |
|
Get the configured start time Environment variable: |
||
Get the configured filter subject Environment variable: |
string |
|
The subject prefix that imports the other account Environment variable: |
string |
|
The delivery subject to use for the push consumer. Environment variable: |
string |
|
Get source, the subject matching filter Environment variable: |
string |
required |
Get destination, the SubjectTransform Subject template Environment variable: |
string |
required |
The durable consumer name used for sourcing. Environment variable: |
string |
required |
The subject to deliver messages to. Environment variable: |
string |
required |
Get the flag indicating if the stream is sealed. Environment variable: |
boolean |
|
Get the flag indicating if the stream allows rollup. Environment variable: |
boolean |
|
Get the flag indicating if the stream allows direct message access. Environment variable: |
boolean |
|
Get the flag indicating if the stream allows higher performance and unified direct access for mirrors as well. Environment variable: |
boolean |
|
Get the flag indicating if deny delete is set for the stream Environment variable: |
boolean |
|
Get the flag indicating if deny purge is set for the stream Environment variable: |
boolean |
|
Whether discard policy with max message per subject is applied per subject. Environment variable: |
boolean |
|
Metadata for the stream Environment variable: |
Map<String,String> |
|
The first sequence used in the stream. Environment variable: |
long |
|
Get the Subject Delete Marker TTL duration. May be null. Environment variable: |
||
Whether Allow Message TTL is set Environment variable: |
boolean |
|
Whether Allow Message Schedules is set Environment variable: |
boolean |
|
Whether Allow Message Counter is set Environment variable: |
boolean |
|
Whether Allow Atomic Publish is set Environment variable: |
boolean |
|
Whether Allow Batched is set Environment variable: |
boolean |
|
Gets the persist mode or null if it was not explicitly set when creating or the server did not send it with stream info Environment variable: |
|
|
Retrieves the name of the consumer. Environment variable: |
string |
required |
If set, clients can have subscriptions bind to the consumer and resume until the consumer is explicitly deleted. Environment variable: |
boolean |
|
A subject that overlaps with the subjects bound to the stream to filter delivery to subscribers. Note: This cannot be used with the filterSubjects field. Environment variable: |
string |
|
A set of subjects that overlap with the subjects bound to the stream to filter delivery to subscribers. Note: This cannot be used with the filterSubject field. Environment variable: |
list of string |
|
The duration that the server will wait for an acknowledge for any individual message once it has been delivered to a consumer. If an ack is not received in time; the message will be re-delivered. Environment variable: |
||
The point in the stream to receive messages from; either DeliverAll; DeliverLast; DeliverNew; DeliverByStartSequence; DeliverByStartTime; or DeliverLastPerSubject Environment variable: |
|
|
Used with the DeliverByStartSequence deliver policy. Environment variable: |
long |
|
Used with the DeliverByStartTime deliver policy. Environment variable: |
||
A description of the consumer. This can be particularly useful for ephemeral consumers to indicate their purpose since a durable name cannot be provided. Environment variable: |
string |
|
Duration that instructs the server to clean up consumers inactive for that long. Prior to 2.9; this only applied to ephemeral consumers. Environment variable: |
||
Defines the maximum number of messages; without acknowledgment; that can be outstanding. Once this limit is reached; message delivery will be suspended. This limit applies across all of the consumer’s bound subscriptions. A value of -1 means there can be any number of pending acknowledgments (i.e.; no flow control). The default is 1000. Environment variable: |
long |
|
The maximum number of times a specific message delivery will be attempted. Applies to any message that is re-sent due to acknowledgment policy (i.e.; due to a negative acknowledgment or no acknowledgment sent by the client). The default is -1 (redeliver until acknowledged). Messages that have reached the maximum delivery count will stay in the stream. Environment variable: |
long |
|
A sequence of delays controlling the re-delivery of messages on acknowledgment timeout (but not on nak). The sequence length must be less than or equal to MaxDeliver. If backoff is not set; a timeout will result in immediate re-delivery. E.g.; MaxDeliver=5 backoff=[5s; 30s; 300s; 3600s; 84000s] will re-deliver a message 5 times over one day. When MaxDeliver is larger than the backoff list; the last delay in the list will apply for the remaining deliveries. Note that backoff is NOT applied to naked messages. A nak will result in immediate re-delivery unless nakWithDelay is used to set the re-delivery delay explicitly. When BackOff is set; it overrides AckWait entirely. The first value in the BackOff determines the AckWait value. Environment variable: |
list of Duration |
|
If the policy is ReplayOriginal; the messages in the stream will be pushed to the client at the same rate they were originally received; simulating the original timing. If the policy is ReplayInstant (default); the messages will be pushed to the client as fast as possible while adhering to the acknowledgment policy; Max Ack Pending; and the client’s ability to consume those messages. Environment variable: |
|
|
Sets the number of replicas for the consumer’s state. By default; when the value is set to Optional.empty(); consumers inherit the number of replicas from the stream. Environment variable: |
int |
|
If set; forces the consumer state to be kept in memory rather than inherit the storage type of the stream (default is file storage). This reduces I/O from acknowledgments; useful for ephemeral consumers. Environment variable: |
boolean |
|
Sets the percentage of acknowledgments that should be sampled for observability; 0-100. This value is a string and allows both 30 and 30% as valid values. Environment variable: |
string |
|
A set of application-defined key-value pairs for associating metadata with the consumer. Environment variable: |
Map<String,String> |
|
Delivers only the headers of messages in the stream; adding a Nats-Msg-Size header indicating the size of the removed payload. Environment variable: |
boolean |
|
The time until the consumer is paused Environment variable: |
||
The number of pulls that can be outstanding on a pull consumer Environment variable: |
long |
|
The max amount of expire time for the server to allow on pull requests. Environment variable: |
||
The maximum batch size a single pull request can make. When set with MaxRequestMaxBytes; the batch size will be constrained by whichever limit is hit first. Environment variable: |
long |
|
The maximum total bytes that can be requested in a given batch. When set with MaxRequestBatch; the batch size will be constrained by whichever limit is hit first. Environment variable: |
long |
|
The deliver subject Environment variable: |
string |
required |
Enables per-subscription flow control using a sliding-window protocol. This protocol relies on the server and client exchanging messages to regulate when and how many messages are pushed to the client. This one-to-one flow control mechanism works in tandem with the one-to-many flow control imposed by MaxAckPending across all subscriptions bound to a consumer Environment variable: |
boolean |
|
If the idle heartbeat period is set, the server will regularly send a status message to the client (i.e. when the period has elapsed) while there are no new messages to send. This lets the client know that the JetStream service is still up and running, even when there is no activity on the stream. The message status header will have a code of 100. Unlike FlowControl, it will have no reply to address. It may have a description such \"Idle Heartbeat\". Note that this heartbeat mechanism is all handled transparently by supported clients and does not need to be handled by the application Environment variable: |
||
Used to throttle the delivery of messages to the consumer, in bits per second. Environment variable: |
long |
|
The optional deliver group to join Environment variable: |
string |
|
Retrieves the name of the stream associated with this consumer. Environment variable: |
string |
required |
Gets the name of the bucket associated with this configuration. Environment variable: |
string |
required |
Gets the description of this bucket. Environment variable: |
string |
|
Gets the storage type for this bucket. Environment variable: |
|
|
Gets the maximum number of bytes for this bucket. Environment variable: |
long |
|
Gets the maximum size for an individual value in the bucket. Environment variable: |
int |
|
Get source, the Published subject matching filter Environment variable: |
string |
required |
Get destination, the RePublish Subject template Environment variable: |
string |
required |
Get headersOnly, Whether to RePublish only headers (no body) Environment variable: |
boolean |
|
Gets the maximum number of history for any one key. Includes the current value. Max History must be from 1 to 64 inclusive Environment variable: |
int |
|
Gets the maximum age for a value in this bucket. Environment variable: |
||
Gets the number of replicas for this stream configuration. Environment variable: |
int |
|
Indicates whether compression is enabled for this configuration. Environment variable: |
boolean |
|
The desired cluster name to place the stream. Environment variable: |
string |
|
Tags required on servers hosting this stream Environment variable: |
list of string |
required |
Retrieves the metadata associated with this configuration. The metadata is represented as a map where keys and values are both strings. Environment variable: |
Map<String,String> |
|
Get the name of the source. Same as getName() Environment variable: |
string |
required |
Get the name of the source. Same as getSourceName() Environment variable: |
string |
required |
Get the configured start sequence Environment variable: |
long |
|
Get the configured start time Environment variable: |
||
Get the configured filter subject Environment variable: |
string |
|
The subject prefix that imports the other account Environment variable: |
string |
|
The delivery subject to use for the push consumer. Environment variable: |
string |
|
Get source, the subject matching filter Environment variable: |
string |
required |
Get destination, the SubjectTransform Subject template Environment variable: |
string |
required |
The durable consumer name used for sourcing. Environment variable: |
string |
required |
The subject to deliver messages to. Environment variable: |
string |
required |
Get the name of the source. Same as getName() Environment variable: |
string |
required |
Get the name of the source. Same as getSourceName() Environment variable: |
string |
required |
Get the configured start sequence Environment variable: |
long |
|
Get the configured start time Environment variable: |
||
Get the configured filter subject Environment variable: |
string |
|
The subject prefix that imports the other account Environment variable: |
string |
|
The delivery subject to use for the push consumer. Environment variable: |
string |
|
Get source, the subject matching filter Environment variable: |
string |
required |
Get destination, the SubjectTransform Subject template Environment variable: |
string |
required |
The durable consumer name used for sourcing. Environment variable: |
string |
required |
The subject to deliver messages to. Environment variable: |
string |
required |
The limit marker ttl if set Environment variable: |
||
Gets the name of the bucket associated with this configuration. Environment variable: |
string |
required |
Provides an optional description for the object store configuration. Environment variable: |
string |
|
Retrieves the maximum allowable size, in bytes, for a single bucket in the object store configuration. This value determines the upper limit for the bucket’s capacity. Environment variable: |
long |
|
Retrieves the storage type for the object store configuration. This determines whether storage is file-based or memory-based. If not explicitly specified, the default storage type is Environment variable: |
|
|
Gets the maximum age for a value in this bucket. Environment variable: |
||
Gets the number of replicas for this stream configuration. Environment variable: |
int |
|
Indicates whether compression is enabled for this configuration. Environment variable: |
boolean |
|
The desired cluster name to place the stream. Environment variable: |
string |
|
Tags required on servers hosting this stream Environment variable: |
list of string |
required |
Retrieves the metadata associated with this configuration. The metadata is represented as a map where keys and values are both strings. Environment variable: |
Map<String,String> |
|
A list of URI’s nats://{host}:{port} to use for establishing the initial connection to the NATS cluster. Environment variable: |
list of string |
required |
The username to connect to the NATS server Environment variable: |
string |
|
The password to connect to the NATS server Environment variable: |
string |
|
The token to connect to the NATS server Environment variable: |
string |
|
The connection timeout Environment variable: |
||
The maximum number of maximumReconnects to attempt to re-connect to NATS. -1 means unlimited. Environment variable: |
int |
|
The size in bytes to make buffers for connections Environment variable: |
int |
|
The tls algorithm. Default is Environment variable: |
string |
|
The name of the TLS configuration (bucket) used for client authentication in the TLS registry. Environment variable: |
string |
|
The path to the credentials file for creating an AuthHandler Environment variable: |
string |
|
Retrieves the optional inbox prefix used for communication. The inbox prefix allows customization of the subscriber’s inbox name. Environment variable: |
string |
|
Retrieves the name as a non-null string. Environment variable: |
string |
required |
Gets the retention policy for this stream configuration. Environment variable: |
|
|
Gets the compression option for this stream configuration. Environment variable: |
|
|
Gets the storage type for this stream configuration. Environment variable: |
|
|
Gets the discard policy for this stream configuration. Environment variable: |
|
|
Gets the description of this stream configuration. Environment variable: |
string |
|
Gets the maximum number of consumers for this stream configuration. Environment variable: |
long |
|
Gets the maximum messages for this stream configuration. Environment variable: |
long |
|
Gets the maximum messages for this stream configuration. Environment variable: |
long |
|
Gets the maximum number of bytes for this stream configuration. Environment variable: |
long |
|
Gets the maximum message age for this stream configuration. Environment variable: |
||
Gets the maximum message size for this stream configuration. Environment variable: |
int |
|
Gets the number of replicas for this stream configuration. Environment variable: |
int |
|
Gets whether acknowledgements are required in this stream configuration. Environment variable: |
boolean |
|
Gets the template json for this stream configuration. Environment variable: |
string |
|
Gets the duplicate checking window stream configuration. Duration.ZERO means duplicate checking is not enabled. Environment variable: |
||
Gets the subjects for this stream configuration. Environment variable: |
list of string |
required |
The desired cluster name to place the stream. Environment variable: |
string |
|
Tags required on servers hosting this stream Environment variable: |
list of string |
required |
Get source, the Published subject matching filter Environment variable: |
string |
required |
Get destination, the RePublish Subject template Environment variable: |
string |
required |
Get headersOnly, Whether to RePublish only headers (no body) Environment variable: |
boolean |
|
Get source, the subject matching filter Environment variable: |
string |
required |
Get destination, the SubjectTransform Subject template Environment variable: |
string |
required |
Maximum value for max_ack_pending for consumers of this stream. Acts as a default when consumers do not set this value. Environment variable: |
long |
|
Maximum value for inactive_threshold for consumers of this stream. Acts as a default when consumers do not set this value. Environment variable: |
||
Get the name of the source. Same as getName() Environment variable: |
string |
required |
Get the name of the source. Same as getSourceName() Environment variable: |
string |
required |
Get the configured start sequence Environment variable: |
long |
|
Get the configured start time Environment variable: |
||
Get the configured filter subject Environment variable: |
string |
|
The subject prefix that imports the other account Environment variable: |
string |
|
The delivery subject to use for the push consumer. Environment variable: |
string |
|
Get source, the subject matching filter Environment variable: |
string |
required |
Get destination, the SubjectTransform Subject template Environment variable: |
string |
required |
The durable consumer name used for sourcing. Environment variable: |
string |
required |
The subject to deliver messages to. Environment variable: |
string |
required |
Get the name of the source. Same as getName() Environment variable: |
string |
required |
Get the name of the source. Same as getSourceName() Environment variable: |
string |
required |
Get the configured start sequence Environment variable: |
long |
|
Get the configured start time Environment variable: |
||
Get the configured filter subject Environment variable: |
string |
|
The subject prefix that imports the other account Environment variable: |
string |
|
The delivery subject to use for the push consumer. Environment variable: |
string |
|
Get source, the subject matching filter Environment variable: |
string |
required |
Get destination, the SubjectTransform Subject template Environment variable: |
string |
required |
The durable consumer name used for sourcing. Environment variable: |
string |
required |
The subject to deliver messages to. Environment variable: |
string |
required |
Get the flag indicating if the stream is sealed. Environment variable: |
boolean |
|
Get the flag indicating if the stream allows rollup. Environment variable: |
boolean |
|
Get the flag indicating if the stream allows direct message access. Environment variable: |
boolean |
|
Get the flag indicating if the stream allows higher performance and unified direct access for mirrors as well. Environment variable: |
boolean |
|
Get the flag indicating if deny delete is set for the stream Environment variable: |
boolean |
|
Get the flag indicating if deny purge is set for the stream Environment variable: |
boolean |
|
Whether discard policy with max message per subject is applied per subject. Environment variable: |
boolean |
|
Metadata for the stream Environment variable: |
Map<String,String> |
|
The first sequence used in the stream. Environment variable: |
long |
|
Get the Subject Delete Marker TTL duration. May be null. Environment variable: |
||
Whether Allow Message TTL is set Environment variable: |
boolean |
|
Whether Allow Message Schedules is set Environment variable: |
boolean |
|
Whether Allow Message Counter is set Environment variable: |
boolean |
|
Whether Allow Atomic Publish is set Environment variable: |
boolean |
|
Whether Allow Batched is set Environment variable: |
boolean |
|
Gets the persist mode or null if it was not explicitly set when creating or the server did not send it with stream info Environment variable: |
|
|
Retrieves the name of the consumer. Environment variable: |
string |
required |
If set, clients can have subscriptions bind to the consumer and resume until the consumer is explicitly deleted. Environment variable: |
boolean |
|
A subject that overlaps with the subjects bound to the stream to filter delivery to subscribers. Note: This cannot be used with the filterSubjects field. Environment variable: |
string |
|
A set of subjects that overlap with the subjects bound to the stream to filter delivery to subscribers. Note: This cannot be used with the filterSubject field. Environment variable: |
list of string |
|
The duration that the server will wait for an acknowledge for any individual message once it has been delivered to a consumer. If an ack is not received in time; the message will be re-delivered. Environment variable: |
||
The point in the stream to receive messages from; either DeliverAll; DeliverLast; DeliverNew; DeliverByStartSequence; DeliverByStartTime; or DeliverLastPerSubject Environment variable: |
|
|
Used with the DeliverByStartSequence deliver policy. Environment variable: |
long |
|
Used with the DeliverByStartTime deliver policy. Environment variable: |
||
A description of the consumer. This can be particularly useful for ephemeral consumers to indicate their purpose since a durable name cannot be provided. Environment variable: |
string |
|
Duration that instructs the server to clean up consumers inactive for that long. Prior to 2.9; this only applied to ephemeral consumers. Environment variable: |
||
Defines the maximum number of messages; without acknowledgment; that can be outstanding. Once this limit is reached; message delivery will be suspended. This limit applies across all of the consumer’s bound subscriptions. A value of -1 means there can be any number of pending acknowledgments (i.e.; no flow control). The default is 1000. Environment variable: |
long |
|
The maximum number of times a specific message delivery will be attempted. Applies to any message that is re-sent due to acknowledgment policy (i.e.; due to a negative acknowledgment or no acknowledgment sent by the client). The default is -1 (redeliver until acknowledged). Messages that have reached the maximum delivery count will stay in the stream. Environment variable: |
long |
|
A sequence of delays controlling the re-delivery of messages on acknowledgment timeout (but not on nak). The sequence length must be less than or equal to MaxDeliver. If backoff is not set; a timeout will result in immediate re-delivery. E.g.; MaxDeliver=5 backoff=[5s; 30s; 300s; 3600s; 84000s] will re-deliver a message 5 times over one day. When MaxDeliver is larger than the backoff list; the last delay in the list will apply for the remaining deliveries. Note that backoff is NOT applied to naked messages. A nak will result in immediate re-delivery unless nakWithDelay is used to set the re-delivery delay explicitly. When BackOff is set; it overrides AckWait entirely. The first value in the BackOff determines the AckWait value. Environment variable: |
list of Duration |
|
If the policy is ReplayOriginal; the messages in the stream will be pushed to the client at the same rate they were originally received; simulating the original timing. If the policy is ReplayInstant (default); the messages will be pushed to the client as fast as possible while adhering to the acknowledgment policy; Max Ack Pending; and the client’s ability to consume those messages. Environment variable: |
|
|
Sets the number of replicas for the consumer’s state. By default; when the value is set to Optional.empty(); consumers inherit the number of replicas from the stream. Environment variable: |
int |
|
If set; forces the consumer state to be kept in memory rather than inherit the storage type of the stream (default is file storage). This reduces I/O from acknowledgments; useful for ephemeral consumers. Environment variable: |
boolean |
|
Sets the percentage of acknowledgments that should be sampled for observability; 0-100. This value is a string and allows both 30 and 30% as valid values. Environment variable: |
string |
|
A set of application-defined key-value pairs for associating metadata with the consumer. Environment variable: |
Map<String,String> |
|
Delivers only the headers of messages in the stream; adding a Nats-Msg-Size header indicating the size of the removed payload. Environment variable: |
boolean |
|
The time until the consumer is paused Environment variable: |
||
The number of pulls that can be outstanding on a pull consumer Environment variable: |
long |
|
The max amount of expire time for the server to allow on pull requests. Environment variable: |
||
The maximum batch size a single pull request can make. When set with MaxRequestMaxBytes; the batch size will be constrained by whichever limit is hit first. Environment variable: |
long |
|
The maximum total bytes that can be requested in a given batch. When set with MaxRequestBatch; the batch size will be constrained by whichever limit is hit first. Environment variable: |
long |
|
The deliver subject Environment variable: |
string |
required |
Enables per-subscription flow control using a sliding-window protocol. This protocol relies on the server and client exchanging messages to regulate when and how many messages are pushed to the client. This one-to-one flow control mechanism works in tandem with the one-to-many flow control imposed by MaxAckPending across all subscriptions bound to a consumer Environment variable: |
boolean |
|
If the idle heartbeat period is set, the server will regularly send a status message to the client (i.e. when the period has elapsed) while there are no new messages to send. This lets the client know that the JetStream service is still up and running, even when there is no activity on the stream. The message status header will have a code of 100. Unlike FlowControl, it will have no reply to address. It may have a description such \"Idle Heartbeat\". Note that this heartbeat mechanism is all handled transparently by supported clients and does not need to be handled by the application Environment variable: |
||
Used to throttle the delivery of messages to the consumer, in bits per second. Environment variable: |
long |
|
The optional deliver group to join Environment variable: |
string |
|
Retrieves the name of the stream associated with this consumer. Environment variable: |
string |
required |
Gets the name of the bucket associated with this configuration. Environment variable: |
string |
required |
Gets the description of this bucket. Environment variable: |
string |
|
Gets the storage type for this bucket. Environment variable: |
|
|
Gets the maximum number of bytes for this bucket. Environment variable: |
long |
|
Gets the maximum size for an individual value in the bucket. Environment variable: |
int |
|
Get source, the Published subject matching filter Environment variable: |
string |
required |
Get destination, the RePublish Subject template Environment variable: |
string |
required |
Get headersOnly, Whether to RePublish only headers (no body) Environment variable: |
boolean |
|
Gets the maximum number of history for any one key. Includes the current value. Max History must be from 1 to 64 inclusive Environment variable: |
int |
|
Gets the maximum age for a value in this bucket. Environment variable: |
||
Gets the number of replicas for this stream configuration. Environment variable: |
int |
|
Indicates whether compression is enabled for this configuration. Environment variable: |
boolean |
|
The desired cluster name to place the stream. Environment variable: |
string |
|
Tags required on servers hosting this stream Environment variable: |
list of string |
required |
Retrieves the metadata associated with this configuration. The metadata is represented as a map where keys and values are both strings. Environment variable: |
Map<String,String> |
|
Get the name of the source. Same as getName() Environment variable: |
string |
required |
Get the name of the source. Same as getSourceName() Environment variable: |
string |
required |
Get the configured start sequence Environment variable: |
long |
|
Get the configured start time Environment variable: |
||
Get the configured filter subject Environment variable: |
string |
|
The subject prefix that imports the other account Environment variable: |
string |
|
The delivery subject to use for the push consumer. Environment variable: |
string |
|
Get source, the subject matching filter Environment variable: |
string |
required |
Get destination, the SubjectTransform Subject template Environment variable: |
string |
required |
The durable consumer name used for sourcing. Environment variable: |
string |
required |
The subject to deliver messages to. Environment variable: |
string |
required |
Get the name of the source. Same as getName() Environment variable: |
string |
required |
Get the name of the source. Same as getSourceName() Environment variable: |
string |
required |
Get the configured start sequence Environment variable: |
long |
|
Get the configured start time Environment variable: |
||
Get the configured filter subject Environment variable: |
string |
|
The subject prefix that imports the other account Environment variable: |
string |
|
The delivery subject to use for the push consumer. Environment variable: |
string |
|
Get source, the subject matching filter Environment variable: |
string |
required |
Get destination, the SubjectTransform Subject template Environment variable: |
string |
required |
The durable consumer name used for sourcing. Environment variable: |
string |
required |
The subject to deliver messages to. Environment variable: |
string |
required |
The limit marker ttl if set Environment variable: |
||
Gets the name of the bucket associated with this configuration. Environment variable: |
string |
required |
Provides an optional description for the object store configuration. Environment variable: |
string |
|
Retrieves the maximum allowable size, in bytes, for a single bucket in the object store configuration. This value determines the upper limit for the bucket’s capacity. Environment variable: |
long |
|
Retrieves the storage type for the object store configuration. This determines whether storage is file-based or memory-based. If not explicitly specified, the default storage type is Environment variable: |
|
|
Gets the maximum age for a value in this bucket. Environment variable: |
||
Gets the number of replicas for this stream configuration. Environment variable: |
int |
|
Indicates whether compression is enabled for this configuration. Environment variable: |
boolean |
|
The desired cluster name to place the stream. Environment variable: |
string |
|
Tags required on servers hosting this stream Environment variable: |
list of string |
required |
Retrieves the metadata associated with this configuration. The metadata is represented as a map where keys and values are both strings. Environment variable: |
Map<String,String> |
|
About the Duration format
To write duration values, use the standard You can also use a simplified format, starting with a number:
In other cases, the simplified format is translated to the
|
NATS JetStream
This extension uses the NATS JetStream client to connect to a NATS JetStream broker.
Reactive Messaging
This extension uses SmallRye Reactive Messaging to build data streaming applications.
If you want to go further, check the documentation of SmallRye Reactive Messaging, the implementation used in Quarkus.