#2059

Committed 15 Jul 2025 06:40PM UTC coverage: 95.593% (-0.06%) from 95.655%

Build # #2059

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #1357 from nats-io/shutdown-internal-connection-executors

[FIX] Shutdown internal executors on connection close.

Run Details

19 of 19 new or added lines in 2 files covered. (100.0%)

16 existing lines in 7 files now uncovered.

11844 of 12390 relevant lines covered (95.59%)

0.96 hits per line

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

90.91

/src/main/java/io/nats/client/impl/ErrorListenerLoggerImpl.java

// Copyright 2021 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.impl;

import io.nats.client.*;
import io.nats.client.support.Status;

import java.util.logging.Logger;

public class ErrorListenerLoggerImpl implements ErrorListener {

    private final static Logger LOGGER = Logger.getLogger(ErrorListenerLoggerImpl.class.getName());

    /**
     * {@inheritDoc}
     */
    @Override
    public void errorOccurred(final Connection conn, final String error) {
        LOGGER.severe(() -> supplyMessage("errorOccurred", conn, null, null, "Error: ", error));
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void exceptionOccurred(final Connection conn, final Exception exp) {
        LOGGER.severe(() -> supplyMessage("exceptionOccurred", conn, null, null, "Exception: ", exp));
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void slowConsumerDetected(final Connection conn, final Consumer consumer) {
        LOGGER.warning(() -> supplyMessage("slowConsumerDetected", conn, consumer, null));
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void messageDiscarded(final Connection conn, final Message msg) {
        LOGGER.info(() -> supplyMessage("messageDiscarded", conn, null, null, "Message: ", msg));
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void heartbeatAlarm(final Connection conn, final JetStreamSubscription sub,
                               final long lastStreamSequence, final long lastConsumerSequence) {
        LOGGER.severe(() -> supplyMessage("heartbeatAlarm", conn, null, sub, "lastStreamSequence: ", lastStreamSequence, "lastConsumerSequence: ", lastConsumerSequence));
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void unhandledStatus(final Connection conn, final JetStreamSubscription sub, final Status status) {
        LOGGER.warning(() -> supplyMessage("unhandledStatus", conn, null, sub, "Status:", status));
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void pullStatusWarning(Connection conn, JetStreamSubscription sub, Status status) {
        LOGGER.warning(() -> supplyMessage("pullStatusWarning", conn, null, sub, "Status:", status));
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void pullStatusError(Connection conn, JetStreamSubscription sub, Status status) {
        LOGGER.severe(() -> supplyMessage("pullStatusError", conn, null, sub, "Status:", status));
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void flowControlProcessed(Connection conn, JetStreamSubscription sub, String id, FlowControlSource source) {
        LOGGER.info(() -> supplyMessage("flowControlProcessed", conn, null, sub, "FlowControlSource:", source));
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void socketWriteTimeout(Connection conn) {
        LOGGER.severe(() -> supplyMessage("socketWriteTimeout", conn, null, null));
    }
}

1	// Copyright 2021 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.impl;
15
16	import io.nats.client.*;
17	import io.nats.client.support.Status;
18
19	import java.util.logging.Logger;
20
21	public class ErrorListenerLoggerImpl implements ErrorListener {	1✔
22
23	private final static Logger LOGGER = Logger.getLogger(ErrorListenerLoggerImpl.class.getName());	1✔
24
25	/**
26	* {@inheritDoc}
27	*/
28	@Override
29	public void errorOccurred(final Connection conn, final String error) {
30	LOGGER.severe(() -> supplyMessage("errorOccurred", conn, null, null, "Error: ", error));	1✔
31	}	1✔
32
33	/**
34	* {@inheritDoc}
35	*/
36	@Override
37	public void exceptionOccurred(final Connection conn, final Exception exp) {
38	LOGGER.severe(() -> supplyMessage("exceptionOccurred", conn, null, null, "Exception: ", exp));	1✔
39	}	1✔
40
41	/**
42	* {@inheritDoc}
43	*/
44	@Override
45	public void slowConsumerDetected(final Connection conn, final Consumer consumer) {
46	LOGGER.warning(() -> supplyMessage("slowConsumerDetected", conn, consumer, null));	1✔
47	}	1✔
48
49	/**
50	* {@inheritDoc}
51	*/
52	@Override
53	public void messageDiscarded(final Connection conn, final Message msg) {
54	LOGGER.info(() -> supplyMessage("messageDiscarded", conn, null, null, "Message: ", msg));	1✔
55	}	1✔
56
57	/**
58	* {@inheritDoc}
59	*/
60	@Override
61	public void heartbeatAlarm(final Connection conn, final JetStreamSubscription sub,
62	final long lastStreamSequence, final long lastConsumerSequence) {
63	LOGGER.severe(() -> supplyMessage("heartbeatAlarm", conn, null, sub, "lastStreamSequence: ", lastStreamSequence, "lastConsumerSequence: ", lastConsumerSequence));	1✔
64	}	1✔
65
66	/**
67	* {@inheritDoc}
68	*/
69	@Override
70	public void unhandledStatus(final Connection conn, final JetStreamSubscription sub, final Status status) {
71	LOGGER.warning(() -> supplyMessage("unhandledStatus", conn, null, sub, "Status:", status));	1✔
72	}	1✔
73
74	/**
75	* {@inheritDoc}
76	*/
77	@Override
78	public void pullStatusWarning(Connection conn, JetStreamSubscription sub, Status status) {
79	LOGGER.warning(() -> supplyMessage("pullStatusWarning", conn, null, sub, "Status:", status));	1✔
80	}	1✔
81
82	/**
83	* {@inheritDoc}
84	*/
85	@Override
86	public void pullStatusError(Connection conn, JetStreamSubscription sub, Status status) {
87	LOGGER.severe(() -> supplyMessage("pullStatusError", conn, null, sub, "Status:", status));	1✔
88	}	1✔
89
90	/**
91	* {@inheritDoc}
92	*/
93	@Override
94	public void flowControlProcessed(Connection conn, JetStreamSubscription sub, String id, FlowControlSource source) {
95	LOGGER.info(() -> supplyMessage("flowControlProcessed", conn, null, sub, "FlowControlSource:", source));	1✔
96	}	1✔
97
98	/**
99	* {@inheritDoc}
100	*/
101	@Override
102	public void socketWriteTimeout(Connection conn) {
UNCOV 103	LOGGER.severe(() -> supplyMessage("socketWriteTimeout", conn, null, null));	×
UNCOV 104	}	×
105	}

nats-io / nats.java / #2059

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