Rsclub2_2/kst4contest
1
0
Fork 0
mirror of https://github.com/praktimarc/kst4contest.git synced 2026-08-27 12:37:03 +02:00
Code Issues Packages Projects Releases Wiki Activity

Fixed KST4Contest caused disconnect on getting no activities of ON4KST chatservers if no new lines arriving

Browse Source
This commit is contained in:
Marc Froehlich
2026-08-27 01:26:50 +02:00
parent 3193e4ac73
commit 7ff7248e7d
5 changed files with 321 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/PROJECT_CONTEXT.md
+9 -1
View File
@@ -1,6 +1,6 @@
# KST4Contest Project Context
Last reviewed: 2026-08-25
Last reviewed: 2026-08-27
This file is the durable technical project context for KST4Contest. It is not a user manual and not a replacement for the changelog. Current code, tests and authoritative external specifications remain the source of truth when this document is stale or ambiguous.
@@ -65,6 +65,14 @@ Known integration areas include:
CR/LF framing, XML framing, ports/transports, callsign normalization and frequency formatting are protocol behaviour and must not be changed as incidental cleanup.
### ON4KST session liveness
- After 90 seconds without inbound data, the application keeps the established empty CRLF heartbeat.
- At about 180 seconds of inbound idle time, the TCP session sends one `RDXQ|<main chat id>|` probe. The probe state belongs to the session, so a two-category login still sends only one probe per idle phase.
- Any subsequent inbound server frame confirms the probe. `DXQ` is accepted as the expected internal response and is not published as chat content.
- If no inbound frame arrives by about 210 seconds, the existing reconnect flow remains responsible for replacing the session.
- Probe diagnostics contain the session id, main category, opcode and timing only. They must not include credentials, complete server frames or normal chat messages.
## User Workflow / UI Invariants
- Contest operating speed and low-friction interaction are primary goals.
src/main/java/kst4contest/controller/MessageBusManagementThread.java
+12 -3
View File
@@ -877,6 +877,12 @@ public class MessageBusManagementThread extends Thread {
|| messageToProcess.getMessageText().isEmpty()) {
// No processable data.
} else {
if (On4KstProtocol.isConnectionProbeResponse(
messageToProcess.getMessageText())) {
// DXQ is the internal response to the active connection probe.
// Liveness was already recorded by the session manager.
return;
}
if (messageToProcess.getMessageText().startsWith(SRVR_LOGSTAT + "|")) {
String[] logstatMessage =
@@ -2177,8 +2183,11 @@ public class MessageBusManagementThread extends Thread {
// e.printStackTrace();
// }
System.out.println(messageTextRaw.getMessageText() + " <- RXed"); // Stdout at
// Console#######################################################TODO:Wichtig
if (!On4KstProtocol.isConnectionProbeResponse(
messageTextRaw.getMessageText())) {
System.out.println(messageTextRaw.getMessageText() + " <- RXed"); // Stdout at
// Console#######################################################TODO:Wichtig
}
try {
processRXMessage23001(messageTextRaw);
@@ -2207,4 +2216,4 @@ public class MessageBusManagementThread extends Thread {
System.out.println("Msgbusmgt: interrupt");
this.interrupt();
}
}
}
src/main/java/kst4contest/controller/On4KstConnectionManager.java
+146 -24
View File
@@ -48,6 +48,8 @@ final class On4KstConnectionManager {
static final long LOGIN_FALLBACK_MILLIS = 2_000L; //Login-Fallback
static final long HANDSHAKE_TIMEOUT_MILLIS = 45_000L; //Handshake-Timeout
static final long APPLICATION_HEARTBEAT_AFTER_MILLIS = 90_000L; //Application-Heartbeat
/** Idle duration after which the server is asked for current DX data. */
static final long CONNECTION_PROBE_AFTER_MILLIS = 180_000L; //Active connection probe
static final long INBOUND_STALE_AFTER_MILLIS = 210_000L; //Stale-Timeout - time without rxed data
static final List<Long> RECONNECT_DELAYS_MILLIS =
List.of(2_000L, 5_000L, 10_000L, 20_000L, 30_000L); //Reconnect-Backoff if no connection possible
@@ -178,7 +180,19 @@ final class On4KstConnectionManager {
session.lastInboundMillis.set(now);
session.lastProgressMillis.set(now);
String opcode = opcode(line);
String opcode = On4KstProtocol.opcode(line);
long probeResponseMillis = session.connectionProbe.acknowledge(now);
if (probeResponseMillis >= 0L) {
LOGGER.log(Level.INFO,
"ON4KST connection probe confirmed: session {0}, "
+ "received opcode {1}, response time {2} ms",
new Object[] {
sessionId,
opcode,
probeResponseMillis
});
}
if ("CK".equals(opcode)) {
sendHeartbeat(session);
}
@@ -244,7 +258,15 @@ final class On4KstConnectionManager {
new LinkedBlockingQueue<>();
LinkedBlockingQueue<ChatMessage> transmitQueue =
new LinkedBlockingQueue<>();
Session session = new Session(token, socket, receiveQueue, transmitQueue);
int mainCategory = controller.getChatPreferences()
.getLoginChatCategoryMain()
.getCategoryNumber();
Session session = new Session(
token,
socket,
receiveQueue,
transmitQueue,
mainCategory);
ReadThread readThread = new ReadThread(
token, socket, receiveQueue, this::isActiveSession,
@@ -252,8 +274,7 @@ final class On4KstConnectionManager {
failure -> onConnectionFailure(token, failure));
WriteThread writeThread = new WriteThread(
token, socket, transmitQueue,
controller.getChatPreferences().getLoginChatCategoryMain()
.getCategoryNumber(),
mainCategory,
this::isActiveSession,
failure -> onConnectionFailure(token, failure),
controller::onOn4KstOutboundFrameRejected);
@@ -535,6 +556,29 @@ final class On4KstConnectionManager {
session.transmitQueue.offer(heartbeat);
}
private void sendConnectionProbe(
Session session,
long now,
long inboundIdle
) {
if (session == null || !isActiveSession(session.id)
|| !session.connectionProbe.tryStart(now)) {
return;
}
LOGGER.log(Level.INFO,
"Sending ON4KST connection probe: session {0}, main category "
+ "{1}, inbound idle {2} seconds",
new Object[] {
session.id,
session.mainCategory,
inboundIdle / 1_000L
});
sendControl(
session,
On4KstProtocol.connectionProbe(session.mainCategory));
}
private void onConnectionFailure(long sessionId, Throwable failure) {
scheduler.execute(() -> failSession(sessionId, failure));
}
@@ -634,18 +678,47 @@ final class On4KstConnectionManager {
return;
}
long inboundIdle = now - session.lastInboundMillis.get();
if (inboundIdle > INBOUND_STALE_AFTER_MILLIS) {
failSession(session.id,
new SocketException("No ON4KST data received for "
+ inboundIdle / 1_000L + " seconds"));
long lastInboundMillis = session.lastInboundMillis.get();
long inboundIdle = now - lastInboundMillis;
IdleAction idleAction = determineIdleAction(
inboundIdle,
session.lastHeartbeatMillis.get() >= lastInboundMillis,
session.connectionProbe.isOutstanding());
if (session.lastInboundMillis.get() != lastInboundMillis) {
return;
}
if (inboundIdle > APPLICATION_HEARTBEAT_AFTER_MILLIS
&& session.lastHeartbeatMillis.get()
< session.lastInboundMillis.get()) {
sendHeartbeat(session);
switch (idleAction) {
case TIMEOUT -> {
if (session.lastInboundMillis.get() != lastInboundMillis) {
return;
}
long probeWaitMillis =
session.connectionProbe.responseWaitMillis(now);
if (probeWaitMillis >= 0L) {
LOGGER.log(Level.WARNING,
"ON4KST connection probe timed out: session "
+ "{0}, main category {1}, no response "
+ "for {2} ms, inbound idle {3} seconds; "
+ "reconnecting",
new Object[] {
session.id,
session.mainCategory,
probeWaitMillis,
inboundIdle / 1_000L
});
}
failSession(session.id,
new SocketException("No ON4KST data received for "
+ inboundIdle / 1_000L + " seconds"));
}
case CONNECTION_PROBE ->
sendConnectionProbe(session, now, inboundIdle);
case HEARTBEAT -> sendHeartbeat(session);
case NONE -> {
// The session is active or already has the required idle action.
}
}
} catch (RuntimeException exception) {
LOGGER.log(Level.WARNING,
@@ -653,6 +726,28 @@ final class On4KstConnectionManager {
}
}
/**
* Selects at most one maintenance action for the current inbound idle phase.
*/
static IdleAction determineIdleAction(
long inboundIdleMillis,
boolean heartbeatSentForIdlePhase,
boolean probeOutstanding
) {
if (inboundIdleMillis > INBOUND_STALE_AFTER_MILLIS) {
return IdleAction.TIMEOUT;
}
if (inboundIdleMillis >= CONNECTION_PROBE_AFTER_MILLIS
&& !probeOutstanding) {
return IdleAction.CONNECTION_PROBE;
}
if (inboundIdleMillis > APPLICATION_HEARTBEAT_AFTER_MILLIS
&& !heartbeatSentForIdlePhase) {
return IdleAction.HEARTBEAT;
}
return IdleAction.NONE;
}
private void validateConfiguration() {
ChatPreferences preferences = controller.getChatPreferences();
On4KstProtocol.login(
@@ -782,15 +877,6 @@ final class On4KstConnectionManager {
}
}
private String opcode(String line) {
if (line == null) {
return "";
}
int separator = line.indexOf('|');
return (separator < 0 ? line : line.substring(0, separator))
.trim().toUpperCase(Locale.ROOT);
}
private String describeFailure(Throwable failure) {
if (failure == null) {
return "unknown error";
@@ -808,12 +894,15 @@ final class On4KstConnectionManager {
private final Socket socket;
private final LinkedBlockingQueue<ChatMessage> receiveQueue;
private final LinkedBlockingQueue<ChatMessage> transmitQueue;
private final int mainCategory;
private final long connectedMillis = System.currentTimeMillis();
private final AtomicLong lastInboundMillis =
new AtomicLong(connectedMillis);
private final AtomicLong lastProgressMillis =
new AtomicLong(connectedMillis);
private final AtomicLong lastHeartbeatMillis = new AtomicLong();
private final ConnectionProbeState connectionProbe =
new ConnectionProbeState();
private final Map<Integer, Map<String, ChatMember>> initialMembers =
new ConcurrentHashMap<>();
@@ -831,12 +920,45 @@ final class On4KstConnectionManager {
long id,
Socket socket,
LinkedBlockingQueue<ChatMessage> receiveQueue,
LinkedBlockingQueue<ChatMessage> transmitQueue
LinkedBlockingQueue<ChatMessage> transmitQueue,
int mainCategory
) {
this.id = id;
this.socket = socket;
this.receiveQueue = receiveQueue;
this.transmitQueue = transmitQueue;
this.mainCategory = mainCategory;
}
}
}
/** Maintenance action selected by the session monitor. */
enum IdleAction {
NONE,
HEARTBEAT,
CONNECTION_PROBE,
TIMEOUT
}
/** Tracks one outstanding liveness probe for the complete TCP session. */
static final class ConnectionProbeState {
private final AtomicLong sentMillis = new AtomicLong();
boolean tryStart(long now) {
return now > 0L && sentMillis.compareAndSet(0L, now);
}
long acknowledge(long now) {
long sent = sentMillis.getAndSet(0L);
return sent == 0L ? -1L : Math.max(0L, now - sent);
}
boolean isOutstanding() {
return sentMillis.get() > 0L;
}
long responseWaitMillis(long now) {
long sent = sentMillis.get();
return sent == 0L ? -1L : Math.max(0L, now - sent);
}
}
}
src/main/java/kst4contest/controller/On4KstProtocol.java
+22 -1
View File
@@ -54,6 +54,27 @@ final class On4KstProtocol {
+ "|0|";
}
/** Builds the active liveness probe for the session's main chat. */
static String connectionProbe(int category) {
return "RDXQ|" + category(category) + "|";
}
/** Returns whether a server frame is the expected liveness-probe response. */
static boolean isConnectionProbeResponse(String frame) {
return "DXQ".equals(opcode(frame));
}
/** Extracts and normalizes the opcode without exposing the remaining frame. */
static String opcode(String frame) {
if (frame == null) {
return "";
}
int separator = frame.indexOf('|');
return (separator < 0 ? frame : frame.substring(0, separator))
.trim()
.toUpperCase(Locale.ROOT);
}
/** Builds a category-qualified locator command after validating Maidenhead syntax. */
static String setLocator(int category, String locator) {
return command(category, "/SETLOC " + locator(locator));
@@ -189,4 +210,4 @@ final class On4KstProtocol {
}
return category;
}
}
}
src/test/java/kst4contest/controller/On4KstConnectionProbeTest.java
+132
View File
@@ -0,0 +1,132 @@
package kst4contest.controller;
import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
import java.net.ServerSocket;
import java.net.Socket;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.Timeout;
import kst4contest.model.ChatMessage;
class On4KstConnectionProbeTest {
@Test
void buildsMainChatProbeAndAcceptsExpectedResponse() {
assertEquals("RDXQ|2|", On4KstProtocol.connectionProbe(2));
assertTrue(On4KstProtocol.isConnectionProbeResponse("DXQ|2|data|"));
assertFalse(On4KstProtocol.isConnectionProbeResponse(
"CH|2|123|DL1ABC|Name|0|text|0|"));
}
@Test
void selectsHeartbeatProbeAndTimeoutAtIdleBoundaries() {
assertEquals(
On4KstConnectionManager.IdleAction.NONE,
idleAction(90_000L, false, false));
assertEquals(
On4KstConnectionManager.IdleAction.HEARTBEAT,
idleAction(90_001L, false, false));
assertEquals(
On4KstConnectionManager.IdleAction.NONE,
idleAction(179_999L, true, false));
assertEquals(
On4KstConnectionManager.IdleAction.CONNECTION_PROBE,
idleAction(180_000L, true, false));
assertEquals(
On4KstConnectionManager.IdleAction.NONE,
idleAction(210_000L, true, true));
assertEquals(
On4KstConnectionManager.IdleAction.TIMEOUT,
idleAction(210_001L, true, true));
}
@Test
void oneSessionProbeIsAcknowledgedByAnyInboundTraffic() {
On4KstConnectionManager.ConnectionProbeState probe =
new On4KstConnectionManager.ConnectionProbeState();
assertTrue(probe.tryStart(1_000L));
assertFalse(probe.tryStart(1_001L),
"A second category must not start another session probe");
assertTrue(probe.isOutstanding());
assertEquals(250L, probe.acknowledge(1_250L));
assertFalse(probe.isOutstanding());
assertEquals(-1L, probe.acknowledge(1_500L));
assertTrue(probe.tryStart(2_000L),
"New inbound activity starts a new idle phase");
}
@Test
@Timeout(5)
void writerUsesExactCrLfForHeartbeatAndConnectionProbe() throws Exception {
byte[] expected = "\r\nRDXQ|2|\r\n".getBytes(StandardCharsets.UTF_8);
try (ServerSocket server = new ServerSocket(0)) {
CompletableFuture<byte[]> received = CompletableFuture.supplyAsync(() -> {
try (Socket accepted = server.accept()) {
accepted.setSoTimeout(2_000);
return accepted.getInputStream().readNBytes(expected.length);
} catch (Exception exception) {
throw new RuntimeException(exception);
}
});
try (Socket client = new Socket("127.0.0.1", server.getLocalPort())) {
LinkedBlockingQueue<ChatMessage> queue =
new LinkedBlockingQueue<>();
AtomicBoolean active = new AtomicBoolean(true);
WriteThread writer = new WriteThread(
11L,
client,
queue,
2,
ignored -> active.get(),
ignored -> { },
ignored -> { });
writer.start();
queue.add(serverFrame(""));
queue.add(serverFrame(On4KstProtocol.connectionProbe(2)));
assertArrayEquals(
expected,
received.get(2, TimeUnit.SECONDS));
active.set(false);
writer.interrupt();
writer.join(Duration.ofSeconds(2).toMillis());
}
}
}
private On4KstConnectionManager.IdleAction idleAction(
long inboundIdleMillis,
boolean heartbeatSent,
boolean probeOutstanding
) {
return On4KstConnectionManager.determineIdleAction(
inboundIdleMillis,
heartbeatSent,
probeOutstanding);
}
private ChatMessage serverFrame(String text) {
ChatMessage message = new ChatMessage();
message.setMessageDirectedToServer(true);
message.setMessageText(text);
return message;
}
}