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

87.23

/common/src/main/java/io/github/hyshmily/hotkey/cache/cachesupport/BroadcastBuffer.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.cache.cachesupport;

import io.github.hyshmily.hotkey.Internal;
import io.github.hyshmily.hotkey.sync.local.CacheSyncPublisher;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import lombok.extern.slf4j.Slf4j;

/**
 * Lossy deferred broadcast buffer for cache-sync messages.
 *
 * <p>Records the latest version for each key; on {@link #flush()} sends the
 * most recent entry per key via the {@link CacheSyncPublisher}. Only the
 * last-writer-wins version is retained per key, so intermediate versions
 * may be lost — this is acceptable because the next flush will broadcast
 * the latest known state.
 *
 * <p>Uses a lazy delayed flush strategy: the first {@link #record} after a
 * quiet period schedules a one-shot flush after a configurable delay
 * (default 500ms). Subsequent records within that window reset the timer,
 * combining multiple writes into a single broadcast cycle.
 *
 * <p>The internal {@code pending} map is lazily initialized on first
 * {@code record()} call to avoid allocating maps that are never used.
 */
@Slf4j
@Internal
public class BroadcastBuffer {

  private static final long DEFAULT_FLUSH_DELAY_MS = 500;

  @SuppressWarnings("java:S3077")
  private volatile ConcurrentHashMap<String, VersionInfo> pending;

  private final ScheduledExecutorService scheduler;

  @SuppressWarnings("OptionalUsedAsFieldOrParameterType")
  private final Optional<CacheSyncPublisher> publisher;

  private final long flushDelayMs;
  private ScheduledFuture<?> scheduledFlush;
  private final Object scheduleLock = new Object();

  /**
   * Creates a BroadcastBuffer with the default flush delay of 500ms.
   *
   * @param scheduler the shared scheduler ({@code hotKeyScheduler})
   * @param publisher the optional sync publisher
   */
  public BroadcastBuffer(
    ScheduledExecutorService scheduler,
    @SuppressWarnings("OptionalUsedAsFieldOrParameterType") Optional<CacheSyncPublisher> publisher
  ) {
    this(scheduler, publisher, DEFAULT_FLUSH_DELAY_MS);
  }

  /**
   * Creates a BroadcastBuffer with a custom flush delay.
   *
   * @param scheduler    the shared scheduler
   * @param publisher    the optional sync publisher
   * @param flushDelayMs delay in milliseconds before pending records are flushed
   */
  public BroadcastBuffer(
    ScheduledExecutorService scheduler,
    @SuppressWarnings("OptionalUsedAsFieldOrParameterType") Optional<CacheSyncPublisher> publisher,
    long flushDelayMs
  ) {
    this.scheduler = scheduler;
    this.publisher = publisher;
    this.flushDelayMs = flushDelayMs;
  }

  /**
   * Record a version update for the given key.  If the key already has a
   * pending entry, it is unconditionally replaced (last-writer-wins).
   * Resets the deferred flush timer on each call.
   *
   * @param key      the cache key
   * @param version  the data version to broadcast
   * @param degraded whether the version is degraded
   */
  @SuppressWarnings("java:S6213")
  public void record(String key, long version, boolean degraded) {
    if (pending == null) {
      pending = new ConcurrentHashMap<>();
    }
    pending.merge(key, new VersionInfo(version, degraded), (old, cur) -> cur);
    rescheduleFlush();
  }

  /**
   * Immediately flush all pending entries to the sync publisher.
   * Safe to call concurrently — swaps the internal map before iterating,
   * so concurrent {@link #record} calls see a fresh map.
   */
  public void flush() {
    publisher.ifPresent(pub -> {
      ConcurrentHashMap<String, VersionInfo> old = pending;
      if (old == null || old.isEmpty()) return;

      pending = new ConcurrentHashMap<>();
      old.forEach((key, vi) -> {
        try {
          pub.broadcastRefresh(key, vi.version, vi.degraded);
        } catch (Exception e) {
          log.warn("Failed to broadcast refresh for key {}", key, e);
        }
      });
    });
    cancelScheduledFlush();
  }

  private void rescheduleFlush() {
    synchronized (scheduleLock) {
      cancelScheduledFlush();
      scheduledFlush = scheduler.schedule(this::flush, flushDelayMs, TimeUnit.MILLISECONDS);
    }
  }

  private void cancelScheduledFlush() {
    if (scheduledFlush != null && !scheduledFlush.isDone()) {
      scheduledFlush.cancel(false);
      scheduledFlush = null;
    }
  }

  record VersionInfo(long version, boolean degraded) {}
}

1	/*
2	* Copyright 2026 Hyshmily. All Rights Reserved.
3	* Licensed under the Apache License, Version 2.0 (the "License");
4	* you may not use this file except in compliance with the License.
5	* You may obtain a copy of the License at
6	*
7	* http://www.apache.org/licenses/LICENSE-2.0
8	*
9	* Unless required by applicable law or agreed to in writing, software
10	* distributed under the License is distributed on an "AS IS" BASIS,
11	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12	* See the License for the specific language governing permissions and
13	* limitations under the License.
14	*/
15	package io.github.hyshmily.hotkey.cache.cachesupport;
16
17	import io.github.hyshmily.hotkey.Internal;
18	import io.github.hyshmily.hotkey.sync.local.CacheSyncPublisher;
19	import java.util.Optional;
20	import java.util.concurrent.ConcurrentHashMap;
21	import java.util.concurrent.ScheduledExecutorService;
22	import java.util.concurrent.ScheduledFuture;
23	import java.util.concurrent.TimeUnit;
24	import lombok.extern.slf4j.Slf4j;
25
26	/**
27	* Lossy deferred broadcast buffer for cache-sync messages.
28	*
29	* <p>Records the latest version for each key; on {@link #flush()} sends the
30	* most recent entry per key via the {@link CacheSyncPublisher}. Only the
31	* last-writer-wins version is retained per key, so intermediate versions
32	* may be lost — this is acceptable because the next flush will broadcast
33	* the latest known state.
34	*
35	* <p>Uses a lazy delayed flush strategy: the first {@link #record} after a
36	* quiet period schedules a one-shot flush after a configurable delay
37	* (default 500ms). Subsequent records within that window reset the timer,
38	* combining multiple writes into a single broadcast cycle.
39	*
40	* <p>The internal {@code pending} map is lazily initialized on first
41	* {@code record()} call to avoid allocating maps that are never used.
42	*/
43	@Slf4j	4✔
44	@Internal
45	public class BroadcastBuffer {
46
47	private static final long DEFAULT_FLUSH_DELAY_MS = 500;
48
49	@SuppressWarnings("java:S3077")
50	private volatile ConcurrentHashMap<String, VersionInfo> pending;
51
52	private final ScheduledExecutorService scheduler;
53
54	@SuppressWarnings("OptionalUsedAsFieldOrParameterType")
55	private final Optional<CacheSyncPublisher> publisher;
56
57	private final long flushDelayMs;
58	private ScheduledFuture<?> scheduledFlush;
59	private final Object scheduleLock = new Object();	5✔
60
61	/**
62	* Creates a BroadcastBuffer with the default flush delay of 500ms.
63	*
64	* @param scheduler the shared scheduler ({@code hotKeyScheduler})
65	* @param publisher the optional sync publisher
66	*/
67	public BroadcastBuffer(
68	ScheduledExecutorService scheduler,
69	@SuppressWarnings("OptionalUsedAsFieldOrParameterType") Optional<CacheSyncPublisher> publisher
70	) {
71	this(scheduler, publisher, DEFAULT_FLUSH_DELAY_MS);	5✔
72	}	1✔
73
74	/**
75	* Creates a BroadcastBuffer with a custom flush delay.
76	*
77	* @param scheduler the shared scheduler
78	* @param publisher the optional sync publisher
79	* @param flushDelayMs delay in milliseconds before pending records are flushed
80	*/
81	public BroadcastBuffer(
82	ScheduledExecutorService scheduler,
83	@SuppressWarnings("OptionalUsedAsFieldOrParameterType") Optional<CacheSyncPublisher> publisher,
84	long flushDelayMs
85	) {	2✔
86	this.scheduler = scheduler;	3✔
87	this.publisher = publisher;	3✔
88	this.flushDelayMs = flushDelayMs;	3✔
89	}	1✔
90
91	/**
92	* Record a version update for the given key. If the key already has a
93	* pending entry, it is unconditionally replaced (last-writer-wins).
94	* Resets the deferred flush timer on each call.
95	*
96	* @param key the cache key
97	* @param version the data version to broadcast
98	* @param degraded whether the version is degraded
99	*/
100	@SuppressWarnings("java:S6213")
101	public void record(String key, long version, boolean degraded) {
102	if (pending == null) {	3!
103	pending = new ConcurrentHashMap<>();	5✔
104	}
105	pending.merge(key, new VersionInfo(version, degraded), (old, cur) -> cur);	11✔
106	rescheduleFlush();	2✔
107	}	1✔
108
109	/**
110	* Immediately flush all pending entries to the sync publisher.
111	* Safe to call concurrently — swaps the internal map before iterating,
112	* so concurrent {@link #record} calls see a fresh map.
113	*/
114	public void flush() {
115	publisher.ifPresent(pub -> {	5✔
116	ConcurrentHashMap<String, VersionInfo> old = pending;	3✔
117	if (old == null \|\| old.isEmpty()) return;	5!
118
119	pending = new ConcurrentHashMap<>();	5✔
120	old.forEach((key, vi) -> {	4✔
121	try {
122	pub.broadcastRefresh(key, vi.version, vi.degraded);	7✔
NEW 123	} catch (Exception e) {	×
NEW 124	log.warn("Failed to broadcast refresh for key {}", key, e);	×
125	}	1✔
126	});	1✔
127	});	1✔
128	cancelScheduledFlush();	2✔
129	}	1✔
130
131	private void rescheduleFlush() {
132	synchronized (scheduleLock) {	5✔
133	cancelScheduledFlush();	2✔
134	scheduledFlush = scheduler.schedule(this::flush, flushDelayMs, TimeUnit.MILLISECONDS);	10✔
135	}	3✔
136	}	1✔
137
138	private void cancelScheduledFlush() {
139	if (scheduledFlush != null && !scheduledFlush.isDone()) {	7!
140	scheduledFlush.cancel(false);	5✔
141	scheduledFlush = null;	3✔
142	}
143	}	1✔
144
145	record VersionInfo(long version, boolean degraded) {}	9✔
146	}

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