28216525937

Committed 26 Jun 2026 04:08AM UTC coverage: 91.258% (-1.4%) from 92.707%

Build # 28216525937

Build Type

push

github

Committed by

Hyshmily

Commit Message

feat: add circuit breaker, null-value caching, and heartbeat refinements

- Introduce HotKeyCircuitBreaker with sliding-window failure detection
- Cache null/miss results with configurable TTL (null-value-ttl-seconds)
- Add per-Worker exponential backoff to heartbeat verifier (verify-max-backoff-ms)
- Support min-alive-workers config for cluster health threshold
- Migrate HotKeyEndpoint from Actuator @Endpoint to Spring MVC @RestController
- Add NumberFormatException handling in StateMachineEndpoint

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

Coverage Stats

1226 of 1419 branches covered (86.4%)

Branch coverage included in aggregate %.

181 of 229 new or added lines in 12 files covered. (79.04%)

1 existing line in 1 file now uncovered.

3503 of 3763 relevant lines covered (93.09%)

4.22 hits per line

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

92.55

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

import io.github.hyshmily.hotkey.sharding.ClusterHealthView;
import lombok.AccessLevel;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.AmqpTimeoutException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.core.RabbitTemplate;

import java.util.Set;
import java.util.concurrent.*;
import java.util.stream.Collectors;

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

/**
 * On-demand active heartbeat verifier that probes suspected-dead Workers when
 * passive heartbeat monitoring indicates a potential failure.
 *
 * <p>Under normal operation, the {@link ClusterHealthView} is updated passively
 * by incoming heartbeat messages. This verifier activates only when one or more
 * Workers have exceeded the {@code heartbeatTimeoutMs} window — it sends a
 * point-to-point PING message to each suspected Worker's dedicated verification
 * queue ({@code hotkey.verify.ping.{workerId}}) via Direct reply-to and awaits
 * a PONG response.
 *
 * <p><b>Failure escalation:</b> If cumulative failures across all suspected Workers
 * reach {@code degradeAfterFailures} and the cluster remains unhealthy, the health
 * view is marked as degraded via {@link ClusterHealthView#degraded}, triggering
 * graceful degradation behaviour (see ADR-0009). Workers that respond are restored
 * to the alive set; Workers that fail repeatedly are marked as stale and excluded
 * from majority-based health checks.
 *
 * <p><b>Thread safety:</b> The verification loop runs on a single daemon thread
 * ({@code hb-verifier}). The {@link ClusterHealthView} uses {@code ConcurrentHashMap}
 * internally, so concurrent heartbeat reception and verification are safe.
 *
 * @see ClusterHealthView
 * @see WorkerHeartbeatMessage
 */
@RequiredArgsConstructor(access = AccessLevel.PRIVATE)
@Slf4j
public class WorkerHeartbeatVerifier {


  private final RabbitTemplate rabbitTemplate;
  private final ClusterHealthView healthView;
  private final String appInstanceId;
  private final long verifyIntervalMs;
  private final long pingTimeoutMs;
  private final int degradeAfterFailures;
  private final long verifyMaxBackoffMs;
  private final ScheduledExecutorService scheduler;
  private final ConcurrentHashMap<String, Long> nextVerifyTime = new ConcurrentHashMap<>();
  private final boolean ownsScheduler;
  private ScheduledFuture<?> verifyTask;

  private static final String QUEUE_VERIFY_PING_PREFIX = "hotkey.verify.ping.";

  /**
   * Creates a verifier with an internally owned single-thread scheduler.
   * <p>
   * This constructor is convenience for standalone use and tests. The internal
   * daemon thread is named {@code hb-verifier}. The caller must invoke
   * {@link #start()} to begin periodic verification and {@link #stop()} to
   * release the scheduler resources.
   *
   * @param rabbitTemplate        the RabbitMQ template for sending PING messages
   * @param healthView            the shared cluster health view to update on PONG/failure
   * @param appInstanceId         the local application instance ID, sent as PING metadata
   * @param verifyIntervalMs      fixed delay between verification cycles (milliseconds)
   * @param pingTimeoutMs         maximum time to wait for a PONG response (milliseconds)
   * @param degradeAfterFailures  number of failed verifications before marking a Worker stale
   *                              and potentially degrading the cluster
   */
  public WorkerHeartbeatVerifier(
    RabbitTemplate rabbitTemplate,
    ClusterHealthView healthView,
    String appInstanceId,
    long verifyIntervalMs,
    long pingTimeoutMs,
    int degradeAfterFailures,
    long verifyMaxBackoffMs
  ) {
    this(rabbitTemplate, healthView, appInstanceId, verifyIntervalMs, pingTimeoutMs, degradeAfterFailures, verifyMaxBackoffMs,
      Executors.newSingleThreadScheduledExecutor(r -> {
        Thread t = new Thread(r, "hb-verifier");
        t.setDaemon(true);
        return t;
      }), true);
  }

  /**
   * Creates a verifier with a caller-supplied external scheduler.
   * <p>
   * The caller is responsible for managing the scheduler's lifecycle. When using
   * this constructor, {@link #stop()} will cancel the verification task but will
   * <em>not</em> shut down the scheduler. This allows sharing a single scheduler
   * across multiple listeners and verifiers.
   *
   * @param rabbitTemplate        the RabbitMQ template for sending PING messages
   * @param healthView            the shared cluster health view to update on PONG/failure
   * @param appInstanceId         the local application instance ID, sent as PING metadata
   * @param verifyIntervalMs      fixed delay between verification cycles (milliseconds)
   * @param pingTimeoutMs         maximum time to wait for a PONG response (milliseconds)
   * @param degradeAfterFailures  number of failed verifications before marking a Worker stale
   * @param scheduler             the external scheduler for running verification tasks;
   *                              will not be shut down by {@link #stop()}
   */
  public WorkerHeartbeatVerifier(
    RabbitTemplate rabbitTemplate,
    ClusterHealthView healthView,
    String appInstanceId,
    long verifyIntervalMs,
    long pingTimeoutMs,
    int degradeAfterFailures,
    long verifyMaxBackoffMs,
    ScheduledExecutorService scheduler
  ) {
    this(rabbitTemplate, healthView, appInstanceId, verifyIntervalMs, pingTimeoutMs, degradeAfterFailures, verifyMaxBackoffMs,
      scheduler, false);
  }

  /**
   * Starts the periodic heartbeat verification at a fixed rate.
   *
   * <p>Every {@code verifyIntervalMs} milliseconds, iterates over all Workers in
   * {@link ClusterHealthView} that have exceeded {@code heartbeatTimeoutMs} without
   * a heartbeat and sends each a PING via Direct reply-to. On PONG, the Worker's
   * health record is restored; on timeout, the failure counter is incremented.
   * <p>
   * The verification runs on a daemon background thread. The task is idempotent:
   * subsequent calls to this method after the verifier is already running are
   * silently ignored.
   *
   * @see #verifySuspectedWorkers
   * @see #stop()
   */
  public void start() {
    if (verifyTask != null) {
      return;
    }
    try {
      verifyTask = scheduler.scheduleAtFixedRate(
        this::verifySuspectedWorkers,
        verifyIntervalMs,
        verifyIntervalMs,
        TimeUnit.MILLISECONDS
      );
    } catch (Exception e) {
      log.error("Failed to start heartbeat verifier scheduler; Worker liveness verification " +
          "will not run. Application continues but stale Workers may not be detected.", e);
    }
  }

  /**
   * Gracefully stops the periodic heartbeat verification.
   *
   * <p>Cancels the scheduled verification task and, if this verifier owns its
   * scheduler (created via the 6-argument constructor), shuts down the scheduler.
   * When an external scheduler is used (7-argument constructor), the scheduler is
   * left running for the caller to manage.
   *
   * <p>After {@code stop()}, the verifier can be restarted by calling {@link #start()}
   * again.
   */
  public void stop() {
    if (verifyTask != null) {
      verifyTask.cancel(false);
    }
    if (ownsScheduler) {
      scheduler.shutdown();
    }
  }

  /**
   * Iterates over all Workers registered in {@link ClusterHealthView} that are
   * <em>not</em> currently marked alive, sends each a PING via Direct reply-to,
   * and updates the health view based on the response.
   *
   * <p>If the cluster is healthy (majority of Workers alive), the verification
   * is skipped entirely — no need for active probing. When cumulative failures
   * across all suspected Workers reach {@code degradeAfterFailures} and the cluster
   * remains unhealthy, the cluster is marked as degraded.
   *
   * <p>This method is called periodically by the scheduled task and is also
   * safe to invoke manually for testing.
   */
  public void verifySuspectedWorkers() {
    try {
      if (healthView.isClusterHealthy()) {
        return;
      }

      Set<String> suspected = healthView
        .getAllWorkerIds()
        .stream()
        .filter(id -> !healthView.getAliveWorkerIds().contains(id))
        .collect(Collectors.toSet());

      if (suspected.isEmpty()) {
        return;
      }

      log.info("Verifying suspected workers: {}", suspected);

      for (String workerId : suspected) {
        Long skipUntil = nextVerifyTime.get(workerId);
        if (skipUntil != null && System.currentTimeMillis() < skipUntil) {
          log.trace("Worker {} in backoff, skip (remaining={}ms)", workerId, skipUntil - System.currentTimeMillis());
          continue;
        }

        boolean alive = sendPingAndWaitPong(workerId);
        if (!alive) {
          healthView.markVerificationFailed(workerId);

          int attempt = healthView.getVerifyFailures(workerId);
          long backoffMs = Math.min(verifyMaxBackoffMs, verifyIntervalMs * (1L << Math.min(attempt, 30)));

          nextVerifyTime.put(workerId, System.currentTimeMillis() + backoffMs);
          log.warn("Worker {} verification failed (attempt={}, backoff={}ms)", workerId, attempt, backoffMs);
        } else {
          nextVerifyTime.remove(workerId);
          healthView.recordPong(workerId);
        }
      }

      boolean anyDegraded = suspected.stream()
        .anyMatch(id -> healthView.getVerifyFailures(id) >= degradeAfterFailures);
      if (anyDegraded && !healthView.isClusterHealthy()) {
        log.warn("Cluster degraded: some workers unreachable after verification (threshold={})", degradeAfterFailures);
        healthView.setDegraded(true);
      }
    } catch (Exception e) {
      log.error("Scheduled verifySuspectedWorkers failed", e);
    }
  }

  /**
   * Sends a point-to-point PING message to the given Worker's dedicated verification
   * queue ({@code hotkey.verify.ping.{workerId}}) and waits for a PONG response
   * via AMQP Direct reply-to ({@code amq.rabbitmq.reply-to}).
   *
   * <p>The PING carries the application instance ID in the
   * {@link io.github.hyshmily.hotkey.constants.HotKeyConstants#AMQP_HEADER_VERIFY_APP_INSTANCE} header so the Worker
   * can identify the requester. The reply timeout is set to {@code pingTimeoutMs}
   * on the RabbitTemplate before each call.
   *
   * @param workerId the Worker to ping; must not be null
   * @return {@code true} if a non-null PONG response was received within
   *         {@code pingTimeoutMs}; {@code false} if the request timed out
   *         ({@link AmqpTimeoutException}) or the queue is unreachable
   */
  public boolean sendPingAndWaitPong(String workerId) {
    MessageProperties props = new MessageProperties();
    props.setHeader(AMQP_HEADER_VERIFY_TYPE, AMQP_HEADER_VERIFY_PING);
    props.setHeader(AMQP_HEADER_VERIFY_APP_INSTANCE, appInstanceId);
    props.setReplyTo("amq.rabbitmq.reply-to");

    Message ping = new Message(new byte[0], props);

    try {
      rabbitTemplate.setReplyTimeout((int) pingTimeoutMs);
      Message pong = rabbitTemplate.sendAndReceive("", QUEUE_VERIFY_PING_PREFIX + workerId, ping);
      return pong != null;
    } catch (AmqpTimeoutException e) {
      return false;
    }
  }
}

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.worker;
17
18	import io.github.hyshmily.hotkey.sharding.ClusterHealthView;
19	import lombok.AccessLevel;
20	import lombok.RequiredArgsConstructor;
21	import lombok.extern.slf4j.Slf4j;
22	import org.springframework.amqp.AmqpTimeoutException;
23	import org.springframework.amqp.core.Message;
24	import org.springframework.amqp.core.MessageProperties;
25	import org.springframework.amqp.rabbit.core.RabbitTemplate;
26
27	import java.util.Set;
28	import java.util.concurrent.*;
29	import java.util.stream.Collectors;
30
31	import static io.github.hyshmily.hotkey.constants.HotKeyConstants.*;
32
33	/**
34	* On-demand active heartbeat verifier that probes suspected-dead Workers when
35	* passive heartbeat monitoring indicates a potential failure.
36	*
37	* <p>Under normal operation, the {@link ClusterHealthView} is updated passively
38	* by incoming heartbeat messages. This verifier activates only when one or more
39	* Workers have exceeded the {@code heartbeatTimeoutMs} window — it sends a
40	* point-to-point PING message to each suspected Worker's dedicated verification
41	* queue ({@code hotkey.verify.ping.{workerId}}) via Direct reply-to and awaits
42	* a PONG response.
43	*
44	* <p><b>Failure escalation:</b> If cumulative failures across all suspected Workers
45	* reach {@code degradeAfterFailures} and the cluster remains unhealthy, the health
46	* view is marked as degraded via {@link ClusterHealthView#degraded}, triggering
47	* graceful degradation behaviour (see ADR-0009). Workers that respond are restored
48	* to the alive set; Workers that fail repeatedly are marked as stale and excluded
49	* from majority-based health checks.
50	*
51	* <p><b>Thread safety:</b> The verification loop runs on a single daemon thread
52	* ({@code hb-verifier}). The {@link ClusterHealthView} uses {@code ConcurrentHashMap}
53	* internally, so concurrent heartbeat reception and verification are safe.
54	*
55	* @see ClusterHealthView
56	* @see WorkerHeartbeatMessage
57	*/
58	@RequiredArgsConstructor(access = AccessLevel.PRIVATE)
59	@Slf4j	4✔
60	public class WorkerHeartbeatVerifier {
61
62
63	private final RabbitTemplate rabbitTemplate;
64	private final ClusterHealthView healthView;
65	private final String appInstanceId;
66	private final long verifyIntervalMs;
67	private final long pingTimeoutMs;
68	private final int degradeAfterFailures;
69	private final long verifyMaxBackoffMs;
70	private final ScheduledExecutorService scheduler;
71	private final ConcurrentHashMap<String, Long> nextVerifyTime = new ConcurrentHashMap<>();
72	private final boolean ownsScheduler;
73	private ScheduledFuture<?> verifyTask;
74
75	private static final String QUEUE_VERIFY_PING_PREFIX = "hotkey.verify.ping.";
76
77	/**
78	* Creates a verifier with an internally owned single-thread scheduler.
79	* <p>
80	* This constructor is convenience for standalone use and tests. The internal
81	* daemon thread is named {@code hb-verifier}. The caller must invoke
82	* {@link #start()} to begin periodic verification and {@link #stop()} to
83	* release the scheduler resources.
84	*
85	* @param rabbitTemplate the RabbitMQ template for sending PING messages
86	* @param healthView the shared cluster health view to update on PONG/failure
87	* @param appInstanceId the local application instance ID, sent as PING metadata
88	* @param verifyIntervalMs fixed delay between verification cycles (milliseconds)
89	* @param pingTimeoutMs maximum time to wait for a PONG response (milliseconds)
90	* @param degradeAfterFailures number of failed verifications before marking a Worker stale
91	* and potentially degrading the cluster
92	*/
93	public WorkerHeartbeatVerifier(
94	RabbitTemplate rabbitTemplate,
95	ClusterHealthView healthView,
96	String appInstanceId,
97	long verifyIntervalMs,
98	long pingTimeoutMs,
99	int degradeAfterFailures,
100	long verifyMaxBackoffMs
101	) {
102	this(rabbitTemplate, healthView, appInstanceId, verifyIntervalMs, pingTimeoutMs, degradeAfterFailures, verifyMaxBackoffMs,	10✔
103	Executors.newSingleThreadScheduledExecutor(r -> {	2✔
104	Thread t = new Thread(r, "hb-verifier");	6✔
105	t.setDaemon(true);	3✔
106	return t;	2✔
107	}), true);
108	}	1✔
109
110	/**
111	* Creates a verifier with a caller-supplied external scheduler.
112	* <p>
113	* The caller is responsible for managing the scheduler's lifecycle. When using
114	* this constructor, {@link #stop()} will cancel the verification task but will
115	* <em>not</em> shut down the scheduler. This allows sharing a single scheduler
116	* across multiple listeners and verifiers.
117	*
118	* @param rabbitTemplate the RabbitMQ template for sending PING messages
119	* @param healthView the shared cluster health view to update on PONG/failure
120	* @param appInstanceId the local application instance ID, sent as PING metadata
121	* @param verifyIntervalMs fixed delay between verification cycles (milliseconds)
122	* @param pingTimeoutMs maximum time to wait for a PONG response (milliseconds)
123	* @param degradeAfterFailures number of failed verifications before marking a Worker stale
124	* @param scheduler the external scheduler for running verification tasks;
125	* will not be shut down by {@link #stop()}
126	*/
127	public WorkerHeartbeatVerifier(
128	RabbitTemplate rabbitTemplate,
129	ClusterHealthView healthView,
130	String appInstanceId,
131	long verifyIntervalMs,
132	long pingTimeoutMs,
133	int degradeAfterFailures,
134	long verifyMaxBackoffMs,
135	ScheduledExecutorService scheduler
136	) {
137	this(rabbitTemplate, healthView, appInstanceId, verifyIntervalMs, pingTimeoutMs, degradeAfterFailures, verifyMaxBackoffMs,	11✔
138	scheduler, false);
139	}	1✔
140
141	/**
142	* Starts the periodic heartbeat verification at a fixed rate.
143	*
144	* <p>Every {@code verifyIntervalMs} milliseconds, iterates over all Workers in
145	* {@link ClusterHealthView} that have exceeded {@code heartbeatTimeoutMs} without
146	* a heartbeat and sends each a PING via Direct reply-to. On PONG, the Worker's
147	* health record is restored; on timeout, the failure counter is incremented.
148	* <p>
149	* The verification runs on a daemon background thread. The task is idempotent:
150	* subsequent calls to this method after the verifier is already running are
151	* silently ignored.
152	*
153	* @see #verifySuspectedWorkers
154	* @see #stop()
155	*/
156	public void start() {
157	if (verifyTask != null) {	3✔
158	return;	1✔
159	}
160	try {
161	verifyTask = scheduler.scheduleAtFixedRate(	12✔
162	this::verifySuspectedWorkers,
163	verifyIntervalMs,
164	verifyIntervalMs,
165	TimeUnit.MILLISECONDS
166	);
167	} catch (Exception e) {	1✔
168	log.error("Failed to start heartbeat verifier scheduler; Worker liveness verification " +	4✔
169	"will not run. Application continues but stale Workers may not be detected.", e);
170	}	1✔
171	}	1✔
172
173	/**
174	* Gracefully stops the periodic heartbeat verification.
175	*
176	* <p>Cancels the scheduled verification task and, if this verifier owns its
177	* scheduler (created via the 6-argument constructor), shuts down the scheduler.
178	* When an external scheduler is used (7-argument constructor), the scheduler is
179	* left running for the caller to manage.
180	*
181	* <p>After {@code stop()}, the verifier can be restarted by calling {@link #start()}
182	* again.
183	*/
184	public void stop() {
185	if (verifyTask != null) {	3✔
186	verifyTask.cancel(false);	5✔
187	}
188	if (ownsScheduler) {	3!
189	scheduler.shutdown();	3✔
190	}
191	}	1✔
192
193	/**
194	* Iterates over all Workers registered in {@link ClusterHealthView} that are
195	* <em>not</em> currently marked alive, sends each a PING via Direct reply-to,
196	* and updates the health view based on the response.
197	*
198	* <p>If the cluster is healthy (majority of Workers alive), the verification
199	* is skipped entirely — no need for active probing. When cumulative failures
200	* across all suspected Workers reach {@code degradeAfterFailures} and the cluster
201	* remains unhealthy, the cluster is marked as degraded.
202	*
203	* <p>This method is called periodically by the scheduled task and is also
204	* safe to invoke manually for testing.
205	*/
206	public void verifySuspectedWorkers() {
207	try {
208	if (healthView.isClusterHealthy()) {	4✔
209	return;	1✔
210	}
211
212	Set<String> suspected = healthView	2✔
213	.getAllWorkerIds()	1✔
214	.stream()	3✔
215	.filter(id -> !healthView.getAliveWorkerIds().contains(id))	11✔
216	.collect(Collectors.toSet());	4✔
217
218	if (suspected.isEmpty()) {	3✔
219	return;	1✔
220	}
221
222	log.info("Verifying suspected workers: {}", suspected);	4✔
223
224	for (String workerId : suspected) {	10✔
225	Long skipUntil = nextVerifyTime.get(workerId);	6✔
226	if (skipUntil != null && System.currentTimeMillis() < skipUntil) {	2!
NEW 227	log.trace("Worker {} in backoff, skip (remaining={}ms)", workerId, skipUntil - System.currentTimeMillis());	×
NEW 228	continue;	×
229	}
230
231	boolean alive = sendPingAndWaitPong(workerId);	4✔
232	if (!alive) {	2✔
233	healthView.markVerificationFailed(workerId);	4✔
234
235	int attempt = healthView.getVerifyFailures(workerId);	5✔
236	long backoffMs = Math.min(verifyMaxBackoffMs, verifyIntervalMs * (1L << Math.min(attempt, 30)));	12✔
237
238	nextVerifyTime.put(workerId, System.currentTimeMillis() + backoffMs);	9✔
239	log.warn("Worker {} verification failed (attempt={}, backoff={}ms)", workerId, attempt, backoffMs);	19✔
240	} else {	1✔
241	nextVerifyTime.remove(workerId);	5✔
242	healthView.recordPong(workerId);	4✔
243	}
244	}	1✔
245
246	boolean anyDegraded = suspected.stream()	4✔
247	.anyMatch(id -> healthView.getVerifyFailures(id) >= degradeAfterFailures);	13✔
248	if (anyDegraded && !healthView.isClusterHealthy()) {	6!
249	log.warn("Cluster degraded: some workers unreachable after verification (threshold={})", degradeAfterFailures);	6✔
250	healthView.setDegraded(true);	4✔
251	}
252	} catch (Exception e) {	1✔
253	log.error("Scheduled verifySuspectedWorkers failed", e);	4✔
254	}	1✔
255	}	1✔
256
257	/**
258	* Sends a point-to-point PING message to the given Worker's dedicated verification
259	* queue ({@code hotkey.verify.ping.{workerId}}) and waits for a PONG response
260	* via AMQP Direct reply-to ({@code amq.rabbitmq.reply-to}).
261	*
262	* <p>The PING carries the application instance ID in the
263	* {@link io.github.hyshmily.hotkey.constants.HotKeyConstants#AMQP_HEADER_VERIFY_APP_INSTANCE} header so the Worker
264	* can identify the requester. The reply timeout is set to {@code pingTimeoutMs}
265	* on the RabbitTemplate before each call.
266	*
267	* @param workerId the Worker to ping; must not be null
268	* @return {@code true} if a non-null PONG response was received within
269	* {@code pingTimeoutMs}; {@code false} if the request timed out
270	* ({@link AmqpTimeoutException}) or the queue is unreachable
271	*/
272	public boolean sendPingAndWaitPong(String workerId) {
273	MessageProperties props = new MessageProperties();	4✔
274	props.setHeader(AMQP_HEADER_VERIFY_TYPE, AMQP_HEADER_VERIFY_PING);	4✔
275	props.setHeader(AMQP_HEADER_VERIFY_APP_INSTANCE, appInstanceId);	5✔
276	props.setReplyTo("amq.rabbitmq.reply-to");	3✔
277
278	Message ping = new Message(new byte[0], props);	7✔
279
280	try {
281	rabbitTemplate.setReplyTimeout((int) pingTimeoutMs);	7✔
282	Message pong = rabbitTemplate.sendAndReceive("", QUEUE_VERIFY_PING_PREFIX + workerId, ping);	8✔
283	return pong != null;	6✔
284	} catch (AmqpTimeoutException e) {	1✔
285	return false;	2✔
286	}
287	}
288	}

Hyshmily / hotkey / 28216525937

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