#1929

Committed 26 Mar 2025 09:24PM UTC coverage: 95.637% (-0.03%) from 95.662%

Build # #1929

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #1295 from nats-io/ttl-211

Per Message TTL Support for 2.11

Run Details

58 of 64 new or added lines in 3 files covered. (90.63%)

1 existing line in 1 file now uncovered.

11617 of 12147 relevant lines covered (95.64%)

0.96 hits per line

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

90.63

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

// Copyright 2015-2018 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 String 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;
    }

    /**
     * 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;
    }

    /**
     * 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;
        String 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 : msgTtlSeconds + "s";
            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) {
            if (messageTtlCustom == null) {
                this.messageTtl = null;
            }
            else {
                this.messageTtl = messageTtlCustom.trim();
                if (this.messageTtl.isEmpty()) {
                    this.messageTtl = null;
                }
            }
            return this;
        }

        /**
         * Sets the TTL for this specific message to be published and never be expired
         * @return The Builder
         */
        public Builder messageTtlNever() {
            this.messageTtl = "never";
            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-2018 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 String 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	/**
62	* Property used to configure a builder from a Properties object.
63	*/
64	public static final String PROP_STREAM_NAME = Options.PFX + "publish.stream";
65
66	/**
67	* Property used to configure a builder from a Properties object..
68	*/
69	public static final String PROP_PUBLISH_TIMEOUT = Options.PFX + "publish.timeout";
70
71	/**
72	* Gets the name of the stream.
73	* @return the name of the stream.
74	*/
75	public String getStream() {
76	return stream;	1✔
77	}
78
79	/**
80	* Gets the publish timeout.
81	* @return the publish timeout.
82	*/
83	public Duration getStreamTimeout() {
84	return streamTimeout;	1✔
85	}
86
87	/**
88	* Gets the expected stream.
89	* @return the stream.
90	*/
91	public String getExpectedStream() {
92	return expectedStream;	1✔
93	}
94
95	/**
96	* Gets the expected last message ID in the stream.
97	* @return the message ID.
98	*/
99	public String getExpectedLastMsgId() {
100	return expectedLastId;	1✔
101	}
102
103	/**
104	* Gets the expected last sequence number of the stream.
105	* @return sequence number
106	*/
107	public long getExpectedLastSequence() {
108	return expectedLastSeq;	1✔
109	}
110
111	/**
112	* Gets the expected last subject sequence number of the stream.
113	* @return sequence number
114	*/
115	public long getExpectedLastSubjectSequence() {
116	return expectedLastSubSeq;	1✔
117	}
118
119	/**
120	* Gets the message ID
121	* @return the message id;
122	*/
123	public String getMessageId() {
124	return this.msgId;	1✔
125	}
126
127	/**
128	* Gets the message ttl string. Might be null. Might be "never".
129	* 10 seconds would be "10s" for the server
130	* @return the message ttl string
131	*/
132	public String getMessageTtl() {
133	return messageTtl;	1✔
134	}
135
136	/**
137	* Creates a builder for the options.
138	* @return the builder
139	*/
140	public static Builder builder() {
141	return new Builder();	1✔
142	}
143
144	/**
145	* PublishOptions are created using a Builder. The builder supports chaining and will
146	* create a default set of options if no methods are calls. The builder can also
147	* be created from a properties object using the property names defined with the
148	* prefix PROP_ in this class.
149	*/
150	public static class Builder {
151	String stream = UNSET_STREAM;	1✔
152	Duration streamTimeout = DEFAULT_TIMEOUT;	1✔
153	String expectedStream;
154	String expectedLastId;
155	long expectedLastSeq = UNSET_LAST_SEQUENCE;	1✔
156	long expectedLastSubSeq = UNSET_LAST_SEQUENCE;	1✔
157	String msgId;
158	String messageTtl;
159
160	/**
161	* Constructs a new publish options Builder with the default values.
162	*/
163	public Builder() {}	1✔
164
165	/**
166	* Constructs a builder from properties
167	* @param properties properties
168	*/
169	public Builder(Properties properties) {	1✔
170	String s = properties.getProperty(PublishOptions.PROP_PUBLISH_TIMEOUT);	1✔
171	if (s != null) {	1✔
172	streamTimeout = Duration.parse(s);	1✔
173	}
174
175	s = properties.getProperty(PublishOptions.PROP_STREAM_NAME);	1✔
176	if (s != null) {	1✔
177	stream = s;	1✔
178	}
179	}	1✔
180
181	/**
182	* Sets the stream name for publishing. The default is undefined.
183	* @param stream The name of the stream.
184	* @return The Builder
185	*/
186	public Builder stream(String stream) {
187	this.stream = validateStreamName(stream, false);	1✔
188	return this;	1✔
189	}
190
191	/**
192	* Sets the timeout to wait for a publish acknowledgement from a JetStream
193	* enabled NATS server.
194	* @param timeout the publish timeout.
195	* @return The Builder
196	*/
197	public Builder streamTimeout(Duration timeout) {
198	this.streamTimeout = validateDurationNotRequiredGtOrEqZero(timeout, DEFAULT_TIMEOUT);	1✔
199	return this;	1✔
200	}
201
202	/**
203	* Sets the expected stream for the publish. If the
204	* stream does not match the server will not save the message.
205	* @param stream expected stream
206	* @return The Builder
207	*/
208	public Builder expectedStream(String stream) {
209	expectedStream = validateStreamName(stream, false);	1✔
210	return this;	1✔
211	}
212
213	/**
214	* Sets the expected last ID of the previously published message. If the
215	* message ID does not match the server will not save the message.
216	* @param lastMsgId the stream
217	* @return The Builder
218	*/
219	public Builder expectedLastMsgId(String lastMsgId) {
220	expectedLastId = emptyAsNull(lastMsgId);	1✔
221	return this;	1✔
222	}
223
224	/**
225	* Sets the expected message sequence of the publish
226	* @param sequence the expected last sequence number
227	* @return The Builder
228	*/
229	public Builder expectedLastSequence(long sequence) {
230	// 0 has NO meaning to expectedLastSequence but we except 0 b/c the sequence is really a ulong
231	expectedLastSeq = validateGtEqMinus1(sequence, "Last Sequence");	1✔
232	return this;	1✔
233	}
234
235	/**
236	* Sets the expected subject message sequence of the publish
237	* @param sequence the expected last subject sequence number
238	* @return The Builder
239	*/
240	public Builder expectedLastSubjectSequence(long sequence) {
241	expectedLastSubSeq = validateGtEqMinus1(sequence, "Last Subject Sequence");	1✔
242	return this;	1✔
243	}
244
245	/**
246	* Sets the message id. Message IDs are used for de-duplication
247	* and should be unique to each message payload.
248	* @param msgId the unique message id.
249	* @return The Builder
250	*/
251	public Builder messageId(String msgId) {
252	this.msgId = emptyAsNull(msgId);	1✔
253	return this;	1✔
254	}
255
256	/**
257	* Sets the TTL for this specific message to be published
258	* @param msgTtlSeconds the ttl in seconds
259	* @return The Builder
260	*/
261	public Builder messageTtlSeconds(int msgTtlSeconds) {
262	this.messageTtl = msgTtlSeconds < 1 ? null : msgTtlSeconds + "s";	1✔
263	return this;	1✔
264	}
265
266	/**
267	* Sets the TTL for this specific message to be published. Use at your own risk.
268	* 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>
269	* @param messageTtlCustom the ttl in seconds
270	* @return The Builder
271	*/
272	public Builder messageTtlCustom(String messageTtlCustom) {
NEW 273	if (messageTtlCustom == null) {	×
NEW 274	this.messageTtl = null;	×
275	}
276	else {
NEW 277	this.messageTtl = messageTtlCustom.trim();	×
NEW 278	if (this.messageTtl.isEmpty()) {	×
NEW 279	this.messageTtl = null;	×
280	}
281	}
NEW 282	return this;	×
283	}
284
285	/**
286	* Sets the TTL for this specific message to be published and never be expired
287	* @return The Builder
288	*/
289	public Builder messageTtlNever() {
290	this.messageTtl = "never";	1✔
291	return this;	1✔
292	}
293
294	/**
295	* Clears the expected so the build can be re-used.
296	* Clears the expectedLastId, expectedLastSequence and messageId fields.
297	* @return The Builder
298	*/
299	public Builder clearExpected() {
300	expectedLastId = null;	1✔
301	expectedLastSeq = UNSET_LAST_SEQUENCE;	1✔
302	expectedLastSubSeq = UNSET_LAST_SEQUENCE;	1✔
303	msgId = null;	1✔
304	return this;	1✔
305	}
306
307	/**
308	* Builds the publish options.
309	* @return publish options
310	*/
311	public PublishOptions build() {
312	return new PublishOptions(this);	1✔
313	}
314	}
315	}

nats-io / nats.java / #1929

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