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

91.49

/worker/src/main/java/io/github/hyshmily/hotkey/worker/config/WorkerConfigNegotiator.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.config;

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 java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;

/**
 * Listens for heartbeat-based config updates from peer Workers and applies them
 * if the received config timestamp is newer than the local one.
 *
 * <p>On startup, waits up to 3 seconds for the first heartbeat to arrive.
 * If none is received, the Worker continues with the values from
 * {@link io.github.hyshmily.hotkey.worker.config.WorkerProperties} — this can happen when all other Workers are down.
 */
/** Default constructor. */
@RequiredArgsConstructor
@Slf4j
public class WorkerConfigNegotiator {

  /** State machine whose config (confirm/cool/grace counts) is updated from heartbeat messages. */
  private final HotKeyStateMachine stateMachine;
  /** Monotonically increasing counter tracking the latest config-change timestamp. */
  private final AtomicLong configTimestampCounter;
  /** Unique identifier for this Worker node, used in queue names and heartbeat identification. */
  private final String nodeId;
  private final CountDownLatch startupLatch = new CountDownLatch(1);

  /**
   * Waits for the first heartbeat from a peer Worker.
   *
   * <p>If a heartbeat with a valid config arrives within 3 seconds the
   * config is applied synchronously.  Otherwise, a warning is logged and
   * the Worker proceeds with configured defaults.
   */
  @PostConstruct
  void syncOnStartup() {
    Thread waitThread = new HotKeyThreadFactory("hotkey-config-sync-startup").newThread(() -> {
      try {
        boolean received = startupLatch.await(3000, TimeUnit.MILLISECONDS);
        if (!received) {
          log.warn("No config heartbeat received within 3s, using WorkerProperties defaults");
        }
      } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
      }
    });
    waitThread.start();
  }

  /**
   * Processes an incoming heartbeat message from a peer Worker and applies
   * its state-machine config if the embedded timestamp is newer.
   *
   * @param msg the raw AMQP heartbeat message
   */
  @RabbitListener(queues = "#{@workerConfigQueue.name}", containerFactory = "workerConfigListenerContainerFactory")
  public void onHeartbeat(Message msg) {
    try {
      doOnHeartbeat(msg);
    } catch (Exception e) {
      log.warn("Uncaught exception in onHeartbeat config negotiation, discarding message to prevent requeue loop", e);
    }
  }

  private void doOnHeartbeat(Message msg) {
    WorkerHeartbeatMessage hb = WorkerHeartbeatMessage.from(msg);
    if (hb == null) {
      return;
    }

    if (hb.workerId().equals(nodeId)) {
      return;
    }

    long remoteTs = hb.configTimestamp();
    long localTs = configTimestampCounter.get();
    if (remoteTs <= localTs) {
      return;
    }

    stateMachine.setConfirmCount(hb.configConfirmCount());
    stateMachine.setCoolCount(hb.configCoolCount());
    stateMachine.setPreCoolGraceCount(hb.configGraceCount());

    configTimestampCounter.set(remoteTs);

    log.debug(
      "Applied newer config from {}: confirmCount={}, coolCount={}, preCoolGraceCount={}",
      hb.workerId(),
      hb.configConfirmCount(),
      hb.configCoolCount(),
      hb.configGraceCount()
    );

    if (startupLatch.getCount() > 0) {
      startupLatch.countDown();
    }
  }
}

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.config;
17
18	import io.github.hyshmily.hotkey.detection.HotKeyStateMachine;
19	import io.github.hyshmily.hotkey.sync.worker.WorkerHeartbeatMessage;
20	import io.github.hyshmily.hotkey.util.HotKeyThreadFactory;
21	import jakarta.annotation.PostConstruct;
22	import java.util.concurrent.CountDownLatch;
23	import java.util.concurrent.TimeUnit;
24	import java.util.concurrent.atomic.AtomicLong;
25	import lombok.RequiredArgsConstructor;
26	import lombok.extern.slf4j.Slf4j;
27	import org.springframework.amqp.core.Message;
28	import org.springframework.amqp.rabbit.annotation.RabbitListener;
29
30	/**
31	* Listens for heartbeat-based config updates from peer Workers and applies them
32	* if the received config timestamp is newer than the local one.
33	*
34	* <p>On startup, waits up to 3 seconds for the first heartbeat to arrive.
35	* If none is received, the Worker continues with the values from
36	* {@link io.github.hyshmily.hotkey.worker.config.WorkerProperties} — this can happen when all other Workers are down.
37	*/
38	/** Default constructor. */
39	@RequiredArgsConstructor
40	@Slf4j	4✔
41	public class WorkerConfigNegotiator {
42
43	/** State machine whose config (confirm/cool/grace counts) is updated from heartbeat messages. */
44	private final HotKeyStateMachine stateMachine;
45	/** Monotonically increasing counter tracking the latest config-change timestamp. */
46	private final AtomicLong configTimestampCounter;
47	/** Unique identifier for this Worker node, used in queue names and heartbeat identification. */
48	private final String nodeId;
49	private final CountDownLatch startupLatch = new CountDownLatch(1);
50
51	/**
52	* Waits for the first heartbeat from a peer Worker.
53	*
54	* <p>If a heartbeat with a valid config arrives within 3 seconds the
55	* config is applied synchronously. Otherwise, a warning is logged and
56	* the Worker proceeds with configured defaults.
57	*/
58	@PostConstruct
59	void syncOnStartup() {
60	Thread waitThread = new HotKeyThreadFactory("hotkey-config-sync-startup").newThread(() -> {	8✔
61	try {
62	boolean received = startupLatch.await(3000, TimeUnit.MILLISECONDS);	6✔
63	if (!received) {	2✔
64	log.warn("No config heartbeat received within 3s, using WorkerProperties defaults");	3✔
65	}
NEW 66	} catch (InterruptedException e) {	×
NEW 67	Thread.currentThread().interrupt();	×
68	}	1✔
69	});	1✔
70	waitThread.start();	2✔
71	}	1✔
72
73	/**
74	* Processes an incoming heartbeat message from a peer Worker and applies
75	* its state-machine config if the embedded timestamp is newer.
76	*
77	* @param msg the raw AMQP heartbeat message
78	*/
79	@RabbitListener(queues = "#{@workerConfigQueue.name}", containerFactory = "workerConfigListenerContainerFactory")
80	public void onHeartbeat(Message msg) {
81	try {
82	doOnHeartbeat(msg);	3✔
83	} catch (Exception e) {	×
84	log.warn("Uncaught exception in onHeartbeat config negotiation, discarding message to prevent requeue loop", e);	×
85	}	1✔
86	}	1✔
87
88	private void doOnHeartbeat(Message msg) {
89	WorkerHeartbeatMessage hb = WorkerHeartbeatMessage.from(msg);	3✔
90	if (hb == null) {	2✔
91	return;	1✔
92	}
93
94	if (hb.workerId().equals(nodeId)) {	6✔
95	return;	1✔
96	}
97
98	long remoteTs = hb.configTimestamp();	3✔
99	long localTs = configTimestampCounter.get();	4✔
100	if (remoteTs <= localTs) {	4✔
101	return;	1✔
102	}
103
104	stateMachine.setConfirmCount(hb.configConfirmCount());	5✔
105	stateMachine.setCoolCount(hb.configCoolCount());	5✔
106	stateMachine.setPreCoolGraceCount(hb.configGraceCount());	5✔
107
108	configTimestampCounter.set(remoteTs);	4✔
109
110	log.debug(	8✔
111	"Applied newer config from {}: confirmCount={}, coolCount={}, preCoolGraceCount={}",
112	hb.workerId(),	5✔
113	hb.configConfirmCount(),	6✔
114	hb.configCoolCount(),	6✔
115	hb.configGraceCount()	3✔
116	);
117
118	if (startupLatch.getCount() > 0) {	6✔
119	startupLatch.countDown();	3✔
120	}
121	}	1✔
122	}

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