Aias00 commented on code in PR #7727:
URL: https://github.com/apache/shenyu/pull/7727#discussion_r4303514356
##########
shenyu-admin/src/main/java/org/apache/shenyu/admin/listener/http/HttpLongPollingDataChangedListener.java:
##########
@@ -410,49 +465,159 @@ class LongPollingClient implements Runnable {
*/
private Future<?> asyncTimeoutFuture;
+ /**
+ * The Heartbeat future.
+ */
+ private Future<?> heartbeatFuture;
+
+ /**
+ * Completed flag to prevent duplicate execution or responses.
+ */
+ private final AtomicBoolean isCompleted = new AtomicBoolean(false);
+
/**
* Instantiates a new Long polling client.
*
* @param ac the ac
* @param ip the ip
* @param timeoutTime the timeout time
+ * @param namespaceId the namespaceId
*/
LongPollingClient(final AsyncContext ac, final String ip, final long
timeoutTime, final String namespaceId) {
+ this(ac, ip, StringUtils.EMPTY, StringUtils.EMPTY, timeoutTime,
namespaceId);
+ }
+
+ /**
+ * Instantiates a new Long polling client with port and bootstrap info.
+ *
+ * @param ac the ac
+ * @param ip the ip
+ * @param port the port
+ * @param bootstrapInfo the bootstrap info
+ * @param timeoutTime the timeout time
+ * @param namespaceId the namespaceId
+ */
+ LongPollingClient(final AsyncContext ac, final String ip, final String
port,
+ final String bootstrapInfo, final long timeoutTime,
final String namespaceId) {
this.asyncContext = ac;
this.ip = ip;
+ this.port = port;
+ this.bootstrapInfo = bootstrapInfo;
this.timeoutTime = timeoutTime;
this.namespaceId = namespaceId;
+ if (Objects.nonNull(ac)) {
+ try {
+ ac.addListener(this);
+ } catch (Exception ex) {
+ log.warn("failed to add async listener", ex);
+ }
+ }
}
@Override
public void run() {
try {
- BlockingQueue<LongPollingClient> namespaceClients =
clientsMap.getOrDefault(namespaceId, new ArrayBlockingQueue<>(1024));
+ BlockingQueue<LongPollingClient> namespaceClients =
clientsMap.computeIfAbsent(
+ namespaceId, k -> new ArrayBlockingQueue<>(1024));
this.asyncTimeoutFuture = scheduler.schedule(() -> {
- namespaceClients.remove(LongPollingClient.this);
List<ConfigGroupEnum> changedGroups =
compareChangedGroup((HttpServletRequest) asyncContext.getRequest());
sendResponse(changedGroups);
log.debug("LongPollingClient {} ",
GsonUtils.getInstance().toJson(changedGroups));
}, timeoutTime, TimeUnit.MILLISECONDS);
+ if (shouldReportInstance(port)) {
+ this.heartbeatFuture = scheduler.scheduleWithFixedDelay(
+ this::refreshLiveness,
+ heartbeatIntervalMs,
+ heartbeatIntervalMs,
+ TimeUnit.MILLISECONDS);
+ }
namespaceClients.add(this);
- clientsMap.put(namespaceId, namespaceClients);
} catch (Exception ex) {
log.error("add long polling client error", ex);
}
}
+ /**
+ * Refreshes the liveness of active held client.
+ */
+ void refreshLiveness() {
+ try {
+ if (isCompleted.get()) {
+ return;
+ }
+ BlockingQueue<LongPollingClient> namespaceClients =
clientsMap.get(namespaceId);
+ if (Objects.isNull(namespaceClients) ||
!namespaceClients.contains(this)) {
+ cleanUp();
+ return;
+ }
+ reportInstanceInfo(ip, port, bootstrapInfo, namespaceId);
+ log.debug("LongPollingClient refreshed liveness for ip={},
port={}, namespaceId={}", ip, port, namespaceId);
+ } catch (Exception e) {
+ log.error("refresh long polling client liveness error", e);
+ }
+ }
+
+ /**
+ * Cleans up futures and removes client from queue.
+ */
+ void cleanUp() {
+ if (Objects.nonNull(asyncTimeoutFuture)) {
+ asyncTimeoutFuture.cancel(false);
+ }
+ if (Objects.nonNull(heartbeatFuture)) {
+ heartbeatFuture.cancel(false);
+ }
+ BlockingQueue<LongPollingClient> namespaceClients =
clientsMap.get(namespaceId);
+ if (Objects.nonNull(namespaceClients)) {
+ namespaceClients.remove(this);
+ }
+ }
+
/**
* Send response.
*
* @param changedGroups the changed groups
*/
void sendResponse(final List<ConfigGroupEnum> changedGroups) {
- // cancel scheduler
- if (Objects.nonNull(asyncTimeoutFuture)) {
- asyncTimeoutFuture.cancel(false);
+ if (!isCompleted.compareAndSet(false, true)) {
Review Comment:
The old code removed the client from `clientsMap` unconditionally in the
timeout task; removal now only happens via `cleanUp()` after this CAS succeeds.
If `onError`/`onComplete` fired before `run()` added the client, `isCompleted`
is already true, the listener's `cleanUp()` removed nothing, `run()` adds the
client anyway, and this branch returns without cleaning up — the client then
stays in the queue forever. Calling `cleanUp()` here (it is idempotent) or
checking `isCompleted` at the top of `run()` closes it.
##########
shenyu-admin/src/main/java/org/apache/shenyu/admin/listener/http/HttpLongPollingDataChangedListener.java:
##########
@@ -410,49 +465,159 @@ class LongPollingClient implements Runnable {
*/
private Future<?> asyncTimeoutFuture;
+ /**
+ * The Heartbeat future.
+ */
+ private Future<?> heartbeatFuture;
+
+ /**
+ * Completed flag to prevent duplicate execution or responses.
+ */
+ private final AtomicBoolean isCompleted = new AtomicBoolean(false);
+
/**
* Instantiates a new Long polling client.
*
* @param ac the ac
* @param ip the ip
* @param timeoutTime the timeout time
+ * @param namespaceId the namespaceId
*/
LongPollingClient(final AsyncContext ac, final String ip, final long
timeoutTime, final String namespaceId) {
+ this(ac, ip, StringUtils.EMPTY, StringUtils.EMPTY, timeoutTime,
namespaceId);
+ }
+
+ /**
+ * Instantiates a new Long polling client with port and bootstrap info.
+ *
+ * @param ac the ac
+ * @param ip the ip
+ * @param port the port
+ * @param bootstrapInfo the bootstrap info
+ * @param timeoutTime the timeout time
+ * @param namespaceId the namespaceId
+ */
+ LongPollingClient(final AsyncContext ac, final String ip, final String
port,
+ final String bootstrapInfo, final long timeoutTime,
final String namespaceId) {
this.asyncContext = ac;
this.ip = ip;
+ this.port = port;
+ this.bootstrapInfo = bootstrapInfo;
this.timeoutTime = timeoutTime;
this.namespaceId = namespaceId;
+ if (Objects.nonNull(ac)) {
+ try {
+ ac.addListener(this);
+ } catch (Exception ex) {
+ log.warn("failed to add async listener", ex);
+ }
+ }
}
@Override
public void run() {
try {
- BlockingQueue<LongPollingClient> namespaceClients =
clientsMap.getOrDefault(namespaceId, new ArrayBlockingQueue<>(1024));
+ BlockingQueue<LongPollingClient> namespaceClients =
clientsMap.computeIfAbsent(
+ namespaceId, k -> new ArrayBlockingQueue<>(1024));
this.asyncTimeoutFuture = scheduler.schedule(() -> {
Review Comment:
Nit: `run()` schedules both futures and enqueues the client without checking
`isCompleted`, which is what makes the race described on `sendResponse`
possible. An early `if (isCompleted.get()) { cleanUp(); return; }` here would
also avoid scheduling a heartbeat for a client that is already gone.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
