#1925

Committed 20 Mar 2025 03:46PM UTC coverage: 95.662% (-0.3%) from 95.95%

Build # #1925

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #1239 from nats-io/main-2-11

Main for server v2.11

Run Details

179 of 219 new or added lines in 15 files covered. (81.74%)

2 existing lines in 2 files now uncovered.

11600 of 12126 relevant lines covered (95.66%)

0.96 hits per line

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

84.34

/src/main/java/io/nats/client/api/MessageInfo.java

// Copyright 2020 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.api;

import io.nats.client.Message;
import io.nats.client.impl.Headers;
import io.nats.client.support.*;

import java.time.ZonedDateTime;

import static io.nats.client.support.ApiConstants.*;
import static io.nats.client.support.JsonUtils.addRawJson;
import static io.nats.client.support.JsonValueUtils.*;
import static io.nats.client.support.NatsJetStreamConstants.*;

/**
 * The MessageInfo class contains information about a JetStream message.
 */
public class MessageInfo extends ApiResponse<MessageInfo> {

    private final String subject;
    private final long seq;
    private final byte[] data;
    private final ZonedDateTime time;
    private final Headers headers;
    private final String stream;
    private final long lastSeq;
    private final long numPending;
    private final Status status;

    /**
     * Create a Message Info
     * @deprecated This signature was public for unit testing but is no longer used.
     * @param msg the message
     */
    @Deprecated
    public MessageInfo(Message msg) {
        this(msg, null, null, false);
    }

    /**
     * Create a Message Info
     * @param msg the message
     * @param streamName the stream name if known
     * @param parseDirect true if the object is being created from a direct api call instead of get message
     */
    public MessageInfo(Message msg, String streamName, boolean parseDirect) {
        this(msg, null, streamName, parseDirect);
    }

    /**
     * Create a Message Info
     * @param status     the status
     * @param streamName the stream name if known
     */
    public MessageInfo(Status status, String streamName) {
        this(null, status, streamName, false);
    }

    private MessageInfo(Message msg, Status status, String streamName, boolean parseDirect) {
        super(parseDirect ? null : msg);

        // working vars because the object vars are final
        String _subject = null;
        long _seq = -1;
        byte[] _data = null;
        ZonedDateTime _time = null;
        Headers _headers = null;
        String _stream = null;
        long _lastSeq = -1;
        long _numPending = -1;
        Status _status = null;

        if (status != null) {
            _status = status;
            _stream = streamName;
        }
        else if (parseDirect) {
            Headers msgHeaders = msg.getHeaders();
            _subject = msgHeaders.getLast(NATS_SUBJECT);
            _data = msg.getData();
            _seq = Long.parseLong(msgHeaders.getLast(NATS_SEQUENCE));
            _time = DateTimeUtils.parseDateTime(msgHeaders.getLast(NATS_TIMESTAMP));
            _stream = msgHeaders.getLast(NATS_STREAM);
            String tempLastSeq = msgHeaders.getLast(NATS_LAST_SEQUENCE);
            if (tempLastSeq != null) {
                _lastSeq = JsonUtils.safeParseLong(tempLastSeq, -1);
            }
            String tempNumPending = msgHeaders.getLast(NATS_NUM_PENDING);
            if (tempNumPending != null) {
                _numPending = Long.parseLong(tempNumPending) - 1;
            }

            // these are control headers, not real headers so don't give them to the user. Must be done last
            _headers = new Headers(msgHeaders, true, MESSAGE_INFO_HEADERS);
        }
        else if (!hasError()){
            JsonValue mjv = readValue(jv, MESSAGE);
            _subject = readString(mjv, SUBJECT);
            _data = readBase64(mjv, DATA);
            _seq = readLong(mjv, SEQ, 0);
            _time = readDate(mjv, TIME);
            byte[] hdrBytes = readBase64(mjv, HDRS);
            _headers = hdrBytes == null ? null : new IncomingHeadersProcessor(hdrBytes).getHeaders();
            _stream = streamName;
        }

        this.subject = _subject;
        this.data = _data;
        this.seq = _seq;
        this.time = _time;
        this.headers = _headers;
        this.stream = _stream;
        this.lastSeq = _lastSeq;
        this.numPending = _numPending;
        this.status = _status;
    }

    /**
     * Get the message subject
     * @return the subject
     */
    public String getSubject() {
        return subject;
    }

    /**
     * Get the message sequence
     * @return the sequence number
     */
    public long getSeq() {
        return seq;
    }

    /**
     * Get the message data
     * @return the data bytes
     */
    public byte[] getData() {
        return data;
    }

    /**
     * Get the time the message was received
     * @return the time
     */
    public ZonedDateTime getTime() {
        return time;
    }

    /**
     * Get the headers
     * @return the headers object or null if there were no headers
     */
    public Headers getHeaders() {
        return headers;
    }

    /**
     * Get the name of the stream. Not always set.
     * @return the stream name or null if the name is not known.
     */
    public String getStream() {
        return stream;
    }

    /**
     * Get the sequence number of the last message in the stream. Not always set.
     * @return the last sequence or -1 if the value is not known.
     */
    public long getLastSeq() {
        return lastSeq;
    }

    /**
     * Amount of pending messages that can be requested with a subsequent batch request.
     * @return number of pending messages
     */
    public long getNumPending() {
        return numPending;
    }

    /**
     * Get the Status object. Null if this MessageInfo is not a Status.
     * @return the status object
     */
    public Status getStatus() {
        return status;
    }

    /**
     * Whether this MessageInfo is a regular message
     * @return true if the MessageInfo is a regular message
     */
    public boolean isMessage() {
        return status == null && !hasError();
    }

    /**
     * Whether this MessageInfo is a status message
     * @return true if this MessageInfo is a status message
     */
    public boolean isStatus() {
        return status != null;
    }

    /**
     * Whether this MessageInfo is a status message and is a direct EOB status
     * @return true if this MessageInfo is a status message and is a direct EOB status
     */
    public boolean isEobStatus() {
        return status != null && status.isEob();
    }

    /**
     * Whether this MessageInfo is a status message and is an error status
     * @return true if this MessageInfo is a status message and is an error status
     */
    public boolean isErrorStatus() {
        return status != null && !status.isEob();
    }

    @Override
    public String toString() {
        StringBuilder sb = JsonUtils.beginJsonPrefixed("\"MessageInfo\":");
        if (status != null) {
            JsonUtils.addField(sb, "status_code", status.getCode());
            JsonUtils.addField(sb, "status_message", status.getMessage());
        }
        else if (hasError()) {
            JsonUtils.addField(sb, ERROR, getError());
        }
        else {
            JsonUtils.addField(sb, SEQ, seq);
            JsonUtils.addField(sb, LAST_SEQ, lastSeq);
            JsonUtils.addFieldWhenGteMinusOne(sb, NUM_PENDING, numPending);
            JsonUtils.addField(sb, STREAM, stream);
            JsonUtils.addField(sb, SUBJECT, subject);
            JsonUtils.addField(sb, TIME, time);
            if (data == null) {
                addRawJson(sb, DATA, "null");
            }
            else {
                JsonUtils.addField(sb, "data_length", data.length);
            }
            JsonUtils.addField(sb, HDRS, headers);
        }
        return JsonUtils.endJson(sb).toString();
    }
}

1	// Copyright 2020 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.api;
15
16	import io.nats.client.Message;
17	import io.nats.client.impl.Headers;
18	import io.nats.client.support.*;
19
20	import java.time.ZonedDateTime;
21
22	import static io.nats.client.support.ApiConstants.*;
23	import static io.nats.client.support.JsonUtils.addRawJson;
24	import static io.nats.client.support.JsonValueUtils.*;
25	import static io.nats.client.support.NatsJetStreamConstants.*;
26
27	/**
28	* The MessageInfo class contains information about a JetStream message.
29	*/
30	public class MessageInfo extends ApiResponse<MessageInfo> {
31
32	private final String subject;
33	private final long seq;
34	private final byte[] data;
35	private final ZonedDateTime time;
36	private final Headers headers;
37	private final String stream;
38	private final long lastSeq;
39	private final long numPending;
40	private final Status status;
41
42	/**
43	* Create a Message Info
44	* @deprecated This signature was public for unit testing but is no longer used.
45	* @param msg the message
46	*/
47	@Deprecated
48	public MessageInfo(Message msg) {
49	this(msg, null, null, false);	1✔
50	}	1✔
51
52	/**
53	* Create a Message Info
54	* @param msg the message
55	* @param streamName the stream name if known
56	* @param parseDirect true if the object is being created from a direct api call instead of get message
57	*/
58	public MessageInfo(Message msg, String streamName, boolean parseDirect) {
59	this(msg, null, streamName, parseDirect);	1✔
60	}	1✔
61
62	/**
63	* Create a Message Info
64	* @param status the status
65	* @param streamName the stream name if known
66	*/
67	public MessageInfo(Status status, String streamName) {
NEW 68	this(null, status, streamName, false);	×
NEW 69	}	×
70
71	private MessageInfo(Message msg, Status status, String streamName, boolean parseDirect) {
72	super(parseDirect ? null : msg);	1✔
73
74	// working vars because the object vars are final
75	String _subject = null;	1✔
76	long _seq = -1;	1✔
77	byte[] _data = null;	1✔
78	ZonedDateTime _time = null;	1✔
79	Headers _headers = null;	1✔
80	String _stream = null;	1✔
81	long _lastSeq = -1;	1✔
82	long _numPending = -1;	1✔
83	Status _status = null;	1✔
84
85	if (status != null) {	1✔
NEW 86	_status = status;	×
NEW 87	_stream = streamName;	×
88	}
89	else if (parseDirect) {	1✔
90	Headers msgHeaders = msg.getHeaders();	1✔
91	_subject = msgHeaders.getLast(NATS_SUBJECT);	1✔
92	_data = msg.getData();	1✔
93	_seq = Long.parseLong(msgHeaders.getLast(NATS_SEQUENCE));	1✔
94	_time = DateTimeUtils.parseDateTime(msgHeaders.getLast(NATS_TIMESTAMP));	1✔
95	_stream = msgHeaders.getLast(NATS_STREAM);	1✔
96	String tempLastSeq = msgHeaders.getLast(NATS_LAST_SEQUENCE);	1✔
97	if (tempLastSeq != null) {	1✔
98	_lastSeq = JsonUtils.safeParseLong(tempLastSeq, -1);	1✔
99	}
100	String tempNumPending = msgHeaders.getLast(NATS_NUM_PENDING);	1✔
101	if (tempNumPending != null) {	1✔
NEW 102	_numPending = Long.parseLong(tempNumPending) - 1;	×
103	}
104
105	// these are control headers, not real headers so don't give them to the user. Must be done last
106	_headers = new Headers(msgHeaders, true, MESSAGE_INFO_HEADERS);	1✔
107	}	1✔
108	else if (!hasError()){	1✔
109	JsonValue mjv = readValue(jv, MESSAGE);	1✔
110	_subject = readString(mjv, SUBJECT);	1✔
111	_data = readBase64(mjv, DATA);	1✔
112	_seq = readLong(mjv, SEQ, 0);	1✔
113	_time = readDate(mjv, TIME);	1✔
114	byte[] hdrBytes = readBase64(mjv, HDRS);	1✔
115	_headers = hdrBytes == null ? null : new IncomingHeadersProcessor(hdrBytes).getHeaders();	1✔
116	_stream = streamName;	1✔
117	}
118
119	this.subject = _subject;	1✔
120	this.data = _data;	1✔
121	this.seq = _seq;	1✔
122	this.time = _time;	1✔
123	this.headers = _headers;	1✔
124	this.stream = _stream;	1✔
125	this.lastSeq = _lastSeq;	1✔
126	this.numPending = _numPending;	1✔
127	this.status = _status;	1✔
128	}	1✔
129
130	/**
131	* Get the message subject
132	* @return the subject
133	*/
134	public String getSubject() {
135	return subject;	1✔
136	}
137
138	/**
139	* Get the message sequence
140	* @return the sequence number
141	*/
142	public long getSeq() {
143	return seq;	1✔
144	}
145
146	/**
147	* Get the message data
148	* @return the data bytes
149	*/
150	public byte[] getData() {
151	return data;	1✔
152	}
153
154	/**
155	* Get the time the message was received
156	* @return the time
157	*/
158	public ZonedDateTime getTime() {
159	return time;	1✔
160	}
161
162	/**
163	* Get the headers
164	* @return the headers object or null if there were no headers
165	*/
166	public Headers getHeaders() {
167	return headers;	1✔
168	}
169
170	/**
171	* Get the name of the stream. Not always set.
172	* @return the stream name or null if the name is not known.
173	*/
174	public String getStream() {
175	return stream;	1✔
176	}
177
178	/**
179	* Get the sequence number of the last message in the stream. Not always set.
180	* @return the last sequence or -1 if the value is not known.
181	*/
182	public long getLastSeq() {
183	return lastSeq;	1✔
184	}
185
186	/**
187	* Amount of pending messages that can be requested with a subsequent batch request.
188	* @return number of pending messages
189	*/
190	public long getNumPending() {
NEW 191	return numPending;	×
192	}
193
194	/**
195	* Get the Status object. Null if this MessageInfo is not a Status.
196	* @return the status object
197	*/
198	public Status getStatus() {
NEW 199	return status;	×
200	}
201
202	/**
203	* Whether this MessageInfo is a regular message
204	* @return true if the MessageInfo is a regular message
205	*/
206	public boolean isMessage() {
NEW 207	return status == null && !hasError();	×
208	}
209
210	/**
211	* Whether this MessageInfo is a status message
212	* @return true if this MessageInfo is a status message
213	*/
214	public boolean isStatus() {
NEW 215	return status != null;	×
216	}
217
218	/**
219	* Whether this MessageInfo is a status message and is a direct EOB status
220	* @return true if this MessageInfo is a status message and is a direct EOB status
221	*/
222	public boolean isEobStatus() {
NEW 223	return status != null && status.isEob();	×
224	}
225
226	/**
227	* Whether this MessageInfo is a status message and is an error status
228	* @return true if this MessageInfo is a status message and is an error status
229	*/
230	public boolean isErrorStatus() {
NEW 231	return status != null && !status.isEob();	×
232	}
233
234	@Override
235	public String toString() {
236	StringBuilder sb = JsonUtils.beginJsonPrefixed("\"MessageInfo\":");	1✔
237	if (status != null) {	1✔
NEW 238	JsonUtils.addField(sb, "status_code", status.getCode());	×
NEW 239	JsonUtils.addField(sb, "status_message", status.getMessage());	×
240	}
241	else if (hasError()) {	1✔
242	JsonUtils.addField(sb, ERROR, getError());	1✔
243	}
244	else {
245	JsonUtils.addField(sb, SEQ, seq);	1✔
246	JsonUtils.addField(sb, LAST_SEQ, lastSeq);	1✔
247	JsonUtils.addFieldWhenGteMinusOne(sb, NUM_PENDING, numPending);	1✔
248	JsonUtils.addField(sb, STREAM, stream);	1✔
249	JsonUtils.addField(sb, SUBJECT, subject);	1✔
250	JsonUtils.addField(sb, TIME, time);	1✔
251	if (data == null) {	1✔
252	addRawJson(sb, DATA, "null");	1✔
253	}
254	else {
255	JsonUtils.addField(sb, "data_length", data.length);	1✔
256	}
257	JsonUtils.addField(sb, HDRS, headers);	1✔
258	}
259	return JsonUtils.endJson(sb).toString();	1✔
260	}
261	}

nats-io / nats.java / #1925

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