Skip to content

Commit ac1e795

Browse files
lusu007lusu007
authored
fix: cancel streams and coalesce runtime polling (#21)
1 parent a4cb50c commit ac1e795

12 files changed

Lines changed: 280 additions & 42 deletions

File tree

‎build.gradle.kts‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,9 @@ dependencies {
2828
// Provided by fabric-loader at runtime; needed on the compile classpath for the Mixin
2929
// annotations (loom does not auto-inject it for this loader/MC combination).
3030
implementation("net.fabricmc:sponge-mixin:0.17.3+mixin.0.8.7")
31+
32+
testImplementation("org.junit.jupiter:junit-jupiter:6.1.0")
33+
testRuntimeOnly("org.junit.platform:junit-platform-launcher")
3134
}
3235

3336
java {
@@ -42,6 +45,10 @@ tasks.withType<JavaCompile>().configureEach {
4245
options.encoding = "UTF-8"
4346
}
4447

48+
tasks.test {
49+
useJUnitPlatform()
50+
}
51+
4552
spotless {
4653
java {
4754
target("src/**/*.java")

‎src/main/java/gg/grounds/connect/api/ForgeApiClient.java‎

Lines changed: 33 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,8 @@
1616
import java.util.HashMap;
1717
import java.util.List;
1818
import java.util.Map;
19+
import java.util.function.Consumer;
20+
import java.util.stream.Stream;
1921

2022
/**
2123
* REST client for the forge platform API ({@code GET /v1/projects}, {@code GET /v1/deployments}).
@@ -457,39 +459,51 @@ public void streamSse(
457459
SseHandler handler,
458460
java.util.function.BooleanSupplier cancelled)
459461
throws Exception {
462+
streamSse(accessToken, path, handler, cancelled, ignored -> {});
463+
}
464+
465+
public void streamSse(
466+
String accessToken,
467+
String path,
468+
SseHandler handler,
469+
java.util.function.BooleanSupplier cancelled,
470+
Consumer<AutoCloseable> closeOnCancel)
471+
throws Exception {
460472
HttpRequest req =
461473
HttpRequest.newBuilder(URI.create(baseUrl + path))
462474
.timeout(Duration.ofMinutes(10))
463475
.header("Authorization", "Bearer " + accessToken)
464476
.header("Accept", "text/event-stream")
465477
.GET()
466478
.build();
467-
HttpResponse<java.util.stream.Stream<String>> res =
468-
http.send(req, HttpResponse.BodyHandlers.ofLines());
479+
HttpResponse<Stream<String>> res = http.send(req, HttpResponse.BodyHandlers.ofLines());
469480
if (res.statusCode() / 100 != 2) {
470481
throw new ForgeApiException(res.statusCode(), "SSE " + path + " -> HTTP " + res.statusCode());
471482
}
472483
String event = "message";
473484
StringBuilder data = new StringBuilder();
474-
java.util.Iterator<String> it = res.body().iterator();
475-
while (it.hasNext()) {
476-
if (cancelled.getAsBoolean()) {
477-
break;
478-
}
479-
String line = it.next();
480-
if (line.isEmpty()) {
481-
if (data.length() > 0) {
482-
handler.onEvent(event, data.toString());
485+
try (Stream<String> lines = res.body()) {
486+
closeOnCancel.accept(lines);
487+
java.util.Iterator<String> it = lines.iterator();
488+
while (it.hasNext()) {
489+
if (cancelled.getAsBoolean()) {
490+
break;
483491
}
484-
event = "message";
485-
data.setLength(0);
486-
} else if (line.startsWith("event:")) {
487-
event = line.substring(6).trim();
488-
} else if (line.startsWith("data:")) {
489-
if (data.length() > 0) {
490-
data.append('\n');
492+
String line = it.next();
493+
if (line.isEmpty()) {
494+
if (data.length() > 0) {
495+
handler.onEvent(event, data.toString());
496+
}
497+
event = "message";
498+
data.setLength(0);
499+
} else if (line.startsWith("event:")) {
500+
event = line.substring(6).trim();
501+
} else if (line.startsWith("data:")) {
502+
if (data.length() > 0) {
503+
data.append('\n');
504+
}
505+
data.append(line.substring(5).trim());
491506
}
492-
data.append(line.substring(5).trim());
493507
}
494508
}
495509
}

‎src/main/java/gg/grounds/connect/core/GroundsServices.java‎

Lines changed: 8 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -11,15 +11,16 @@
1111
/** Shared service graph for Grounds client features. */
1212
public final class GroundsServices {
1313
private final ClientTaskRunner runner = new ClientTaskRunner();
14+
private final SessionLifecycle lifecycle = new SessionLifecycle();
1415
private final AuthService auth = new AuthService(runner);
1516
private final AuthenticatedApi api = new AuthenticatedApi(auth);
1617
private final PlatformService platform = new PlatformService(runner, api);
1718
private final ProjectService projects = new ProjectService(runner, api);
18-
private final ServerService servers = new ServerService(runner, api);
19+
private final ServerService servers = new ServerService(runner, api, lifecycle);
1920
private final DeploymentService deployments = new DeploymentService(runner, api);
20-
private final PushWatcher pushes = new PushWatcher(runner, api, auth, projects);
21-
private final LogService logs = new LogService(runner, api);
22-
private final NatsService nats = new NatsService(runner, api);
21+
private final PushWatcher pushes = new PushWatcher(runner, api, auth, projects, lifecycle);
22+
private final LogService logs = new LogService(runner, api, lifecycle);
23+
private final NatsService nats = new NatsService(runner, api, lifecycle);
2324

2425
public AuthService auth() {
2526
return auth;
@@ -54,6 +55,9 @@ public NatsService nats() {
5455
}
5556

5657
public void logout() {
58+
lifecycle.reset();
59+
pushes.clearWatchedPushes();
60+
servers.clearGroundsJoin();
5761
auth.logout();
5862
projects.clearCache();
5963
}

‎src/main/java/gg/grounds/connect/core/RequestCoalescer.java‎

Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,19 @@
1+
package gg.grounds.connect.core;
2+
3+
import java.util.Set;
4+
import java.util.concurrent.ConcurrentHashMap;
5+
6+
/** Suppresses duplicate in-flight requests for the same logical resource. */
7+
public final class RequestCoalescer {
8+
private final Set<Key> inFlight = ConcurrentHashMap.newKeySet();
9+
10+
public boolean begin(String scope, String name) {
11+
return inFlight.add(new Key(scope, name));
12+
}
13+
14+
public void finish(String scope, String name) {
15+
inFlight.remove(new Key(scope, name));
16+
}
17+
18+
private record Key(String scope, String name) {}
19+
}

‎src/main/java/gg/grounds/connect/core/SessionLifecycle.java‎

Lines changed: 75 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,75 @@
1+
package gg.grounds.connect.core;
2+
3+
import java.util.Set;
4+
import java.util.concurrent.ConcurrentHashMap;
5+
import java.util.concurrent.atomic.AtomicBoolean;
6+
7+
/** Tracks account-scoped async work so logout can cancel stale callbacks and streams. */
8+
public final class SessionLifecycle {
9+
private final Set<Lease> leases = ConcurrentHashMap.newKeySet();
10+
11+
public Lease openLease() {
12+
Lease lease = new Lease();
13+
leases.add(lease);
14+
return lease;
15+
}
16+
17+
public void reset() {
18+
for (Lease lease : leases) {
19+
lease.cancel();
20+
}
21+
leases.clear();
22+
}
23+
24+
public final class Lease implements AutoCloseable {
25+
private final AtomicBoolean cancelled = new AtomicBoolean();
26+
private final Set<AutoCloseable> resources = ConcurrentHashMap.newKeySet();
27+
28+
private Lease() {}
29+
30+
public boolean isCancelled() {
31+
return cancelled.get();
32+
}
33+
34+
public void closeOnCancel(AutoCloseable resource) {
35+
if (resource == null) {
36+
return;
37+
}
38+
if (isCancelled()) {
39+
closeQuietly(resource);
40+
return;
41+
}
42+
resources.add(resource);
43+
if (isCancelled() && resources.remove(resource)) {
44+
closeQuietly(resource);
45+
}
46+
}
47+
48+
private void cancel() {
49+
if (cancelled.compareAndSet(false, true)) {
50+
closeResources();
51+
}
52+
}
53+
54+
@Override
55+
public void close() {
56+
leases.remove(this);
57+
closeResources();
58+
}
59+
60+
private void closeResources() {
61+
for (AutoCloseable resource : resources) {
62+
closeQuietly(resource);
63+
}
64+
resources.clear();
65+
}
66+
}
67+
68+
private static void closeQuietly(AutoCloseable closeable) {
69+
try {
70+
closeable.close();
71+
} catch (Exception ignored) {
72+
// Best-effort cancellation cleanup.
73+
}
74+
}
75+
}

‎src/main/java/gg/grounds/connect/deployment/PushWatcher.java‎

Lines changed: 15 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@
77
import gg.grounds.connect.auth.AuthService;
88
import gg.grounds.connect.core.AuthenticatedApi;
99
import gg.grounds.connect.core.ClientTaskRunner;
10+
import gg.grounds.connect.core.SessionLifecycle;
1011
import gg.grounds.connect.project.ProjectService;
1112
import java.util.List;
1213
import java.util.Set;
@@ -19,15 +20,21 @@ public final class PushWatcher {
1920
private final AuthenticatedApi api;
2021
private final AuthService auth;
2122
private final ProjectService projects;
23+
private final SessionLifecycle lifecycle;
2224
private final Set<String> watchedPushes = ConcurrentHashMap.newKeySet();
2325
private volatile boolean started;
2426

2527
public PushWatcher(
26-
ClientTaskRunner runner, AuthenticatedApi api, AuthService auth, ProjectService projects) {
28+
ClientTaskRunner runner,
29+
AuthenticatedApi api,
30+
AuthService auth,
31+
ProjectService projects,
32+
SessionLifecycle lifecycle) {
2733
this.runner = runner;
2834
this.api = api;
2935
this.auth = auth;
3036
this.projects = projects;
37+
this.lifecycle = lifecycle;
3138
}
3239

3340
/** Background poller: shows build/deploy toasts even when the Grounds screen is closed. */
@@ -79,10 +86,14 @@ public void watchInFlightPushes(String projectId, PushStatusSink sink) {
7986
});
8087
}
8188

89+
public void clearWatchedPushes() {
90+
watchedPushes.clear();
91+
}
92+
8293
private void streamPush(Push push, PushStatusSink sink) {
8394
runner.execute(
8495
() -> {
85-
try {
96+
try (SessionLifecycle.Lease lease = lifecycle.openLease()) {
8697
runner.onClient(() -> sink.onStatus(push.id(), push.appName(), push.status()));
8798
String token = api.withAuthRetry(t -> t);
8899
api.api()
@@ -97,7 +108,8 @@ private void streamPush(Push push, PushStatusSink sink) {
97108
}
98109
}
99110
},
100-
() -> false);
111+
lease::isCancelled,
112+
lease::closeOnCancel);
101113
} catch (Throwable t) {
102114
Constants.LOG.debug("[grounds] push stream {} ended: {}", push.id(), t.toString());
103115
} finally {

‎src/main/java/gg/grounds/connect/logs/LogService.java‎

Lines changed: 14 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -3,16 +3,19 @@
33
import com.google.gson.JsonParser;
44
import gg.grounds.connect.core.AuthenticatedApi;
55
import gg.grounds.connect.core.ClientTaskRunner;
6+
import gg.grounds.connect.core.SessionLifecycle;
67
import java.util.function.BooleanSupplier;
78

89
/** Tails Grounds SSE log endpoints. */
910
public final class LogService {
1011
private final ClientTaskRunner runner;
1112
private final AuthenticatedApi api;
13+
private final SessionLifecycle lifecycle;
1214

13-
public LogService(ClientTaskRunner runner, AuthenticatedApi api) {
15+
public LogService(ClientTaskRunner runner, AuthenticatedApi api, SessionLifecycle lifecycle) {
1416
this.runner = runner;
1517
this.api = api;
18+
this.lifecycle = lifecycle;
1619
}
1720

1821
/**
@@ -22,7 +25,8 @@ public LogService(ClientTaskRunner runner, AuthenticatedApi api) {
2225
public void stream(String path, LogSink sink, BooleanSupplier cancelled) {
2326
runner.execute(
2427
() -> {
25-
try {
28+
SessionLifecycle.Lease lease = lifecycle.openLease();
29+
try (lease) {
2630
String token = api.withAuthRetry(t -> t);
2731
api.api()
2832
.streamSse(
@@ -40,10 +44,15 @@ public void stream(String path, LogSink sink, BooleanSupplier cancelled) {
4044
runner.onClient(() -> sink.onLine(line));
4145
}
4246
},
43-
cancelled);
44-
runner.onClient(() -> sink.onLine("== stream ended =="));
47+
() -> cancelled.getAsBoolean() || lease.isCancelled(),
48+
lease::closeOnCancel);
49+
if (!cancelled.getAsBoolean() && !lease.isCancelled()) {
50+
runner.onClient(() -> sink.onLine("== stream ended =="));
51+
}
4552
} catch (Throwable t) {
46-
runner.onClient(() -> sink.onLine("== error: " + t.getMessage() + " =="));
53+
if (!cancelled.getAsBoolean() && !lease.isCancelled()) {
54+
runner.onClient(() -> sink.onLine("== error: " + t.getMessage() + " =="));
55+
}
4756
}
4857
});
4958
}

‎src/main/java/gg/grounds/connect/nats/NatsService.java‎

Lines changed: 14 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@
55
import gg.grounds.connect.core.AsyncCallback;
66
import gg.grounds.connect.core.AuthenticatedApi;
77
import gg.grounds.connect.core.ClientTaskRunner;
8+
import gg.grounds.connect.core.SessionLifecycle;
89
import java.net.URLEncoder;
910
import java.nio.charset.StandardCharsets;
1011
import java.util.List;
@@ -14,10 +15,12 @@
1415
public final class NatsService {
1516
private final ClientTaskRunner runner;
1617
private final AuthenticatedApi api;
18+
private final SessionLifecycle lifecycle;
1719

18-
public NatsService(ClientTaskRunner runner, AuthenticatedApi api) {
20+
public NatsService(ClientTaskRunner runner, AuthenticatedApi api, SessionLifecycle lifecycle) {
1921
this.runner = runner;
2022
this.api = api;
23+
this.lifecycle = lifecycle;
2124
}
2225

2326
/** {@code GET /v1/cluster/nats} — broker stats, declared events, connections, JetStream. */
@@ -42,7 +45,8 @@ public void tail(
4245
String projectId, List<String> subjects, NatsTailSink sink, BooleanSupplier cancelled) {
4346
runner.execute(
4447
() -> {
45-
try {
48+
SessionLifecycle.Lease lease = lifecycle.openLease();
49+
try (lease) {
4650
StringBuilder path =
4751
new StringBuilder("/v1/cluster/nats/tail?projectId=")
4852
.append(URLEncoder.encode(projectId, StandardCharsets.UTF_8));
@@ -75,10 +79,15 @@ public void tail(
7579
default -> {}
7680
}
7781
},
78-
cancelled);
79-
runner.onClient(() -> sink.onInfo("== tail ended =="));
82+
() -> cancelled.getAsBoolean() || lease.isCancelled(),
83+
lease::closeOnCancel);
84+
if (!cancelled.getAsBoolean() && !lease.isCancelled()) {
85+
runner.onClient(() -> sink.onInfo("== tail ended =="));
86+
}
8087
} catch (Throwable t) {
81-
runner.onClient(() -> sink.onInfo("== error: " + message(t) + " =="));
88+
if (!cancelled.getAsBoolean() && !lease.isCancelled()) {
89+
runner.onClient(() -> sink.onInfo("== error: " + message(t) + " =="));
90+
}
8291
}
8392
});
8493
}

0 commit comments

Comments
 (0)