28916241081

Committed 08 Jul 2026 03:53AM UTC coverage: 89.468% (-0.7%) from 90.12%

Build # 28916241081

Build Type

push

github

Committed by

Hyshmily

Commit Message

refactor(core): extract central dispatcher and broadcast buffer

Introduce CentralDispatcher for centralized event routing and BroadcastBuffer for cache synchronization. Refactor HotKeyCache, SlidingWindowDetector, and related components to use the new dispatching mechanism, improving separation of concerns and reducing coupling.

Signed-off-by: Hyshmily <cxm8607@outlook.com>

Coverage Stats

1328 of 1564 branches covered (84.91%)

Branch coverage included in aggregate %.

102 of 134 new or added lines in 17 files covered. (76.12%)

3 existing lines in 2 files now uncovered.

3769 of 4133 relevant lines covered (91.19%)

4.21 hits per line

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

93.75

/worker/src/main/java/io/github/hyshmily/hotkey/worker/dispatch/WorkerHeartbeatProducer.java

/*
 * Copyright 2026 Hyshmily. All Rights Reserved.
 *
 * 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.github.hyshmily.hotkey.worker.dispatch;

import static io.github.hyshmily.hotkey.constants.HotKeyConstants.ROUTING_KEY_HEARTBEAT;

import io.github.hyshmily.hotkey.detection.HotKeyStateMachine;
import io.github.hyshmily.hotkey.sync.worker.WorkerHeartbeatMessage;
import io.github.hyshmily.hotkey.util.HotKeyThreadFactory;
import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;
import java.lang.management.ManagementFactory;
import java.lang.management.OperatingSystemMXBean;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import lombok.AccessLevel;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.annotation.DependsOn;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.StringRedisTemplate;

/**
 * Enhanced Worker heartbeat sender.
 *
 * <p>Replaces the old ping-only heartbeat approach with
 * structured heartbeats containing epoch, decisionVersionHwm, loadFactor,
 * and readyToServe flags. Broadcast on the dedicated heartbeat exchange.
 *
 * <p>Epoch is persisted to Redis (fallback to local file), incremented
 * atomically on each process start for restart detection by Apps.
 */
@RequiredArgsConstructor(access = AccessLevel.PRIVATE)
@Slf4j
@DependsOn("rabbitAdmin")
@SuppressWarnings("SpringDependsOnUnresolvedBeanInspection")
public class WorkerHeartbeatProducer {

  /** RabbitMQ template for publishing heartbeat messages. */
  private final RabbitTemplate rabbitTemplate;
  /** Target topic exchange for heartbeat messages. */
  private final String heartbeatExchange;
  /** Unique identity of this Worker node. */
  private final String workerId;
  /** State machine providing config-gossip fields (confirm/cool/grace counts). */
  private final HotKeyStateMachine stateMachine;
  /** Broadcaster for reading the current decision version watermark. */
  private final WorkerBroadcaster broadcaster;
  /** Monotonically increasing epoch, persisted across restarts. */
  private final long epoch;
  /** JVM start timestamp used for the ready-to-serve grace period. */
  private final long startTime;
  /** Scheduler for periodic heartbeat sends. */
  private final ScheduledExecutorService scheduler;
  /** Whether this instance owns the scheduler (self-created). */
  private final boolean ownsScheduler;
  /** Shared monotonic counter for config-change timestamps embedded in heartbeats. */
  private final AtomicLong configTimestampCounter;
  /** Interval between consecutive heartbeat sends (milliseconds). */
  private final long pingIntervalMs;
  /** Handle for the scheduled heartbeat task. */
  private ScheduledFuture<?> heartbeatTask;

  private static final String EPOCH_REDIS_KEY_PREFIX = "hotkey:worker:epoch:";

  /**
   * Creates a new heartbeat producer with a shared external scheduler.
   *
   * @param rabbitTemplate         the RabbitMQ template for publishing heartbeat messages
   * @param heartbeatExchange      the target topic exchange for heartbeat messages
   * @param workerId               unique identity of this Worker node
   * @param stateMachine           the state machine providing config-gossip fields
   * @param broadcaster            the broadcaster for reading the current decision version watermark
   * @param configTimestampCounter the shared monotonic counter for config-change timestamps
   * @param redisConnectionFactory the Redis connection factory for epoch initialization
   * @param pingIntervalMs         interval between consecutive heartbeat sends (milliseconds)
   * @param scheduler              the shared external scheduler for periodic heartbeat sends
   */
  public WorkerHeartbeatProducer(
    RabbitTemplate rabbitTemplate,
    String heartbeatExchange,
    String workerId,
    HotKeyStateMachine stateMachine,
    WorkerBroadcaster broadcaster,
    AtomicLong configTimestampCounter,
    RedisConnectionFactory redisConnectionFactory,
    long pingIntervalMs,
    ScheduledExecutorService scheduler
  ) {
    this(
      rabbitTemplate,
      heartbeatExchange,
      workerId,
      stateMachine,
      broadcaster,
      initEpoch(workerId, redisConnectionFactory),
      System.currentTimeMillis(),
      scheduler,
      false,
      configTimestampCounter,
      pingIntervalMs
    );
  }

  /**
   * Creates a new heartbeat producer with its own internal scheduler
   * (primarily for testing, with mocked Redis).
   *
   * @param rabbitTemplate         the RabbitMQ template for publishing heartbeat messages
   * @param heartbeatExchange      the target topic exchange for heartbeat messages
   * @param workerId               unique identity of this Worker node
   * @param stateMachine           the state machine providing config-gossip fields
   * @param broadcaster            the broadcaster for reading the current decision version watermark
   * @param configTimestampCounter the shared monotonic counter for config-change timestamps
   * @param pingIntervalMs         interval between consecutive heartbeat sends (milliseconds)
   */
  public WorkerHeartbeatProducer(
    RabbitTemplate rabbitTemplate,
    String heartbeatExchange,
    String workerId,
    HotKeyStateMachine stateMachine,
    WorkerBroadcaster broadcaster,
    AtomicLong configTimestampCounter,
    long pingIntervalMs
  ) {
    this(
      rabbitTemplate,
      heartbeatExchange,
      workerId,
      stateMachine,
      broadcaster,
      initEpoch(workerId, null),
      System.currentTimeMillis(),
      Executors.newSingleThreadScheduledExecutor(new HotKeyThreadFactory("hotkey-hb-producer")),
      true,
      configTimestampCounter,
      pingIntervalMs
    );
  }

  private static long initEpoch(String workerId, RedisConnectionFactory redisConnectionFactory) {
    StringRedisTemplate template = new StringRedisTemplate(redisConnectionFactory);
    template.afterPropertiesSet();
    return doInitEpoch(template, workerId);
  }

  /**
   * Initializes the epoch by atomically incrementing a Redis counter.
   * Falls back to a local file if Redis is unavailable.
   *
   * @param redis the Redis template (created ad-hoc for this one-time operation)
   * @return the new epoch value for this Worker incarnation
   */
  private static long doInitEpoch(StringRedisTemplate redis, String workerId) {
    try {
      String key = EPOCH_REDIS_KEY_PREFIX + workerId;
      String val = redis.opsForValue().get(key);
      long next = (val != null) ? Long.parseLong(val) + 1 : 1;
      redis.opsForValue().set(key, String.valueOf(next));
      log.info("Epoch initialized via Redis: workerId={}, epoch={}", workerId, next);
      return next;
    } catch (Exception e) {
      log.warn("Redis unavailable for epoch, using local file fallback", e);
      return initEpochFromLocalFile(workerId);
    }
  }

  /**
   * Fallback epoch initialization using a local temp file when Redis is
   * unavailable.  Reads the previous value, increments, and writes back.
   *
   * <p>If both Redis and local file fallback fail, falls back to
   * {@code System.currentTimeMillis()} as a last resort epoch value
   * (non-monotonic risk across restarts is logged at error level).
   *
   * @return the new epoch value, or {@code System.currentTimeMillis()} on complete failure
   */
  private static long initEpochFromLocalFile(String workerId) {
    try {
      Path path = Path.of(System.getProperty("java.io.tmpdir"), "hotkey-epoch-" + workerId);
      long next = 1;
      if (Files.exists(path)) {
        String content = Files.readString(path);
        next = Long.parseLong(content.trim()) + 1;
      }
      Files.writeString(path, String.valueOf(next));
      log.info("Epoch initialized via local file: workerId={}, epoch={}", workerId, next);
      return next;
    } catch (Exception e) {
      log.error("Epoch local file fallback failed, using System.currentTimeMillis()", e);
      return System.currentTimeMillis();
    }
  }

  /**
   * Computes the current CPU load factor (0.0 – 1.0) via the platform MXBean.
   *
   * @return CPU load clamped to [0, 1], or 0.0 if unavailable
   */
  private double computeLoadFactor() {
    double cpuLoad = 0.0;
    try {
      OperatingSystemMXBean osBean = ManagementFactory.getOperatingSystemMXBean();
      if (osBean instanceof com.sun.management.OperatingSystemMXBean sunOsBean) {
        cpuLoad = sunOsBean.getCpuLoad();
      }
    } catch (Exception e) {
      log.warn("Failed to compute CPU load factor; defaulting to 0.0", e);
    }
    return Math.min(1.0, cpuLoad);
  }

  /**
   * Whether this Worker is ready to serve hot-key detection.
   * Returns {@code true} after a 3-second startup grace period or as soon as
   * the state machine has tracked at least one key.
   *
   * @return {@code true} if the Worker considers itself ready
   */
  private boolean isReadyToServe() {
    long uptime = System.currentTimeMillis() - startTime;
    return uptime > 3000 || stateMachine.getTrackedKeys() > 0;
  }

  /**
   * Starts the periodic heartbeat sender at the configured ping interval.
   *
   * <p>First heartbeat is delayed by {@code pingIntervalMs} to allow the
   * RabbitMQ connection and exchange declarations (RabbitAdmin) to complete,
   * preventing channel-level NOT_FOUND errors when the heartbeat exchange
   * has not yet been declared.
   */
  @PostConstruct
  public void start() {
    try {
      heartbeatTask = scheduler.scheduleAtFixedRate(
        this::sendHeartbeat,
        pingIntervalMs,
        pingIntervalMs,
        TimeUnit.MILLISECONDS
      );
    } catch (Exception e) {
      log.error(
        "Failed to start heartbeat scheduler; Worker heartbeat will not be sent. " +
          "Application continues but App instances may mark this Worker as dead.",
        e
      );
    }
  }

  /**
   * Gracefully stops the heartbeat sender.
   *
   * <p>Cancels the scheduled task; shuts down the scheduler only if owned.
   */
  @PreDestroy
  public void stop() {
    if (heartbeatTask != null) {
      heartbeatTask.cancel(false);
    }
    if (ownsScheduler) {
      scheduler.shutdown();
    }
  }

  /**
   * Builds and sends a single heartbeat message containing epoch, decision
   * version watermark, load factor, ready-to-serve flag, config fingerprint,
   * and state-machine config gossip fields.
   *
   * <p>Published to the configured heartbeat exchange with routing key
   * {@code heartbeat.<workerId>}.  Silently drops if the channel or
   * connection is unavailable (fire-and-forget).
   */
  void sendHeartbeat() {
    try {
      WorkerHeartbeatMessage hb = new WorkerHeartbeatMessage(
        workerId,
        epoch,
        broadcaster.getCurrentDecisionVersion(),
        computeLoadFactor(),
        isReadyToServe(),
        stateMachine.getConfirmCount(),
        stateMachine.getCoolCount(),
        stateMachine.getPreCoolGraceCount(),
        configTimestampCounter.get()
      );
      rabbitTemplate.send(heartbeatExchange, ROUTING_KEY_HEARTBEAT + workerId, hb.toMessage());
    } catch (Exception e) {
      log.error("Scheduled sendHeartbeat failed", e);
    }
  }
}

1	/*
2	* Copyright 2026 Hyshmily. All Rights Reserved.
3	*
4	* Licensed under the Apache License, Version 2.0 (the "License");
5	* you may not use this file except in compliance with the License.
6	* You may obtain a copy of the License at
7	*
8	* http://www.apache.org/licenses/LICENSE-2.0
9	*
10	* Unless required by applicable law or agreed to in writing, software
11	* distributed under the License is distributed on an "AS IS" BASIS,
12	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13	* See the License for the specific language governing permissions and
14	* limitations under the License.
15	*/
16	package io.github.hyshmily.hotkey.worker.dispatch;
17
18	import static io.github.hyshmily.hotkey.constants.HotKeyConstants.ROUTING_KEY_HEARTBEAT;
19
20	import io.github.hyshmily.hotkey.detection.HotKeyStateMachine;
21	import io.github.hyshmily.hotkey.sync.worker.WorkerHeartbeatMessage;
22	import io.github.hyshmily.hotkey.util.HotKeyThreadFactory;
23	import jakarta.annotation.PostConstruct;
24	import jakarta.annotation.PreDestroy;
25	import java.lang.management.ManagementFactory;
26	import java.lang.management.OperatingSystemMXBean;
27	import java.nio.file.Files;
28	import java.nio.file.Path;
29	import java.util.concurrent.Executors;
30	import java.util.concurrent.ScheduledExecutorService;
31	import java.util.concurrent.ScheduledFuture;
32	import java.util.concurrent.TimeUnit;
33	import java.util.concurrent.atomic.AtomicLong;
34	import lombok.AccessLevel;
35	import lombok.RequiredArgsConstructor;
36	import lombok.extern.slf4j.Slf4j;
37	import org.springframework.amqp.rabbit.core.RabbitTemplate;
38	import org.springframework.context.annotation.DependsOn;
39	import org.springframework.data.redis.connection.RedisConnectionFactory;
40	import org.springframework.data.redis.core.StringRedisTemplate;
41
42	/**
43	* Enhanced Worker heartbeat sender.
44	*
45	* <p>Replaces the old ping-only heartbeat approach with
46	* structured heartbeats containing epoch, decisionVersionHwm, loadFactor,
47	* and readyToServe flags. Broadcast on the dedicated heartbeat exchange.
48	*
49	* <p>Epoch is persisted to Redis (fallback to local file), incremented
50	* atomically on each process start for restart detection by Apps.
51	*/
52	@RequiredArgsConstructor(access = AccessLevel.PRIVATE)
53	@Slf4j	4✔
54	@DependsOn("rabbitAdmin")
55	@SuppressWarnings("SpringDependsOnUnresolvedBeanInspection")
56	public class WorkerHeartbeatProducer {
57
58	/** RabbitMQ template for publishing heartbeat messages. */
59	private final RabbitTemplate rabbitTemplate;
60	/** Target topic exchange for heartbeat messages. */
61	private final String heartbeatExchange;
62	/** Unique identity of this Worker node. */
63	private final String workerId;
64	/** State machine providing config-gossip fields (confirm/cool/grace counts). */
65	private final HotKeyStateMachine stateMachine;
66	/** Broadcaster for reading the current decision version watermark. */
67	private final WorkerBroadcaster broadcaster;
68	/** Monotonically increasing epoch, persisted across restarts. */
69	private final long epoch;
70	/** JVM start timestamp used for the ready-to-serve grace period. */
71	private final long startTime;
72	/** Scheduler for periodic heartbeat sends. */
73	private final ScheduledExecutorService scheduler;
74	/** Whether this instance owns the scheduler (self-created). */
75	private final boolean ownsScheduler;
76	/** Shared monotonic counter for config-change timestamps embedded in heartbeats. */
77	private final AtomicLong configTimestampCounter;
78	/** Interval between consecutive heartbeat sends (milliseconds). */
79	private final long pingIntervalMs;
80	/** Handle for the scheduled heartbeat task. */
81	private ScheduledFuture<?> heartbeatTask;
82
83	private static final String EPOCH_REDIS_KEY_PREFIX = "hotkey:worker:epoch:";
84
85	/**
86	* Creates a new heartbeat producer with a shared external scheduler.
87	*
88	* @param rabbitTemplate the RabbitMQ template for publishing heartbeat messages
89	* @param heartbeatExchange the target topic exchange for heartbeat messages
90	* @param workerId unique identity of this Worker node
91	* @param stateMachine the state machine providing config-gossip fields
92	* @param broadcaster the broadcaster for reading the current decision version watermark
93	* @param configTimestampCounter the shared monotonic counter for config-change timestamps
94	* @param redisConnectionFactory the Redis connection factory for epoch initialization
95	* @param pingIntervalMs interval between consecutive heartbeat sends (milliseconds)
96	* @param scheduler the shared external scheduler for periodic heartbeat sends
97	*/
98	public WorkerHeartbeatProducer(
99	RabbitTemplate rabbitTemplate,
100	String heartbeatExchange,
101	String workerId,
102	HotKeyStateMachine stateMachine,
103	WorkerBroadcaster broadcaster,
104	AtomicLong configTimestampCounter,
105	RedisConnectionFactory redisConnectionFactory,
106	long pingIntervalMs,
107	ScheduledExecutorService scheduler
108	) {
109	this(	9✔
110	rabbitTemplate,
111	heartbeatExchange,
112	workerId,
113	stateMachine,
114	broadcaster,
115	initEpoch(workerId, redisConnectionFactory),	1✔
116	System.currentTimeMillis(),	5✔
117	scheduler,
118	false,
119	configTimestampCounter,
120	pingIntervalMs
121	);
122	}	1✔
123
124	/**
125	* Creates a new heartbeat producer with its own internal scheduler
126	* (primarily for testing, with mocked Redis).
127	*
128	* @param rabbitTemplate the RabbitMQ template for publishing heartbeat messages
129	* @param heartbeatExchange the target topic exchange for heartbeat messages
130	* @param workerId unique identity of this Worker node
131	* @param stateMachine the state machine providing config-gossip fields
132	* @param broadcaster the broadcaster for reading the current decision version watermark
133	* @param configTimestampCounter the shared monotonic counter for config-change timestamps
134	* @param pingIntervalMs interval between consecutive heartbeat sends (milliseconds)
135	*/
136	public WorkerHeartbeatProducer(
137	RabbitTemplate rabbitTemplate,
138	String heartbeatExchange,
139	String workerId,
140	HotKeyStateMachine stateMachine,
141	WorkerBroadcaster broadcaster,
142	AtomicLong configTimestampCounter,
143	long pingIntervalMs
144	) {
145	this(	9✔
146	rabbitTemplate,
147	heartbeatExchange,
148	workerId,
149	stateMachine,
150	broadcaster,
151	initEpoch(workerId, null),	1✔
152	System.currentTimeMillis(),	5✔
153	Executors.newSingleThreadScheduledExecutor(new HotKeyThreadFactory("hotkey-hb-producer")),	4✔
154	true,
155	configTimestampCounter,
156	pingIntervalMs
157	);
158	}	1✔
159
160	private static long initEpoch(String workerId, RedisConnectionFactory redisConnectionFactory) {
161	StringRedisTemplate template = new StringRedisTemplate(redisConnectionFactory);	5✔
162	template.afterPropertiesSet();	2✔
163	return doInitEpoch(template, workerId);	4✔
164	}
165
166	/**
167	* Initializes the epoch by atomically incrementing a Redis counter.
168	* Falls back to a local file if Redis is unavailable.
169	*
170	* @param redis the Redis template (created ad-hoc for this one-time operation)
171	* @return the new epoch value for this Worker incarnation
172	*/
173	private static long doInitEpoch(StringRedisTemplate redis, String workerId) {
174	try {
175	String key = EPOCH_REDIS_KEY_PREFIX + workerId;	3✔
176	String val = redis.opsForValue().get(key);	6✔
177	long next = (val != null) ? Long.parseLong(val) + 1 : 1;	9✔
178	redis.opsForValue().set(key, String.valueOf(next));	6✔
179	log.info("Epoch initialized via Redis: workerId={}, epoch={}", workerId, next);	6✔
180	return next;	2✔
181	} catch (Exception e) {	1✔
182	log.warn("Redis unavailable for epoch, using local file fallback", e);	4✔
183	return initEpochFromLocalFile(workerId);	3✔
184	}
185	}
186
187	/**
188	* Fallback epoch initialization using a local temp file when Redis is
189	* unavailable. Reads the previous value, increments, and writes back.
190	*
191	* <p>If both Redis and local file fallback fail, falls back to
192	* {@code System.currentTimeMillis()} as a last resort epoch value
193	* (non-monotonic risk across restarts is logged at error level).
194	*
195	* @return the new epoch value, or {@code System.currentTimeMillis()} on complete failure
196	*/
197	private static long initEpochFromLocalFile(String workerId) {
198	try {
199	Path path = Path.of(System.getProperty("java.io.tmpdir"), "hotkey-epoch-" + workerId);	11✔
200	long next = 1;	2✔
201	if (Files.exists(path)) {	5✔
202	String content = Files.readString(path);	3✔
203	next = Long.parseLong(content.trim()) + 1;	6✔
204	}
205	Files.writeString(path, String.valueOf(next));	7✔
206	log.info("Epoch initialized via local file: workerId={}, epoch={}", workerId, next);	6✔
207	return next;	2✔
208	} catch (Exception e) {	1✔
209	log.error("Epoch local file fallback failed, using System.currentTimeMillis()", e);	4✔
210	return System.currentTimeMillis();	2✔
211	}
212	}
213
214	/**
215	* Computes the current CPU load factor (0.0 – 1.0) via the platform MXBean.
216	*
217	* @return CPU load clamped to [0, 1], or 0.0 if unavailable
218	*/
219	private double computeLoadFactor() {
220	double cpuLoad = 0.0;	2✔
221	try {
222	OperatingSystemMXBean osBean = ManagementFactory.getOperatingSystemMXBean();	2✔
223	if (osBean instanceof com.sun.management.OperatingSystemMXBean sunOsBean) {	6!
224	cpuLoad = sunOsBean.getCpuLoad();	3✔
225	}
NEW 226	} catch (Exception e) {	×
NEW 227	log.warn("Failed to compute CPU load factor; defaulting to 0.0", e);	×
228	}	1✔
229	return Math.min(1.0, cpuLoad);	4✔
230	}
231
232	/**
233	* Whether this Worker is ready to serve hot-key detection.
234	* Returns {@code true} after a 3-second startup grace period or as soon as
235	* the state machine has tracked at least one key.
236	*
237	* @return {@code true} if the Worker considers itself ready
238	*/
239	private boolean isReadyToServe() {
240	long uptime = System.currentTimeMillis() - startTime;	5✔
241	return uptime > 3000 \|\| stateMachine.getTrackedKeys() > 0;	12✔
242	}
243
244	/**
245	* Starts the periodic heartbeat sender at the configured ping interval.
246	*
247	* <p>First heartbeat is delayed by {@code pingIntervalMs} to allow the
248	* RabbitMQ connection and exchange declarations (RabbitAdmin) to complete,
249	* preventing channel-level NOT_FOUND errors when the heartbeat exchange
250	* has not yet been declared.
251	*/
252	@PostConstruct
253	public void start() {
254	try {
255	heartbeatTask = scheduler.scheduleAtFixedRate(	12✔
256	this::sendHeartbeat,
257	pingIntervalMs,
258	pingIntervalMs,
259	TimeUnit.MILLISECONDS
260	);
261	} catch (Exception e) {	×
262	log.error(	×
263	"Failed to start heartbeat scheduler; Worker heartbeat will not be sent. " +
264	"Application continues but App instances may mark this Worker as dead.",
265	e
266	);
267	}	1✔
268	}	1✔
269
270	/**
271	* Gracefully stops the heartbeat sender.
272	*
273	* <p>Cancels the scheduled task; shuts down the scheduler only if owned.
274	*/
275	@PreDestroy
276	public void stop() {
277	if (heartbeatTask != null) {	3✔
278	heartbeatTask.cancel(false);	5✔
279	}
280	if (ownsScheduler) {	3✔
281	scheduler.shutdown();	3✔
282	}
283	}	1✔
284
285	/**
286	* Builds and sends a single heartbeat message containing epoch, decision
287	* version watermark, load factor, ready-to-serve flag, config fingerprint,
288	* and state-machine config gossip fields.
289	*
290	* <p>Published to the configured heartbeat exchange with routing key
291	* {@code heartbeat.<workerId>}. Silently drops if the channel or
292	* connection is unavailable (fire-and-forget).
293	*/
294	void sendHeartbeat() {
295	try {
296	WorkerHeartbeatMessage hb = new WorkerHeartbeatMessage(	8✔
297	workerId,
298	epoch,
299	broadcaster.getCurrentDecisionVersion(),	2✔
300	computeLoadFactor(),	2✔
301	isReadyToServe(),	3✔
302	stateMachine.getConfirmCount(),	3✔
303	stateMachine.getCoolCount(),	3✔
304	stateMachine.getPreCoolGraceCount(),	3✔
305	configTimestampCounter.get()	3✔
306	);
307	rabbitTemplate.send(heartbeatExchange, ROUTING_KEY_HEARTBEAT + workerId, hb.toMessage());	10✔
308	} catch (Exception e) {	1✔
309	log.error("Scheduled sendHeartbeat failed", e);	4✔
310	}	1✔
311	}	1✔
312	}

Hyshmily / hotkey / 28916241081

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