#2231

Committed 26 Sep 2025 04:51PM UTC coverage: 95.521% (-0.001%) from 95.522%

Build # #2231

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #1437 from nats-io/fix_size_in_bytes

Properly return size in bytes

Run Details

3 of 7 new or added lines in 2 files covered. (42.86%)

6 existing lines in 2 files now uncovered.

12155 of 12725 relevant lines covered (95.52%)

0.96 hits per line

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

80.77

/src/main/java/io/nats/client/ErrorListener.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 io.nats.client.api.ServerInfo;
import io.nats.client.support.Status;

/**
 * This library groups problems into four categories:
 * <dl>
 * <dt>Errors</dt>
 * <dd>The server sent an error message using the {@code -err} protocol operation.</dd>
 * <dt>Exceptions</dt>
 * <dd>A Java exception occurred, and was handled by the library.</dd>
 * <dt>Slow Consumers</dt>
 * <dd>One of the connections consumers, Subscription or Dispatcher, is slow, and starting to drop messages.</dd>
 * <dt>Fast Producers</dt>
 * <dd>One of the connections producers is too fast, and is discarding messages</dd>
 * </dl>
 * <p>All of these problems are reported to the application code using the ErrorListener. The
 * listener is configured in the {@link Options Options} at creation time.
 */
public interface ErrorListener {
    /**
     * NATs related errors that occur asynchronously in the client library are sent
     * to an ErrorListener via errorOccurred. The ErrorListener can use the error text to decide what to do about the problem.
     * <p>The text for an error is described in the protocol doc at `https://nats.io/documentation/internals/nats-protocol`.
     * <p>In some cases the server will close the clients connection after sending one of these errors. In that case, the
     * connections {@link ConnectionListener ConnectionListener} will be notified.
     * @param conn The connection associated with the error
     * @param error The text of error that has occurred, directly from the server
     */
    default void errorOccurred(Connection conn, String error) {};

    /**
     * Exceptions that occur in the "normal" course of operations are sent to the
     * ErrorListener using exceptionOccurred. Examples include, application exceptions
     * during Dispatcher callbacks, IOExceptions from the underlying socket, etc..
     * The library will try to handle these, via reconnect or catching them, but they are
     * forwarded here in case the application code needs them for debugging purposes.
     *
     * @param conn The connection associated with the error
     * @param exp The exception that has occurred, and was handled by the library
     */
    default void exceptionOccurred(Connection conn, Exception exp) {};

    /**
     * Called by the connection when a &quot;slow&quot; consumer is detected. This call is only made once
     * until the consumer stops being slow. At which point it will be called again if the consumer starts
     * being slow again.
     *
     * <p>See {@link Consumer#setPendingLimits(long, long) Consumer.setPendingLimits}
     * for information on how to configure when this method is fired.
     *
     * <p> Slow consumers will result in dropped messages each consumer provides a method
     * for retrieving the count of dropped messages, see {@link Consumer#getDroppedCount() Consumer.getDroppedCount}.
     *
     * @param conn The connection associated with the error
     * @param consumer The consumer that is being marked slow
     */
    default void slowConsumerDetected(Connection conn, Consumer consumer) {};

    /**
     * Called by the connection when a message is discarded.
     *
     * @param conn The connection that discarded the message
     * @param msg The message that is discarded
     */
    default void messageDiscarded(Connection conn, Message msg) {}

    /**
     * Called when subscription heartbeats are missed according to the configured period and threshold.
     * The consumer must be configured with an idle heartbeat time.
     *
     * @param conn The connection that had the issue
     * @param sub the JetStreamSubscription that this occurred on
     * @param lastStreamSequence the last received stream sequence
     * @param lastConsumerSequence the last received consumer sequence
     */
    default void heartbeatAlarm(Connection conn, JetStreamSubscription sub,
                                long lastStreamSequence, long lastConsumerSequence) {}

    /**
     * Called when an unhandled status is received in a push subscription.
     * @param conn The connection that had the issue
     * @param sub the JetStreamSubscription that this occurred on
     * @param status the status
     */
    default void unhandledStatus(Connection conn, JetStreamSubscription sub, Status status) {}

    /**
     * Called when a pull subscription receives a status message that indicates either
     * the subscription or pull might be problematic
     *
     * @param conn   The connection that had the issue
     * @param sub    the JetStreamSubscription that this occurred on
     * @param status the status
     */
    default void pullStatusWarning(Connection conn, JetStreamSubscription sub, Status status) {}

    /**
     * Called when a pull subscription receives a status message that indicates either
     * the subscription cannot continue or the pull request cannot be processed.
     *
     * @param conn   The connection that had the issue
     * @param sub    the JetStreamSubscription that this occurred on
     * @param status the status
     */
    default void pullStatusError(Connection conn, JetStreamSubscription sub, Status status) {}

    enum FlowControlSource { FLOW_CONTROL, HEARTBEAT }

    /**
     * Called by the connection when a flow control is processed.
     *
     * @param conn The connection that had the issue
     * @param sub the JetStreamSubscription that this occurred on
     * @param subject the flow control subject that was handled
     * @param source enum indicating flow control handling in response to which type of message
     */
    default void flowControlProcessed(Connection conn, JetStreamSubscription sub, String subject, FlowControlSource source) {}

    /**
     * Called by the connection when a low level socket write timeout occurs.
     *
     * @param conn The connection that had the issue
     */
    default void socketWriteTimeout(Connection conn) {}

    /**
     * General message producing function which understands the possible parameters to listener calls.
     * @param label the label for the message
     * @param conn The connection that had the issue, if provided.
     * @param consumer The consumer that is being marked slow, if applicable
     * @param sub the JetStreamSubscription that this occurred on, if applicable
     * @param pairs custom string pairs. I.E. "foo: ", fooObject, "bar-", barObject will be appended
     *              to the message like ", foo: &lt;fooValue&gt;, bar-&lt;barValue&gt;".
     * @return the message
     */
    default String supplyMessage(String label, Connection conn, Consumer consumer, Subscription sub, Object... pairs) {
        StringBuilder sb = new StringBuilder(label == null ? "" : label);
        if (conn != null) {
            ServerInfo si = conn.getServerInfo();
            if (si != null) {
                sb.append(", Connection: ").append(conn.getServerInfo().getClientId());
            }
        }
        if (consumer != null) {
            sb.append(", Consumer: ").append(consumer.hashCode());
        }
        if (sub != null) {
            sb.append(", Subscription: ").append(sub.hashCode());
            if (sub instanceof JetStreamSubscription) {
                JetStreamSubscription jssub = (JetStreamSubscription)sub;
                sb.append(", Consumer Name: ").append(jssub.getConsumerName());
            }
        }
        for (int x = 0; x < pairs.length; x++) {
            sb.append(", ").append(pairs[x]).append(pairs[++x]);
        }
        return sb.toString();
    }

}

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 io.nats.client.api.ServerInfo;
17	import io.nats.client.support.Status;
18
19	/**
20	* This library groups problems into four categories:
21	* <dl>
22	* <dt>Errors</dt>
23	* <dd>The server sent an error message using the {@code -err} protocol operation.</dd>
24	* <dt>Exceptions</dt>
25	* <dd>A Java exception occurred, and was handled by the library.</dd>
26	* <dt>Slow Consumers</dt>
27	* <dd>One of the connections consumers, Subscription or Dispatcher, is slow, and starting to drop messages.</dd>
28	* <dt>Fast Producers</dt>
29	* <dd>One of the connections producers is too fast, and is discarding messages</dd>
30	* </dl>
31	* <p>All of these problems are reported to the application code using the ErrorListener. The
32	* listener is configured in the {@link Options Options} at creation time.
33	*/
34	public interface ErrorListener {
35	/**
36	* NATs related errors that occur asynchronously in the client library are sent
37	* to an ErrorListener via errorOccurred. The ErrorListener can use the error text to decide what to do about the problem.
38	* <p>The text for an error is described in the protocol doc at `https://nats.io/documentation/internals/nats-protocol`.
39	* <p>In some cases the server will close the clients connection after sending one of these errors. In that case, the
40	* connections {@link ConnectionListener ConnectionListener} will be notified.
41	* @param conn The connection associated with the error
42	* @param error The text of error that has occurred, directly from the server
43	*/
44	default void errorOccurred(Connection conn, String error) {};	1✔
45
46	/**
47	* Exceptions that occur in the "normal" course of operations are sent to the
48	* ErrorListener using exceptionOccurred. Examples include, application exceptions
49	* during Dispatcher callbacks, IOExceptions from the underlying socket, etc..
50	* The library will try to handle these, via reconnect or catching them, but they are
51	* forwarded here in case the application code needs them for debugging purposes.
52	*
53	* @param conn The connection associated with the error
54	* @param exp The exception that has occurred, and was handled by the library
55	*/
56	default void exceptionOccurred(Connection conn, Exception exp) {};	1✔
57
58	/**
59	* Called by the connection when a "slow" consumer is detected. This call is only made once
60	* until the consumer stops being slow. At which point it will be called again if the consumer starts
61	* being slow again.
62	*
63	* <p>See {@link Consumer#setPendingLimits(long, long) Consumer.setPendingLimits}
64	* for information on how to configure when this method is fired.
65	*
66	* <p> Slow consumers will result in dropped messages each consumer provides a method
67	* for retrieving the count of dropped messages, see {@link Consumer#getDroppedCount() Consumer.getDroppedCount}.
68	*
69	* @param conn The connection associated with the error
70	* @param consumer The consumer that is being marked slow
71	*/
72	default void slowConsumerDetected(Connection conn, Consumer consumer) {};	1✔
73
74	/**
75	* Called by the connection when a message is discarded.
76	*
77	* @param conn The connection that discarded the message
78	* @param msg The message that is discarded
79	*/
80	default void messageDiscarded(Connection conn, Message msg) {}	1✔
81
82	/**
83	* Called when subscription heartbeats are missed according to the configured period and threshold.
84	* The consumer must be configured with an idle heartbeat time.
85	*
86	* @param conn The connection that had the issue
87	* @param sub the JetStreamSubscription that this occurred on
88	* @param lastStreamSequence the last received stream sequence
89	* @param lastConsumerSequence the last received consumer sequence
90	*/
91	default void heartbeatAlarm(Connection conn, JetStreamSubscription sub,
92	long lastStreamSequence, long lastConsumerSequence) {}	1✔
93
94	/**
95	* Called when an unhandled status is received in a push subscription.
96	* @param conn The connection that had the issue
97	* @param sub the JetStreamSubscription that this occurred on
98	* @param status the status
99	*/
100	default void unhandledStatus(Connection conn, JetStreamSubscription sub, Status status) {}	1✔
101
102	/**
103	* Called when a pull subscription receives a status message that indicates either
104	* the subscription or pull might be problematic
105	*
106	* @param conn The connection that had the issue
107	* @param sub the JetStreamSubscription that this occurred on
108	* @param status the status
109	*/
110	default void pullStatusWarning(Connection conn, JetStreamSubscription sub, Status status) {}	1✔
111
112	/**
113	* Called when a pull subscription receives a status message that indicates either
114	* the subscription cannot continue or the pull request cannot be processed.
115	*
116	* @param conn The connection that had the issue
117	* @param sub the JetStreamSubscription that this occurred on
118	* @param status the status
119	*/
120	default void pullStatusError(Connection conn, JetStreamSubscription sub, Status status) {}	1✔
121
122	enum FlowControlSource { FLOW_CONTROL, HEARTBEAT }	1✔
123
124	/**
125	* Called by the connection when a flow control is processed.
126	*
127	* @param conn The connection that had the issue
128	* @param sub the JetStreamSubscription that this occurred on
129	* @param subject the flow control subject that was handled
130	* @param source enum indicating flow control handling in response to which type of message
131	*/
132	default void flowControlProcessed(Connection conn, JetStreamSubscription sub, String subject, FlowControlSource source) {}	1✔
133
134	/**
135	* Called by the connection when a low level socket write timeout occurs.
136	*
137	* @param conn The connection that had the issue
138	*/
139	default void socketWriteTimeout(Connection conn) {}	×
140
141	/**
142	* General message producing function which understands the possible parameters to listener calls.
143	* @param label the label for the message
144	* @param conn The connection that had the issue, if provided.
145	* @param consumer The consumer that is being marked slow, if applicable
146	* @param sub the JetStreamSubscription that this occurred on, if applicable
147	* @param pairs custom string pairs. I.E. "foo: ", fooObject, "bar-", barObject will be appended
148	* to the message like ", foo: <fooValue>, bar-<barValue>".
149	* @return the message
150	*/
151	default String supplyMessage(String label, Connection conn, Consumer consumer, Subscription sub, Object... pairs) {
152	StringBuilder sb = new StringBuilder(label == null ? "" : label);	1✔
153	if (conn != null) {	1✔
154	ServerInfo si = conn.getServerInfo();	1✔
155	if (si != null) {	1✔
156	sb.append(", Connection: ").append(conn.getServerInfo().getClientId());	1✔
157	}
158	}
159	if (consumer != null) {	1✔
160	sb.append(", Consumer: ").append(consumer.hashCode());	1✔
161	}
162	if (sub != null) {	1✔
UNCOV 163	sb.append(", Subscription: ").append(sub.hashCode());	×
UNCOV 164	if (sub instanceof JetStreamSubscription) {	×
UNCOV 165	JetStreamSubscription jssub = (JetStreamSubscription)sub;	×
UNCOV 166	sb.append(", Consumer Name: ").append(jssub.getConsumerName());	×
167	}
168	}
169	for (int x = 0; x < pairs.length; x++) {	1✔
170	sb.append(", ").append(pairs[x]).append(pairs[++x]);	1✔
171	}
172	return sb.toString();	1✔
173	}
174
175	}

nats-io / nats.java / #2231

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