#1957

Committed 30 Apr 2025 08:01PM UTC coverage: 95.654% (+0.01%) from 95.641%

Build # #1957

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #1310 from nats-io/kv-limit-marker

KV Limit Marker

Run Details

55 of 59 new or added lines in 8 files covered. (93.22%)

4 existing lines in 3 files now uncovered.

11687 of 12218 relevant lines covered (95.65%)

0.96 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

96.88

/src/main/java/io/nats/client/PublishOptions.java

// Copyright 2015-2025 The NATS Authors
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at:
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package io.nats.client;

import java.time.Duration;
import java.util.Properties;

import static io.nats.client.support.Validator.*;

/**
 * The PublishOptions class specifies the options for publishing with JetStream enabled servers.
 * Options are created using a {@link PublishOptions.Builder Builder}.
 */
public class PublishOptions {
    /**
     * Use this variable for timeout in publish options.
     */
    public static final Duration DEFAULT_TIMEOUT = Options.DEFAULT_CONNECTION_TIMEOUT;

    /**
     * Use this variable to unset a stream in publish options.
     */
    public static final String UNSET_STREAM = null;

    /**
     * Use this variable to unset a sequence number in publish options.
     */
    public static final long UNSET_LAST_SEQUENCE = -1;

    private final String stream;
    private final Duration streamTimeout;
    private final String expectedStream;
    private final String expectedLastId;
    private final long expectedLastSeq;
    private final long expectedLastSubSeq;
    private final String msgId;
    private final MessageTtl messageTtl;

    private PublishOptions(Builder b) {
        this.stream = b.stream;
        this.streamTimeout = b.streamTimeout;
        this.expectedStream = b.expectedStream;
        this.expectedLastId = b.expectedLastId;
        this.expectedLastSeq = b.expectedLastSeq;
        this.expectedLastSubSeq = b.expectedLastSubSeq;
        this.msgId = b.msgId;
        this.messageTtl = b.messageTtl;
    }

    @Override
    public String toString() {
        return "PublishOptions{" +
            "stream='" + stream + '\'' +
            ", streamTimeout=" + streamTimeout +
            ", expectedStream='" + expectedStream + '\'' +
            ", expectedLastId='" + expectedLastId + '\'' +
            ", expectedLastSeq=" + expectedLastSeq +
            ", expectedLastSubSeq=" + expectedLastSubSeq +
            ", msgId='" + msgId + '\'' +
            ", messageTtl=" + getMessageTtl() +
            '}';
    }

    /**
     * Property used to configure a builder from a Properties object.
     */
    public static final String PROP_STREAM_NAME = Options.PFX + "publish.stream";

    /**
     * Property used to configure a builder from a Properties object..
     */
    public static final String PROP_PUBLISH_TIMEOUT = Options.PFX + "publish.timeout";

    /**
     * Gets the name of the stream.
     * @return the name of the stream.
     */
    public String getStream() {
        return stream;
    }

    /**
     * Gets the publish timeout.
     * @return the publish timeout.
     */
    public Duration getStreamTimeout() {
        return streamTimeout;
    }

    /**
     * Gets the expected stream.
     * @return the stream.
     */
    public String getExpectedStream() {
        return expectedStream;
    }

    /**
     * Gets the expected last message ID in the stream.
     * @return the message ID.
     */
    public String getExpectedLastMsgId() {
        return expectedLastId;
    }

    /**
     * Gets the expected last sequence number of the stream.
     * @return sequence number
     */
    public long getExpectedLastSequence() {
        return expectedLastSeq;
    }

    /**
     * Gets the expected last subject sequence number of the stream.
     * @return sequence number
     */
    public long getExpectedLastSubjectSequence() {
        return expectedLastSubSeq;
    }

    /**
     * Gets the message ID
     * @return the message id;
     */
    public String getMessageId() {
        return this.msgId;
    }

    /**
     * Gets the message ttl string. Might be null. Might be "never".
     * 10 seconds would be "10s" for the server
     * @return the message ttl string
     */
    public String getMessageTtl() {
        return messageTtl == null ? null : messageTtl.getMessageTtl();
    }

    /**
     * Creates a builder for the options.
     * @return the builder
     */
    public static Builder builder() {
        return new Builder();
    }

    /**
     * PublishOptions are created using a Builder. The builder supports chaining and will
     * create a default set of options if no methods are calls. The builder can also
     * be created from a properties object using the property names defined with the
     * prefix PROP_ in this class.
     */
    public static class Builder {
        String stream = UNSET_STREAM;
        Duration streamTimeout = DEFAULT_TIMEOUT;
        String expectedStream;
        String expectedLastId;
        long expectedLastSeq = UNSET_LAST_SEQUENCE;
        long expectedLastSubSeq = UNSET_LAST_SEQUENCE;
        String msgId;
        MessageTtl messageTtl;

        /**
         * Constructs a new publish options Builder with the default values.
         */
        public Builder() {}

        /**
         * Constructs a builder from properties
         * @param properties properties
         */
        public Builder(Properties properties) {
            String s = properties.getProperty(PublishOptions.PROP_PUBLISH_TIMEOUT);
            if (s != null) {
                streamTimeout = Duration.parse(s);
            }

            s = properties.getProperty(PublishOptions.PROP_STREAM_NAME);
            if (s != null) {
                stream = s;
            }
        }

        /**
         * Sets the stream name for publishing. The default is undefined.
         * @param stream The name of the stream.
         * @return The Builder
         */
        public Builder stream(String stream) {
            this.stream = validateStreamName(stream, false);
            return this;
        }

        /**
         * Sets the timeout to wait for a publish acknowledgement from a JetStream
         * enabled NATS server.
         * @param timeout the publish timeout.
         * @return The Builder
         */
        public Builder streamTimeout(Duration timeout) {
            this.streamTimeout = validateDurationNotRequiredGtOrEqZero(timeout, DEFAULT_TIMEOUT);
            return this;
        }

        /**
         * Sets the expected stream for the publish. If the
         * stream does not match the server will not save the message.
         * @param stream expected stream
         * @return The Builder
         */
        public Builder expectedStream(String stream) {
            expectedStream = validateStreamName(stream, false);
            return this;
        }

        /**
         * Sets the expected last ID of the previously published message.  If the
         * message ID does not match the server will not save the message.
         * @param lastMsgId the stream
         * @return The Builder
         */
        public Builder expectedLastMsgId(String lastMsgId) {
            expectedLastId = emptyAsNull(lastMsgId);
            return this;
        }

        /**
         * Sets the expected message sequence of the publish
         * @param sequence the expected last sequence number
         * @return The Builder
         */
        public Builder expectedLastSequence(long sequence) {
            // 0 has NO meaning to expectedLastSequence but we except 0 b/c the sequence is really a ulong
            expectedLastSeq = validateGtEqMinus1(sequence, "Last Sequence");
            return this;
        }

        /**
         * Sets the expected subject message sequence of the publish
         * @param sequence the expected last subject sequence number
         * @return The Builder
         */
        public Builder expectedLastSubjectSequence(long sequence) {
            expectedLastSubSeq = validateGtEqMinus1(sequence, "Last Subject Sequence");
            return this;
        }

        /**
         * Sets the message id. Message IDs are used for de-duplication
         * and should be unique to each message payload.
         * @param msgId the unique message id.
         * @return The Builder
         */
        public Builder messageId(String msgId) {
            this.msgId = emptyAsNull(msgId);
            return this;
        }

        /**
         * Sets the TTL for this specific message to be published
         * @param msgTtlSeconds the ttl in seconds
         * @return The Builder
         */
        public Builder messageTtlSeconds(int msgTtlSeconds) {
            this.messageTtl = msgTtlSeconds < 1 ? null : MessageTtl.seconds(msgTtlSeconds);
            return this;
        }

        /**
         * Sets the TTL for this specific message to be published. Use at your own risk.
         * The current specification can be found here @see <a href="https://github.com/nats-io/nats-architecture-and-design/blob/main/adr/ADR-43.md#per-message-ttl">JetStream Per-Message TTL</a>
         * @param messageTtlCustom the ttl in seconds
         * @return The Builder
         */
        public Builder messageTtlCustom(String messageTtlCustom) {
            this.messageTtl = nullOrEmpty(messageTtlCustom) ? null : MessageTtl.custom(messageTtlCustom);
            return this;
        }

        /**
         * Sets the TTL for this specific message to be published and never be expired
         * @return The Builder
         */
        public Builder messageTtlNever() {
            this.messageTtl = MessageTtl.never();
            return this;
        }

        /**
         * Sets the TTL for this specific message to be published
         * @return The Builder
         */
        public Builder messageTtl(MessageTtl messageTtl) {
            this.messageTtl = messageTtl;
            return this;
        }

        /**
         * Clears the expected so the build can be re-used.
         * Clears the expectedLastId, expectedLastSequence and messageId fields.
         * @return The Builder
         */
        public Builder clearExpected() {
            expectedLastId = null;
            expectedLastSeq = UNSET_LAST_SEQUENCE;
            expectedLastSubSeq = UNSET_LAST_SEQUENCE;
            msgId = null;
            return this;
        }

        /**
         * Builds the publish options.
         * @return publish options
         */
        public PublishOptions build() {
            return new PublishOptions(this);
        }
    }
}

1	// Copyright 2015-2025 The NATS Authors
2	// Licensed under the Apache License, Version 2.0 (the "License");
3	// you may not use this file except in compliance with the License.
4	// You may obtain a copy of the License at:
5	//
6	// http://www.apache.org/licenses/LICENSE-2.0
7	//
8	// Unless required by applicable law or agreed to in writing, software
9	// distributed under the License is distributed on an "AS IS" BASIS,
10	// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
11	// See the License for the specific language governing permissions and
12	// limitations under the License.
13
14	package io.nats.client;
15
16	import java.time.Duration;
17	import java.util.Properties;
18
19	import static io.nats.client.support.Validator.*;
20
21	/**
22	* The PublishOptions class specifies the options for publishing with JetStream enabled servers.
23	* Options are created using a {@link PublishOptions.Builder Builder}.
24	*/
25	public class PublishOptions {
26	/**
27	* Use this variable for timeout in publish options.
28	*/
29	public static final Duration DEFAULT_TIMEOUT = Options.DEFAULT_CONNECTION_TIMEOUT;	1✔
30
31	/**
32	* Use this variable to unset a stream in publish options.
33	*/
34	public static final String UNSET_STREAM = null;	1✔
35
36	/**
37	* Use this variable to unset a sequence number in publish options.
38	*/
39	public static final long UNSET_LAST_SEQUENCE = -1;
40
41	private final String stream;
42	private final Duration streamTimeout;
43	private final String expectedStream;
44	private final String expectedLastId;
45	private final long expectedLastSeq;
46	private final long expectedLastSubSeq;
47	private final String msgId;
48	private final MessageTtl messageTtl;
49
50	private PublishOptions(Builder b) {	1✔
51	this.stream = b.stream;	1✔
52	this.streamTimeout = b.streamTimeout;	1✔
53	this.expectedStream = b.expectedStream;	1✔
54	this.expectedLastId = b.expectedLastId;	1✔
55	this.expectedLastSeq = b.expectedLastSeq;	1✔
56	this.expectedLastSubSeq = b.expectedLastSubSeq;	1✔
57	this.msgId = b.msgId;	1✔
58	this.messageTtl = b.messageTtl;	1✔
59	}	1✔
60
61	@Override
62	public String toString() {
NEW 63	return "PublishOptions{" +	×
64	"stream='" + stream + '\'' +
65	", streamTimeout=" + streamTimeout +
66	", expectedStream='" + expectedStream + '\'' +
67	", expectedLastId='" + expectedLastId + '\'' +
68	", expectedLastSeq=" + expectedLastSeq +
69	", expectedLastSubSeq=" + expectedLastSubSeq +
70	", msgId='" + msgId + '\'' +
NEW 71	", messageTtl=" + getMessageTtl() +	×
72	'}';
73	}
74
75	/**
76	* Property used to configure a builder from a Properties object.
77	*/
78	public static final String PROP_STREAM_NAME = Options.PFX + "publish.stream";
79
80	/**
81	* Property used to configure a builder from a Properties object..
82	*/
83	public static final String PROP_PUBLISH_TIMEOUT = Options.PFX + "publish.timeout";
84
85	/**
86	* Gets the name of the stream.
87	* @return the name of the stream.
88	*/
89	public String getStream() {
90	return stream;	1✔
91	}
92
93	/**
94	* Gets the publish timeout.
95	* @return the publish timeout.
96	*/
97	public Duration getStreamTimeout() {
98	return streamTimeout;	1✔
99	}
100
101	/**
102	* Gets the expected stream.
103	* @return the stream.
104	*/
105	public String getExpectedStream() {
106	return expectedStream;	1✔
107	}
108
109	/**
110	* Gets the expected last message ID in the stream.
111	* @return the message ID.
112	*/
113	public String getExpectedLastMsgId() {
114	return expectedLastId;	1✔
115	}
116
117	/**
118	* Gets the expected last sequence number of the stream.
119	* @return sequence number
120	*/
121	public long getExpectedLastSequence() {
122	return expectedLastSeq;	1✔
123	}
124
125	/**
126	* Gets the expected last subject sequence number of the stream.
127	* @return sequence number
128	*/
129	public long getExpectedLastSubjectSequence() {
130	return expectedLastSubSeq;	1✔
131	}
132
133	/**
134	* Gets the message ID
135	* @return the message id;
136	*/
137	public String getMessageId() {
138	return this.msgId;	1✔
139	}
140
141	/**
142	* Gets the message ttl string. Might be null. Might be "never".
143	* 10 seconds would be "10s" for the server
144	* @return the message ttl string
145	*/
146	public String getMessageTtl() {
147	return messageTtl == null ? null : messageTtl.getMessageTtl();	1✔
148	}
149
150	/**
151	* Creates a builder for the options.
152	* @return the builder
153	*/
154	public static Builder builder() {
155	return new Builder();	1✔
156	}
157
158	/**
159	* PublishOptions are created using a Builder. The builder supports chaining and will
160	* create a default set of options if no methods are calls. The builder can also
161	* be created from a properties object using the property names defined with the
162	* prefix PROP_ in this class.
163	*/
164	public static class Builder {
165	String stream = UNSET_STREAM;	1✔
166	Duration streamTimeout = DEFAULT_TIMEOUT;	1✔
167	String expectedStream;
168	String expectedLastId;
169	long expectedLastSeq = UNSET_LAST_SEQUENCE;	1✔
170	long expectedLastSubSeq = UNSET_LAST_SEQUENCE;	1✔
171	String msgId;
172	MessageTtl messageTtl;
173
174	/**
175	* Constructs a new publish options Builder with the default values.
176	*/
177	public Builder() {}	1✔
178
179	/**
180	* Constructs a builder from properties
181	* @param properties properties
182	*/
183	public Builder(Properties properties) {	1✔
184	String s = properties.getProperty(PublishOptions.PROP_PUBLISH_TIMEOUT);	1✔
185	if (s != null) {	1✔
186	streamTimeout = Duration.parse(s);	1✔
187	}
188
189	s = properties.getProperty(PublishOptions.PROP_STREAM_NAME);	1✔
190	if (s != null) {	1✔
191	stream = s;	1✔
192	}
193	}	1✔
194
195	/**
196	* Sets the stream name for publishing. The default is undefined.
197	* @param stream The name of the stream.
198	* @return The Builder
199	*/
200	public Builder stream(String stream) {
201	this.stream = validateStreamName(stream, false);	1✔
202	return this;	1✔
203	}
204
205	/**
206	* Sets the timeout to wait for a publish acknowledgement from a JetStream
207	* enabled NATS server.
208	* @param timeout the publish timeout.
209	* @return The Builder
210	*/
211	public Builder streamTimeout(Duration timeout) {
212	this.streamTimeout = validateDurationNotRequiredGtOrEqZero(timeout, DEFAULT_TIMEOUT);	1✔
213	return this;	1✔
214	}
215
216	/**
217	* Sets the expected stream for the publish. If the
218	* stream does not match the server will not save the message.
219	* @param stream expected stream
220	* @return The Builder
221	*/
222	public Builder expectedStream(String stream) {
223	expectedStream = validateStreamName(stream, false);	1✔
224	return this;	1✔
225	}
226
227	/**
228	* Sets the expected last ID of the previously published message. If the
229	* message ID does not match the server will not save the message.
230	* @param lastMsgId the stream
231	* @return The Builder
232	*/
233	public Builder expectedLastMsgId(String lastMsgId) {
234	expectedLastId = emptyAsNull(lastMsgId);	1✔
235	return this;	1✔
236	}
237
238	/**
239	* Sets the expected message sequence of the publish
240	* @param sequence the expected last sequence number
241	* @return The Builder
242	*/
243	public Builder expectedLastSequence(long sequence) {
244	// 0 has NO meaning to expectedLastSequence but we except 0 b/c the sequence is really a ulong
245	expectedLastSeq = validateGtEqMinus1(sequence, "Last Sequence");	1✔
246	return this;	1✔
247	}
248
249	/**
250	* Sets the expected subject message sequence of the publish
251	* @param sequence the expected last subject sequence number
252	* @return The Builder
253	*/
254	public Builder expectedLastSubjectSequence(long sequence) {
255	expectedLastSubSeq = validateGtEqMinus1(sequence, "Last Subject Sequence");	1✔
256	return this;	1✔
257	}
258
259	/**
260	* Sets the message id. Message IDs are used for de-duplication
261	* and should be unique to each message payload.
262	* @param msgId the unique message id.
263	* @return The Builder
264	*/
265	public Builder messageId(String msgId) {
266	this.msgId = emptyAsNull(msgId);	1✔
267	return this;	1✔
268	}
269
270	/**
271	* Sets the TTL for this specific message to be published
272	* @param msgTtlSeconds the ttl in seconds
273	* @return The Builder
274	*/
275	public Builder messageTtlSeconds(int msgTtlSeconds) {
276	this.messageTtl = msgTtlSeconds < 1 ? null : MessageTtl.seconds(msgTtlSeconds);	1✔
277	return this;	1✔
278	}
279
280	/**
281	* Sets the TTL for this specific message to be published. Use at your own risk.
282	* The current specification can be found here @see <a href="https://github.com/nats-io/nats-architecture-and-design/blob/main/adr/ADR-43.md#per-message-ttl">JetStream Per-Message TTL</a>
283	* @param messageTtlCustom the ttl in seconds
284	* @return The Builder
285	*/
286	public Builder messageTtlCustom(String messageTtlCustom) {
287	this.messageTtl = nullOrEmpty(messageTtlCustom) ? null : MessageTtl.custom(messageTtlCustom);	1✔
288	return this;	1✔
289	}
290
291	/**
292	* Sets the TTL for this specific message to be published and never be expired
293	* @return The Builder
294	*/
295	public Builder messageTtlNever() {
296	this.messageTtl = MessageTtl.never();	1✔
297	return this;	1✔
298	}
299
300	/**
301	* Sets the TTL for this specific message to be published
302	* @return The Builder
303	*/
304	public Builder messageTtl(MessageTtl messageTtl) {
305	this.messageTtl = messageTtl;	1✔
306	return this;	1✔
307	}
308
309	/**
310	* Clears the expected so the build can be re-used.
311	* Clears the expectedLastId, expectedLastSequence and messageId fields.
312	* @return The Builder
313	*/
314	public Builder clearExpected() {
315	expectedLastId = null;	1✔
316	expectedLastSeq = UNSET_LAST_SEQUENCE;	1✔
317	expectedLastSubSeq = UNSET_LAST_SEQUENCE;	1✔
318	msgId = null;	1✔
319	return this;	1✔
320	}
321
322	/**
323	* Builds the publish options.
324	* @return publish options
325	*/
326	public PublishOptions build() {
327	return new PublishOptions(this);	1✔
328	}
329	}
330	}

nats-io / nats.java / #1957

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous