#2042

Committed 04 Jul 2025 01:42PM UTC coverage: 95.608% (-0.003%) from 95.611%

Build # #2042

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #1348 from nats-io/fix-what-i-broke-2-21-3

Fix heartbeat timer handling broken when replacing timer with scheduler.

Run Details

44 of 50 new or added lines in 7 files covered. (88.0%)

6 existing lines in 4 files now uncovered.

11798 of 12340 relevant lines covered (95.61%)

0.96 hits per line

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

72.0

/src/main/java/io/nats/client/support/ScheduledTask.java

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

import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;


/**
 * !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!! *
 * WARNING: THIS CLASS IS PUBLIC BUT ITS API IS NOT GUARANTEED TO *
 * BE BACKWARD COMPATIBLE AS IT IS INTENDED AS AN INTERNAL CLASS  *
 * !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!! *
 */
public class ScheduledTask implements Runnable {
    private static final AtomicLong ID_GENERATOR = new AtomicLong();

    private final String id;
    private final Runnable runnable;
    protected final AtomicReference<ScheduledFuture<?>> scheduledFutureRef;

    protected final AtomicBoolean notShutdown;
    protected final AtomicBoolean executing;
    protected final long initialDelayNanos;
    protected final long periodNanos;

    public ScheduledTask(ScheduledExecutorService ses, long initialAndPeriodMillis, Runnable runnable) {
        this(null, ses, initialAndPeriodMillis, initialAndPeriodMillis, TimeUnit.MILLISECONDS, runnable);
    }

    public ScheduledTask(String id, ScheduledExecutorService ses, long initialAndPeriodMillis, Runnable runnable) {
        this(id, ses, initialAndPeriodMillis, initialAndPeriodMillis, TimeUnit.MILLISECONDS, runnable);
    }

    public ScheduledTask(ScheduledExecutorService ses, long initialAndPeriod, TimeUnit unit, Runnable runnable) {
        this(null, ses, initialAndPeriod, initialAndPeriod, unit, runnable);
    }

    public ScheduledTask(String id, ScheduledExecutorService ses, long initialAndPeriod, TimeUnit unit, Runnable runnable) {
        this(id, ses, initialAndPeriod, initialAndPeriod, unit, runnable);
    }

    public ScheduledTask(ScheduledExecutorService ses, long initialDelay, long period, TimeUnit unit, Runnable runnable) {
        this(null, ses, initialDelay, period, unit, runnable);
    }

    public ScheduledTask(String id, ScheduledExecutorService ses, long initialDelay, long period, TimeUnit unit, Runnable runnable) {
        this.id = id == null || id.isEmpty() ? "st-" + ID_GENERATOR.getAndIncrement() : id;
        this.runnable = runnable;
        notShutdown = new AtomicBoolean(true);
        executing = new AtomicBoolean(false);
        this.initialDelayNanos = unit.toNanos(initialDelay);
        this.periodNanos = unit.toNanos(period);
        scheduledFutureRef = new AtomicReference<>(
            ses.scheduleAtFixedRate(this, initialDelay, period, unit));
    }

    public long getInitialDelayNanos() {
        return initialDelayNanos;
    }

    public long getPeriodNanos() {
        return periodNanos;
    }

    @Override
    public void run() {
        try {
            if (notShutdown.get()) {
                executing.set(true);
                runnable.run();
            }
        }
        finally {
            executing.set(false);
        }
    }

    public boolean isShutdown() {
        return !notShutdown.get();
    }

    public boolean isExecuting() {
        return executing.get();
    }

    public boolean isDone() {
        ScheduledFuture<?> f = scheduledFutureRef.get();
        return f == null || f.isDone();
    }

    public String getId() {
        return id;
    }

    public void shutdown() {
        try {
            notShutdown.set(false);
            ScheduledFuture<?> f = scheduledFutureRef.get();
            if (f != null) {
                scheduledFutureRef.set(null); // just releasing resources.
                if (!f.isDone()) {
                    f.cancel(false);
                }
            }
        }
        catch (Exception ignore) {
            // don't want this to be passed along
        }
    }

    @Override
    public String toString() {
        StringBuilder sb = new StringBuilder(id);
        if (notShutdown.get()) {
            sb.append(" [live");
        }
        else {
            sb.append(" [shutdown");
        }
        sb.append(isDone() ? "/done" : "/!done");
        sb.append(executing.get() ? "/executing" : "/!executing");
        sb.append("]");
        return sb.toString();
    }
}

1	// Copyright 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.support;
15
16	import java.util.concurrent.ScheduledExecutorService;
17	import java.util.concurrent.ScheduledFuture;
18	import java.util.concurrent.TimeUnit;
19	import java.util.concurrent.atomic.AtomicBoolean;
20	import java.util.concurrent.atomic.AtomicLong;
21	import java.util.concurrent.atomic.AtomicReference;
22
23
24	/**
25	* !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!! *
26	* WARNING: THIS CLASS IS PUBLIC BUT ITS API IS NOT GUARANTEED TO *
27	* BE BACKWARD COMPATIBLE AS IT IS INTENDED AS AN INTERNAL CLASS *
28	* !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!! *
29	*/
30	public class ScheduledTask implements Runnable {
31	private static final AtomicLong ID_GENERATOR = new AtomicLong();	1✔
32
33	private final String id;
34	private final Runnable runnable;
35	protected final AtomicReference<ScheduledFuture<?>> scheduledFutureRef;
36
37	protected final AtomicBoolean notShutdown;
38	protected final AtomicBoolean executing;
39	protected final long initialDelayNanos;
40	protected final long periodNanos;
41
42	public ScheduledTask(ScheduledExecutorService ses, long initialAndPeriodMillis, Runnable runnable) {
43	this(null, ses, initialAndPeriodMillis, initialAndPeriodMillis, TimeUnit.MILLISECONDS, runnable);	1✔
44	}	1✔
45
46	public ScheduledTask(String id, ScheduledExecutorService ses, long initialAndPeriodMillis, Runnable runnable) {
47	this(id, ses, initialAndPeriodMillis, initialAndPeriodMillis, TimeUnit.MILLISECONDS, runnable);	×
48	}	×
49
50	public ScheduledTask(ScheduledExecutorService ses, long initialAndPeriod, TimeUnit unit, Runnable runnable) {
51	this(null, ses, initialAndPeriod, initialAndPeriod, unit, runnable);	1✔
52	}	1✔
53
54	public ScheduledTask(String id, ScheduledExecutorService ses, long initialAndPeriod, TimeUnit unit, Runnable runnable) {
55	this(id, ses, initialAndPeriod, initialAndPeriod, unit, runnable);	×
56	}	×
57
58	public ScheduledTask(ScheduledExecutorService ses, long initialDelay, long period, TimeUnit unit, Runnable runnable) {
59	this(null, ses, initialDelay, period, unit, runnable);	1✔
60	}	1✔
61
62	public ScheduledTask(String id, ScheduledExecutorService ses, long initialDelay, long period, TimeUnit unit, Runnable runnable) {	1✔
63	this.id = id == null \|\| id.isEmpty() ? "st-" + ID_GENERATOR.getAndIncrement() : id;	1✔
64	this.runnable = runnable;	1✔
65	notShutdown = new AtomicBoolean(true);	1✔
66	executing = new AtomicBoolean(false);	1✔
67	this.initialDelayNanos = unit.toNanos(initialDelay);	1✔
68	this.periodNanos = unit.toNanos(period);	1✔
69	scheduledFutureRef = new AtomicReference<>(	1✔
70	ses.scheduleAtFixedRate(this, initialDelay, period, unit));	1✔
71	}	1✔
72
73	public long getInitialDelayNanos() {
74	return initialDelayNanos;	1✔
75	}
76
77	public long getPeriodNanos() {
78	return periodNanos;	1✔
79	}
80
81	@Override
82	public void run() {
83	try {
84	if (notShutdown.get()) {	1✔
85	executing.set(true);	1✔
86	runnable.run();	1✔
87	}
88	}
89	finally {
90	executing.set(false);	1✔
91	}
92	}	1✔
93
94	public boolean isShutdown() {
95	return !notShutdown.get();	1✔
96	}
97
98	public boolean isExecuting() {
99	return executing.get();	1✔
100	}
101
102	public boolean isDone() {
103	ScheduledFuture<?> f = scheduledFutureRef.get();	1✔
104	return f == null \|\| f.isDone();	1✔
105	}
106
107	public String getId() {
108	return id;	×
109	}
110
111	public void shutdown() {
112	try {
113	notShutdown.set(false);	1✔
114	ScheduledFuture<?> f = scheduledFutureRef.get();	1✔
115	if (f != null) {	1✔
116	scheduledFutureRef.set(null); // just releasing resources.	1✔
117	if (!f.isDone()) {	1✔
118	f.cancel(false);	1✔
119	}
120	}
121	}
122	catch (Exception ignore) {	×
123	// don't want this to be passed along
124	}	1✔
125	}	1✔
126
127	@Override
128	public String toString() {
129	StringBuilder sb = new StringBuilder(id);	×
130	if (notShutdown.get()) {	×
131	sb.append(" [live");	×
132	}
133	else {
134	sb.append(" [shutdown");	×
135	}
NEW 136	sb.append(isDone() ? "/done" : "/!done");	×
NEW 137	sb.append(executing.get() ? "/executing" : "/!executing");	×
138	sb.append("]");	×
139	return sb.toString();	×
140	}
141	}

nats-io / nats.java / #2042

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