28356107387

Committed 29 Jun 2026 07:35AM UTC coverage: 91.168% (-0.01%) from 91.178%

Build # 28356107387

Build Type

push

github

Committed by

Hyshmily

Commit Message

test : fix CI tests

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

Coverage Stats

1276 of 1461 branches covered (87.34%)

Branch coverage included in aggregate %.

10 of 10 new or added lines in 4 files covered. (100.0%)

117 existing lines in 15 files now uncovered.

3596 of 3883 relevant lines covered (92.61%)

4.24 hits per line

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

91.95

/common/src/main/java/io/github/hyshmily/hotkey/sync/local/CacheSyncPublisher.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.sync.local;

import static io.github.hyshmily.hotkey.cache.CacheKeysPolicy.invalidCacheKey;
import static io.github.hyshmily.hotkey.constants.HotKeyConstants.*;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.github.benmanes.caffeine.cache.Cache;
import com.github.benmanes.caffeine.cache.Caffeine;
import io.github.hyshmily.hotkey.sync.worker.WorkerListener;
import io.github.hyshmily.hotkey.util.version.VersionController;
import jakarta.annotation.PostConstruct;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.core.RabbitTemplate;

/**
 * Publishes cache synchronization messages (INVALIDATE / REFRESH / RULES_SYNC)
 * to all
 * peer application instances via the {@code hotkey.sync.exchange}
 * FanoutExchange.
 *
 * <p>
 * This is the outbound half of the instance-to-instance cache coherence
 * protocol.
 * The inbound half is {@link CacheSyncListener}. Together they ensure that a
 * data
 * mutation on one instance (write, evict, rule change) is propagated to all
 * other
 * instances within the same deployment.
 *
 * <p>
 * <b>Responsibility boundary:</b> This publisher handles
 * <em>application-level</em>
 * data changes only (e.g. {@code @CachePut}, {@code @CacheEvict},
 * {@code invalidateAllLocal}).
 * HOT/COOL lifecycle decisions from the Worker cluster are handled separately
 * by the
 * Worker's {@code WorkerBroadcaster} and received by {@link WorkerListener}.
 *
 * <p>
 * <b>Deduplication:</b> A {@link Caffeine} dedup cache keyed by
 * {@code "{type}:{cacheKey}"} prevents redundant broadcasts of the same
 * type+key
 * with a stale version within a configurable time window (see
 * {@link CacheSyncProperties#dedupWindowSeconds}).
 *
 * <p>
 * <b>Batch operations:</b> Large invalidations are split into chunks of at most
 * {@value #BATCH_SIZE} keys per AMQP message to stay within reasonable message
 * size limits.
 *
 * @see CacheSyncListener
 * @see SyncMessage
 */
@RequiredArgsConstructor
@Slf4j
public class CacheSyncPublisher {

  /** RabbitMQ template for publishing messages to the sync FanoutExchange. */
  private final RabbitTemplate rabbitTemplate;

  /**
   * Configuration for sync exchange name, dedup window, and other sync settings.
   */
  private final CacheSyncProperties properties;

  /**
   * Dedup cache keyed by {@code "type:cacheKey"} → dataVersion. Prevents
   * redundant
   * broadcasts of the same type+key with a stale version within the configured
   * window.
   * Initialized in {@link #init()} via {@link PostConstruct}.
   */
  private Cache<String, Long> recentBroadcasts;

  /**
   * Shared Jackson {@link ObjectMapper} for serializing batch-invalidation key
   * lists to JSON.
   */
  private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();

  /**
   * Maximum number of keys per single batch-invalidation AMQP message ({@value}).
   * Batches larger than this are split into multiple messages.
   */
  private static final int BATCH_SIZE = 1000;

  /**
   * Initializes the deduplication cache after bean construction.
   * <p>
   * Creates a {@link Caffeine} cache with:
   * <ul>
   * <li>Expire-after-write of {@link CacheSyncProperties#dedupWindowSeconds}
   * seconds</li>
   * <li>Maximum size of {@link CacheSyncProperties#dedupMaxSize} entries</li>
   * </ul>
   * Called automatically by the Spring container after all dependencies are
   * injected.
   */
  @PostConstruct
  public void init() {
    this.recentBroadcasts = Caffeine.newBuilder()
      .expireAfterWrite(properties.getDedupWindowSeconds(), TimeUnit.SECONDS)
      .maximumSize(properties.getDedupMaxSize())
      .build();
  }

  /**
   * Returns the current estimated size of the deduplication cache.
   * <p>
   * This is a Caffeine {@code estimatedSizeOfKeysCount()} — it is an approximation and
   * should
   * not be relied upon for exact accounting.
   *
   * @return estimated number of entries in the dedup cache; {@code 0} if
   *         {@link #init()} has not been called yet
   */
  public long getDedupCacheSize() {
    return recentBroadcasts == null ? 0L : recentBroadcasts.estimatedSize();
  }

  /**
   * Broadcasts a REFRESH sync message to all peer instances, triggering them
   * to reload the value for the given key from Redis.
   *
   * <p>
   * Deduplicated: if a REFRESH for the same key with a higher or equal
   * {@code dataVersion} was broadcast within the dedup window, this call is
   * silently skipped.
   *
   * @param cacheKey the affected cache key; must not be null or empty
   * @param version  the {@code dataVersion} at which the operation occurred;
   *                 see {@link VersionController#nextVersion}
   * @param degraded {@code true} if the version was obtained from the local
   *                 fallback counter (Redis unavailable)
   */
  public void broadcastRefresh(String cacheKey, long version, boolean degraded) {
    sendDeduped(cacheKey, SyncMessage.TYPE_REFRESH, version, degraded);
  }

  /**
   * Broadcasts an INVALIDATE sync message to all peer instances, requesting them
   * to remove the specified key from their local cache.
   *
   * <p>
   * Deduplicated: if an INVALIDATE for the same key with a higher or equal
   * {@code dataVersion} was broadcast within the dedup window, this call is
   * silently skipped.
   *
   * @param cacheKey the affected cache key; must not be null or empty
   * @param version  the {@code dataVersion} at which the invalidation occurred
   * @param degraded {@code true} if the version was obtained from the local
   *                 fallback counter (Redis unavailable)
   */
  public void broadcastLocalInvalidate(String cacheKey, long version, boolean degraded) {
    sendDeduped(cacheKey, SyncMessage.TYPE_INVALIDATE, version, degraded);
  }

  /**
   * Batch-invalidates multiple keys in a single AMQP message (or multiple
   * messages
   * for very large batches).
   *
   * <p>
   * The body of each message is a JSON array of key strings. The receiver
   * ({@link CacheSyncListener}) calls {@code caffeineCache.invalidateAllLocal()}
   * once per batch, which is more efficient
   * than sending individual INVALIDATE messages for each key.
   *
   * <p>
   * Large collections are automatically split into chunks of at most
   * {@value #BATCH_SIZE} keys per AMQP message to avoid exceeding broker
   * message size limits. Serialization or send failures are logged at ERROR
   * level and do not propagate to the caller.
   *
   * <p>
   * Note: this method does <em>not</em> go through the version-based dedup
   * cache. All keys are unconditionally broadcast.
   *
   * @param keys the keys to invalidate; if null or empty the call is a silent
   *             no-op
   */
  public void broadcastLocalInvalidateAll(Collection<String> keys) {
    if (keys == null || keys.isEmpty()) {
      return;
    }

    List<String> keyList = new ArrayList<>(keys);
    for (int i = 0; i < keyList.size(); i += BATCH_SIZE) {
      int end = Math.min(i + BATCH_SIZE, keyList.size());
      List<String> batch = keyList.subList(i, end);

      try {
        String json = OBJECT_MAPPER.writeValueAsString(batch);
        MessageProperties props = new MessageProperties();
        props.setHeader(AMQP_HEADER_TYPE, SyncMessage.TYPE_INVALIDATE_ALL);
        Message message = new Message(json.getBytes(StandardCharsets.UTF_8), props);

        rabbitTemplate.send(properties.getExchangeName(), "", message);
      } catch (AmqpException | JsonProcessingException e) {
        log.error("Failed to serialize batch invalidate keys", e);
      }
    }
  }

  /**
   * Broadcasts the full rule-set JSON to all peer instances for cross-instance
   * rule synchronization.
   *
   * <p>
   * Each receiver's {@link io.github.hyshmily.hotkey.rule.RuleMatcher#syncRules}
   * merges the incoming rules with the local set, guarded by the
   * {@code rulesVersion}
   * to prevent stale overwrites (newer versions win).
   *
   * <p>
   * If the payload is null or blank, the call is a silent no-op. AMQP publish
   * failures are logged at ERROR level and do not propagate.
   *
   * @param rulesJson    the serialized rule-set JSON (new format with version
   *                     wrapper);
   *                     must be a valid JSON string
   * @param rulesVersion the current rules version at the time of serialization;
   *                     receivers use this for conflict resolution
   */
  public void broadcastAllLocalRules(String rulesJson, long rulesVersion) {
    if (rulesJson == null || rulesJson.isBlank()) {
      return;
    }
    try {
      MessageProperties props = new MessageProperties();
      props.setHeader(AMQP_HEADER_TYPE, SyncMessage.TYPE_RULES_SYNC);
      props.setHeader(AMQP_HEADER_RULES_VERSION, rulesVersion);
      Message message = new Message(rulesJson.getBytes(StandardCharsets.UTF_8), props);

      rabbitTemplate.send(properties.getExchangeName(), "", message);
    } catch (AmqpException e) {
      log.error("Failed to send rules sync message", e);
    }
  }

  /**
   * Core deduplicated send implementation: only publishes if no recent broadcast
   * of the same {@code type:cacheKey} composite with a greater or equal
   * {@code dataVersion} exists in the dedup window.
   *
   * <p>
   * <b>Dedup algorithm:</b> Uses {@code Caffeine.asMap().compute()} to atomically
   * check and update the dedup cache entry:
   * <ol>
   * <li>If an existing entry has {@code oldVersion >= newVersion}, the broadcast
   * is skipped (the stale flag is set).</li>
   * <li>Otherwise, the dedup cache is updated to the new version and the message
   * is published to the sync FanoutExchange.</li>
   * </ol>
   *
   * <p>
   * The dedup cache entry expires after
   * {@link CacheSyncProperties#getDedupWindowSeconds}
   * seconds, after which a subsequent call with any version will be accepted.
   *
   * @param cacheKey the affected cache key; must not be null or empty
   * @param type     the sync message type (e.g.
   *                 {@link SyncMessage#TYPE_INVALIDATE},
   *                 {@link SyncMessage#TYPE_REFRESH})
   * @param version  the {@code dataVersion} of the operation
   * @param degraded whether the version was obtained in degraded mode; passed
   *                 through
   *                 to the message headers; also prefixes the dedup compositeKey
   *                 with "D:"
   *                 to prevent degraded broadcasts from being blocked by normal
   *                 ones
   */
  private void sendDeduped(String cacheKey, String type, long version, boolean degraded) {
    if (invalidCacheKey(cacheKey) || invalidCacheKey(type)) {
      log.debug("Invalid cacheKey or type, skip sync: cacheKey={}, type={}", cacheKey, type);
      return;
    }
    String compositeKey = degraded ? "D:" + type + ":" + cacheKey : type + ":" + cacheKey;

    AtomicBoolean skipped = new AtomicBoolean(false);
    recentBroadcasts
      .asMap()
      .compute(compositeKey, (k, oldVersion) -> {
        if (oldVersion != null && oldVersion >= version) {
          log.debug(
            "Skip sync due to recent broadcast with same or newer version: compositeKey={}, oldVersion={}, newVersion={}",
            compositeKey,
            oldVersion,
            version
          );
          skipped.set(true);
          return oldVersion;
        }
        return version;
      });

    if (!skipped.get()) {
      try {
        MessageProperties props = new MessageProperties();

        props.setHeader(AMQP_HEADER_TYPE, type);
        props.setHeader(AMQP_HEADER_VERSION, version);
        props.setHeader(AMQP_HEADER_IS_VERSION_DEGRADED, degraded);

        Message message = new Message(cacheKey.getBytes(StandardCharsets.UTF_8), props);

        rabbitTemplate.send(properties.getExchangeName(), "", message);
      } catch (AmqpException e) {
        log.error("Failed to send sync message", 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.sync.local;
17
18	import static io.github.hyshmily.hotkey.cache.CacheKeysPolicy.invalidCacheKey;
19	import static io.github.hyshmily.hotkey.constants.HotKeyConstants.*;
20
21	import com.fasterxml.jackson.core.JsonProcessingException;
22	import com.fasterxml.jackson.databind.ObjectMapper;
23	import com.github.benmanes.caffeine.cache.Cache;
24	import com.github.benmanes.caffeine.cache.Caffeine;
25	import io.github.hyshmily.hotkey.sync.worker.WorkerListener;
26	import io.github.hyshmily.hotkey.util.version.VersionController;
27	import jakarta.annotation.PostConstruct;
28	import java.nio.charset.StandardCharsets;
29	import java.util.ArrayList;
30	import java.util.Collection;
31	import java.util.List;
32	import java.util.concurrent.TimeUnit;
33	import java.util.concurrent.atomic.AtomicBoolean;
34	import lombok.RequiredArgsConstructor;
35	import lombok.extern.slf4j.Slf4j;
36	import org.springframework.amqp.AmqpException;
37	import org.springframework.amqp.core.Message;
38	import org.springframework.amqp.core.MessageProperties;
39	import org.springframework.amqp.rabbit.core.RabbitTemplate;
40
41	/**
42	* Publishes cache synchronization messages (INVALIDATE / REFRESH / RULES_SYNC)
43	* to all
44	* peer application instances via the {@code hotkey.sync.exchange}
45	* FanoutExchange.
46	*
47	* <p>
48	* This is the outbound half of the instance-to-instance cache coherence
49	* protocol.
50	* The inbound half is {@link CacheSyncListener}. Together they ensure that a
51	* data
52	* mutation on one instance (write, evict, rule change) is propagated to all
53	* other
54	* instances within the same deployment.
55	*
56	* <p>
57	* <b>Responsibility boundary:</b> This publisher handles
58	* <em>application-level</em>
59	* data changes only (e.g. {@code @CachePut}, {@code @CacheEvict},
60	* {@code invalidateAllLocal}).
61	* HOT/COOL lifecycle decisions from the Worker cluster are handled separately
62	* by the
63	* Worker's {@code WorkerBroadcaster} and received by {@link WorkerListener}.
64	*
65	* <p>
66	* <b>Deduplication:</b> A {@link Caffeine} dedup cache keyed by
67	* {@code "{type}:{cacheKey}"} prevents redundant broadcasts of the same
68	* type+key
69	* with a stale version within a configurable time window (see
70	* {@link CacheSyncProperties#dedupWindowSeconds}).
71	*
72	* <p>
73	* <b>Batch operations:</b> Large invalidations are split into chunks of at most
74	* {@value #BATCH_SIZE} keys per AMQP message to stay within reasonable message
75	* size limits.
76	*
77	* @see CacheSyncListener
78	* @see SyncMessage
79	*/
80	@RequiredArgsConstructor
81	@Slf4j	3✔
82	public class CacheSyncPublisher {
83
84	/** RabbitMQ template for publishing messages to the sync FanoutExchange. */
85	private final RabbitTemplate rabbitTemplate;
86
87	/**
88	* Configuration for sync exchange name, dedup window, and other sync settings.
89	*/
90	private final CacheSyncProperties properties;
91
92	/**
93	* Dedup cache keyed by {@code "type:cacheKey"} → dataVersion. Prevents
94	* redundant
95	* broadcasts of the same type+key with a stale version within the configured
96	* window.
97	* Initialized in {@link #init()} via {@link PostConstruct}.
98	*/
99	private Cache<String, Long> recentBroadcasts;
100
101	/**
102	* Shared Jackson {@link ObjectMapper} for serializing batch-invalidation key
103	* lists to JSON.
104	*/
105	private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();	5✔
106
107	/**
108	* Maximum number of keys per single batch-invalidation AMQP message ({@value}).
109	* Batches larger than this are split into multiple messages.
110	*/
111	private static final int BATCH_SIZE = 1000;
112
113	/**
114	* Initializes the deduplication cache after bean construction.
115	* <p>
116	* Creates a {@link Caffeine} cache with:
117	* <ul>
118	* <li>Expire-after-write of {@link CacheSyncProperties#dedupWindowSeconds}
119	* seconds</li>
120	* <li>Maximum size of {@link CacheSyncProperties#dedupMaxSize} entries</li>
121	* </ul>
122	* Called automatically by the Spring container after all dependencies are
123	* injected.
124	*/
125	@PostConstruct
126	public void init() {
127	this.recentBroadcasts = Caffeine.newBuilder()	4✔
128	.expireAfterWrite(properties.getDedupWindowSeconds(), TimeUnit.SECONDS)	6✔
129	.maximumSize(properties.getDedupMaxSize())	3✔
130	.build();	2✔
131	}	1✔
132
133	/**
134	* Returns the current estimated size of the deduplication cache.
135	* <p>
136	* This is a Caffeine {@code estimatedSizeOfKeysCount()} — it is an approximation and
137	* should
138	* not be relied upon for exact accounting.
139	*
140	* @return estimated number of entries in the dedup cache; {@code 0} if
141	* {@link #init()} has not been called yet
142	*/
143	public long getDedupCacheSize() {
144	return recentBroadcasts == null ? 0L : recentBroadcasts.estimatedSize();	9✔
145	}
146
147	/**
148	* Broadcasts a REFRESH sync message to all peer instances, triggering them
149	* to reload the value for the given key from Redis.
150	*
151	* <p>
152	* Deduplicated: if a REFRESH for the same key with a higher or equal
153	* {@code dataVersion} was broadcast within the dedup window, this call is
154	* silently skipped.
155	*
156	* @param cacheKey the affected cache key; must not be null or empty
157	* @param version the {@code dataVersion} at which the operation occurred;
158	* see {@link VersionController#nextVersion}
159	* @param degraded {@code true} if the version was obtained from the local
160	* fallback counter (Redis unavailable)
161	*/
162	public void broadcastRefresh(String cacheKey, long version, boolean degraded) {
163	sendDeduped(cacheKey, SyncMessage.TYPE_REFRESH, version, degraded);	6✔
164	}	1✔
165
166	/**
167	* Broadcasts an INVALIDATE sync message to all peer instances, requesting them
168	* to remove the specified key from their local cache.
169	*
170	* <p>
171	* Deduplicated: if an INVALIDATE for the same key with a higher or equal
172	* {@code dataVersion} was broadcast within the dedup window, this call is
173	* silently skipped.
174	*
175	* @param cacheKey the affected cache key; must not be null or empty
176	* @param version the {@code dataVersion} at which the invalidation occurred
177	* @param degraded {@code true} if the version was obtained from the local
178	* fallback counter (Redis unavailable)
179	*/
180	public void broadcastLocalInvalidate(String cacheKey, long version, boolean degraded) {
181	sendDeduped(cacheKey, SyncMessage.TYPE_INVALIDATE, version, degraded);	6✔
182	}	1✔
183
184	/**
185	* Batch-invalidates multiple keys in a single AMQP message (or multiple
186	* messages
187	* for very large batches).
188	*
189	* <p>
190	* The body of each message is a JSON array of key strings. The receiver
191	* ({@link CacheSyncListener}) calls {@code caffeineCache.invalidateAllLocal()}
192	* once per batch, which is more efficient
193	* than sending individual INVALIDATE messages for each key.
194	*
195	* <p>
196	* Large collections are automatically split into chunks of at most
197	* {@value #BATCH_SIZE} keys per AMQP message to avoid exceeding broker
198	* message size limits. Serialization or send failures are logged at ERROR
199	* level and do not propagate to the caller.
200	*
201	* <p>
202	* Note: this method does <em>not</em> go through the version-based dedup
203	* cache. All keys are unconditionally broadcast.
204	*
205	* @param keys the keys to invalidate; if null or empty the call is a silent
206	* no-op
207	*/
208	public void broadcastLocalInvalidateAll(Collection<String> keys) {
209	if (keys == null \|\| keys.isEmpty()) {	5✔
210	return;	1✔
211	}
212
213	List<String> keyList = new ArrayList<>(keys);	5✔
214	for (int i = 0; i < keyList.size(); i += BATCH_SIZE) {	8✔
215	int end = Math.min(i + BATCH_SIZE, keyList.size());	7✔
216	List<String> batch = keyList.subList(i, end);	5✔
217
218	try {
219	String json = OBJECT_MAPPER.writeValueAsString(batch);	4✔
220	MessageProperties props = new MessageProperties();	4✔
221	props.setHeader(AMQP_HEADER_TYPE, SyncMessage.TYPE_INVALIDATE_ALL);	4✔
222	Message message = new Message(json.getBytes(StandardCharsets.UTF_8), props);	8✔
223
224	rabbitTemplate.send(properties.getExchangeName(), "", message);	8✔
225	} catch (AmqpException \| JsonProcessingException e) {	×
226	log.error("Failed to serialize batch invalidate keys", e);	×
227	}	1✔
228	}
229	}	1✔
230
231	/**
232	* Broadcasts the full rule-set JSON to all peer instances for cross-instance
233	* rule synchronization.
234	*
235	* <p>
236	* Each receiver's {@link io.github.hyshmily.hotkey.rule.RuleMatcher#syncRules}
237	* merges the incoming rules with the local set, guarded by the
238	* {@code rulesVersion}
239	* to prevent stale overwrites (newer versions win).
240	*
241	* <p>
242	* If the payload is null or blank, the call is a silent no-op. AMQP publish
243	* failures are logged at ERROR level and do not propagate.
244	*
245	* @param rulesJson the serialized rule-set JSON (new format with version
246	* wrapper);
247	* must be a valid JSON string
248	* @param rulesVersion the current rules version at the time of serialization;
249	* receivers use this for conflict resolution
250	*/
251	public void broadcastAllLocalRules(String rulesJson, long rulesVersion) {
252	if (rulesJson == null \|\| rulesJson.isBlank()) {	5✔
253	return;	1✔
254	}
255	try {
256	MessageProperties props = new MessageProperties();	4✔
257	props.setHeader(AMQP_HEADER_TYPE, SyncMessage.TYPE_RULES_SYNC);	4✔
258	props.setHeader(AMQP_HEADER_RULES_VERSION, rulesVersion);	5✔
259	Message message = new Message(rulesJson.getBytes(StandardCharsets.UTF_8), props);	8✔
260
261	rabbitTemplate.send(properties.getExchangeName(), "", message);	8✔
262	} catch (AmqpException e) {	×
263	log.error("Failed to send rules sync message", e);	×
264	}	1✔
265	}	1✔
266
267	/**
268	* Core deduplicated send implementation: only publishes if no recent broadcast
269	* of the same {@code type:cacheKey} composite with a greater or equal
270	* {@code dataVersion} exists in the dedup window.
271	*
272	* <p>
273	* <b>Dedup algorithm:</b> Uses {@code Caffeine.asMap().compute()} to atomically
274	* check and update the dedup cache entry:
275	* <ol>
276	* <li>If an existing entry has {@code oldVersion >= newVersion}, the broadcast
277	* is skipped (the stale flag is set).</li>
278	* <li>Otherwise, the dedup cache is updated to the new version and the message
279	* is published to the sync FanoutExchange.</li>
280	* </ol>
281	*
282	* <p>
283	* The dedup cache entry expires after
284	* {@link CacheSyncProperties#getDedupWindowSeconds}
285	* seconds, after which a subsequent call with any version will be accepted.
286	*
287	* @param cacheKey the affected cache key; must not be null or empty
288	* @param type the sync message type (e.g.
289	* {@link SyncMessage#TYPE_INVALIDATE},
290	* {@link SyncMessage#TYPE_REFRESH})
291	* @param version the {@code dataVersion} of the operation
292	* @param degraded whether the version was obtained in degraded mode; passed
293	* through
294	* to the message headers; also prefixes the dedup compositeKey
295	* with "D:"
296	* to prevent degraded broadcasts from being blocked by normal
297	* ones
298	*/
299	private void sendDeduped(String cacheKey, String type, long version, boolean degraded) {
300	if (invalidCacheKey(cacheKey) \|\| invalidCacheKey(type)) {	6!
301	log.debug("Invalid cacheKey or type, skip sync: cacheKey={}, type={}", cacheKey, type);	5✔
302	return;	1✔
303	}
304	String compositeKey = degraded ? "D:" + type + ":" + cacheKey : type + ":" + cacheKey;	10✔
305
306	AtomicBoolean skipped = new AtomicBoolean(false);	5✔
307	recentBroadcasts	2✔
308	.asMap()	6✔
309	.compute(compositeKey, (k, oldVersion) -> {	2✔
310	if (oldVersion != null && oldVersion >= version) {	7✔
311	log.debug(	16✔
312	"Skip sync due to recent broadcast with same or newer version: compositeKey={}, oldVersion={}, newVersion={}",
313	compositeKey,
314	oldVersion,
315	version	2✔
316	);
317	skipped.set(true);	3✔
318	return oldVersion;	2✔
319	}
320	return version;	3✔
321	});
322
323	if (!skipped.get()) {	3✔
324	try {
325	MessageProperties props = new MessageProperties();	4✔
326
327	props.setHeader(AMQP_HEADER_TYPE, type);	4✔
328	props.setHeader(AMQP_HEADER_VERSION, version);	5✔
329	props.setHeader(AMQP_HEADER_IS_VERSION_DEGRADED, degraded);	5✔
330
331	Message message = new Message(cacheKey.getBytes(StandardCharsets.UTF_8), props);	8✔
332
333	rabbitTemplate.send(properties.getExchangeName(), "", message);	8✔
334	} catch (AmqpException e) {	×
UNCOV 335	log.error("Failed to send sync message", e);	×
336	}	1✔
337	}
338	}	1✔
339	}

Hyshmily / hotkey / 28356107387

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