|
|
|
@@ -0,0 +1,194 @@
|
|
|
|
|
package com.alttd.chat.web;
|
|
|
|
|
|
|
|
|
|
import com.alttd.chat.config.Config;
|
|
|
|
|
import com.alttd.chat.util.ALogger;
|
|
|
|
|
import com.fasterxml.jackson.databind.ObjectMapper;
|
|
|
|
|
|
|
|
|
|
import java.io.BufferedReader;
|
|
|
|
|
import java.io.InputStream;
|
|
|
|
|
import java.io.InputStreamReader;
|
|
|
|
|
import java.net.URI;
|
|
|
|
|
import java.net.http.HttpClient;
|
|
|
|
|
import java.net.http.HttpRequest;
|
|
|
|
|
import java.net.http.HttpResponse;
|
|
|
|
|
import java.nio.charset.StandardCharsets;
|
|
|
|
|
import java.time.Duration;
|
|
|
|
|
import java.util.Map;
|
|
|
|
|
import java.util.concurrent.ConcurrentHashMap;
|
|
|
|
|
import java.util.logging.Level;
|
|
|
|
|
import java.util.logging.Logger;
|
|
|
|
|
|
|
|
|
|
/**
|
|
|
|
|
* Client for subscribing to: GET /api/chat/send/subscribe/{server} served as text/event-stream (SseEmitter) by
|
|
|
|
|
* CommandsToServerController.
|
|
|
|
|
* <p>
|
|
|
|
|
* Handlers for individual SSE event types can be registered up front via {@link #register(String, WebHandler)}. Each
|
|
|
|
|
* handler says "when an event named X arrives, deserialize its JSON payload as type T and pass it to this handler."
|
|
|
|
|
* Unregistered event names are logged and skipped, not thrown.
|
|
|
|
|
* <p>
|
|
|
|
|
* If the connection fails or drops for any reason, the client logs it and retries once a minute, indefinitely, until
|
|
|
|
|
* stop() is called or the thread is interrupted.
|
|
|
|
|
* <p>
|
|
|
|
|
* Server-side auth: the controller checks @AuthenticationPrincipal Token token, then token.getKey().equals(validToken).
|
|
|
|
|
* That principal is populated from the "Authorization: Bearer <token>" header. validToken is currently hardcoded
|
|
|
|
|
* server-side; swap TOKEN below for a config value once that TODO is resolved.
|
|
|
|
|
*/
|
|
|
|
|
public class SseSubscribeClient implements Runnable {
|
|
|
|
|
|
|
|
|
|
private static final Logger log = Logger.getLogger(SseSubscribeClient.class.getName());
|
|
|
|
|
private static final Duration RETRY_DELAY = Duration.ofMinutes(1);
|
|
|
|
|
|
|
|
|
|
private final String baseUrl;
|
|
|
|
|
private final String server;
|
|
|
|
|
private final HttpClient httpClient;
|
|
|
|
|
private final ObjectMapper objectMapper = new ObjectMapper();
|
|
|
|
|
private final Map<String, WebHandler<?>> handlers = new ConcurrentHashMap<>();
|
|
|
|
|
|
|
|
|
|
private volatile boolean running = false;
|
|
|
|
|
|
|
|
|
|
public SseSubscribeClient(String baseUrl, String server, String token) {
|
|
|
|
|
this.baseUrl = baseUrl;
|
|
|
|
|
this.server = server;
|
|
|
|
|
this.httpClient = HttpClient.newBuilder()
|
|
|
|
|
.connectTimeout(Duration.ofSeconds(10))
|
|
|
|
|
.build();
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
/**
|
|
|
|
|
* Register a handler for a given SSE event name. When an event with this name arrives, its data payload is
|
|
|
|
|
* deserialized using the handler's type and passed to the handler.
|
|
|
|
|
*/
|
|
|
|
|
public <T> void register(String eventName, WebHandler<T> handler) {
|
|
|
|
|
handlers.put(eventName, handler);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
/**
|
|
|
|
|
* Runs the subscribe/retry loop on the calling thread. Blocks until stop() is called or the thread is interrupted.
|
|
|
|
|
*/
|
|
|
|
|
@Override
|
|
|
|
|
public void run() {
|
|
|
|
|
running = true;
|
|
|
|
|
while (running && !Thread.currentThread().isInterrupted()) {
|
|
|
|
|
String token = Config.CHAT_WEB_TOKEN;
|
|
|
|
|
if (token.equals("invalid-token")) {
|
|
|
|
|
log.warning("Invalid token provided for SSE connection as '" + server + "'");
|
|
|
|
|
sleepBeforeRetry();
|
|
|
|
|
continue;
|
|
|
|
|
}
|
|
|
|
|
try {
|
|
|
|
|
connectAndListen(token);
|
|
|
|
|
log.warning("SSE stream as '" + server + "' ended; reconnecting in " + RETRY_DELAY.toMinutes() + " minute(s)");
|
|
|
|
|
} catch (Exception e) {
|
|
|
|
|
log.log(Level.WARNING, "SSE connection to '" + server + "' failed; retrying in "
|
|
|
|
|
+ RETRY_DELAY.toMinutes() + " minute(s)", e
|
|
|
|
|
);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
if (running) {
|
|
|
|
|
sleepBeforeRetry();
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
public void stop() {
|
|
|
|
|
running = false;
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
private void connectAndListen(String token) throws Exception {
|
|
|
|
|
HttpResponse<InputStream> response = openConnection(token);
|
|
|
|
|
checkStatus(response);
|
|
|
|
|
readEvents(response.body());
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
private HttpResponse<InputStream> openConnection(String token) throws Exception {
|
|
|
|
|
HttpRequest request = HttpRequest.newBuilder()
|
|
|
|
|
.uri(URI.create(baseUrl + "/api/chat/server/send/subscribe/" + server))
|
|
|
|
|
.header("X-Altitude-Token", token)
|
|
|
|
|
.header("Accept", "text/event-stream")
|
|
|
|
|
.GET()
|
|
|
|
|
.build();
|
|
|
|
|
|
|
|
|
|
return httpClient.send(request, HttpResponse.BodyHandlers.ofInputStream());
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
private void checkStatus(HttpResponse<InputStream> response) {
|
|
|
|
|
int status = response.statusCode();
|
|
|
|
|
if (status == 401) {
|
|
|
|
|
throw new RuntimeException("Unauthorized - missing/invalid token");
|
|
|
|
|
}
|
|
|
|
|
if (status == 403) {
|
|
|
|
|
throw new RuntimeException("Forbidden - token did not match server's validToken");
|
|
|
|
|
}
|
|
|
|
|
if (status != 200) {
|
|
|
|
|
throw new RuntimeException("Unexpected status: " + status);
|
|
|
|
|
}
|
|
|
|
|
ALogger.info("SSE connection as '" + server + "' established");
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
/**
|
|
|
|
|
* Reads the SSE stream line by line, accumulating one event at a time, and dispatches each complete event as it's
|
|
|
|
|
* parsed.
|
|
|
|
|
*/
|
|
|
|
|
private void readEvents(InputStream body) throws Exception {
|
|
|
|
|
try (BufferedReader reader = new BufferedReader(new InputStreamReader(body, StandardCharsets.UTF_8))) {
|
|
|
|
|
String eventName = null;
|
|
|
|
|
StringBuilder dataBuffer = new StringBuilder();
|
|
|
|
|
String line;
|
|
|
|
|
|
|
|
|
|
while ((line = reader.readLine()) != null) {
|
|
|
|
|
if (line.isEmpty()) {
|
|
|
|
|
if (!dataBuffer.isEmpty()) {
|
|
|
|
|
dispatch(eventName, dataBuffer.toString());
|
|
|
|
|
}
|
|
|
|
|
eventName = null;
|
|
|
|
|
dataBuffer.setLength(0);
|
|
|
|
|
continue;
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
if (line.startsWith("event:")) {
|
|
|
|
|
eventName = line.substring(6).trim();
|
|
|
|
|
} else if (line.startsWith("data:")) {
|
|
|
|
|
if (!dataBuffer.isEmpty()) {
|
|
|
|
|
dataBuffer.append('\n');
|
|
|
|
|
}
|
|
|
|
|
dataBuffer.append(line.substring(5).trim());
|
|
|
|
|
}
|
|
|
|
|
// lines starting with ":" are comments/heartbeats - ignored
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
private void dispatch(String eventName, String data) {
|
|
|
|
|
if (eventName == null) {
|
|
|
|
|
log.warning("Received event with no 'event:' name, ignoring. Data: " + data);
|
|
|
|
|
return;
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
WebHandler<?> handler = handlers.get(eventName);
|
|
|
|
|
if (handler == null) {
|
|
|
|
|
log.warning("No handler registered for event type '" + eventName + "', ignoring");
|
|
|
|
|
return;
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
dispatchTyped(eventName, handler, data);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
// Captures the wildcard type from the handler so parsing/casting is type-safe.
|
|
|
|
|
private <T> void dispatchTyped(String eventName, WebHandler<T> handler, String data) {
|
|
|
|
|
try {
|
|
|
|
|
T event = objectMapper.readValue(data, handler.type());
|
|
|
|
|
handler.handle(event);
|
|
|
|
|
} catch (Exception e) {
|
|
|
|
|
log.log(Level.WARNING, "Failed to parse/handle event '" + eventName + "'", e);
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
private void sleepBeforeRetry() {
|
|
|
|
|
try {
|
|
|
|
|
Thread.sleep(RETRY_DELAY.toMillis());
|
|
|
|
|
} catch (InterruptedException e) {
|
|
|
|
|
Thread.currentThread().interrupt();
|
|
|
|
|
running = false;
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
}
|