mirror of
https://github.com/praktimarc/kst4contest.git
synced 2026-08-27 12:37:03 +02:00
Fixed KST4Contest caused disconnect on getting no activities of ON4KST chatservers if no new lines arriving
This commit is contained in:
@@ -1,6 +1,6 @@
|
|||||||
# KST4Contest Project Context
|
# 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.
|
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.
|
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
|
## User Workflow / UI Invariants
|
||||||
|
|
||||||
- Contest operating speed and low-friction interaction are primary goals.
|
- Contest operating speed and low-friction interaction are primary goals.
|
||||||
|
|||||||
@@ -877,6 +877,12 @@ public class MessageBusManagementThread extends Thread {
|
|||||||
|| messageToProcess.getMessageText().isEmpty()) {
|
|| messageToProcess.getMessageText().isEmpty()) {
|
||||||
// No processable data.
|
// No processable data.
|
||||||
} else {
|
} 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 + "|")) {
|
if (messageToProcess.getMessageText().startsWith(SRVR_LOGSTAT + "|")) {
|
||||||
String[] logstatMessage =
|
String[] logstatMessage =
|
||||||
@@ -2177,8 +2183,11 @@ public class MessageBusManagementThread extends Thread {
|
|||||||
// e.printStackTrace();
|
// e.printStackTrace();
|
||||||
// }
|
// }
|
||||||
|
|
||||||
System.out.println(messageTextRaw.getMessageText() + " <- RXed"); // Stdout at
|
if (!On4KstProtocol.isConnectionProbeResponse(
|
||||||
// Console#######################################################TODO:Wichtig
|
messageTextRaw.getMessageText())) {
|
||||||
|
System.out.println(messageTextRaw.getMessageText() + " <- RXed"); // Stdout at
|
||||||
|
// Console#######################################################TODO:Wichtig
|
||||||
|
}
|
||||||
|
|
||||||
try {
|
try {
|
||||||
processRXMessage23001(messageTextRaw);
|
processRXMessage23001(messageTextRaw);
|
||||||
@@ -2207,4 +2216,4 @@ public class MessageBusManagementThread extends Thread {
|
|||||||
System.out.println("Msgbusmgt: interrupt");
|
System.out.println("Msgbusmgt: interrupt");
|
||||||
this.interrupt();
|
this.interrupt();
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -48,6 +48,8 @@ final class On4KstConnectionManager {
|
|||||||
static final long LOGIN_FALLBACK_MILLIS = 2_000L; //Login-Fallback
|
static final long LOGIN_FALLBACK_MILLIS = 2_000L; //Login-Fallback
|
||||||
static final long HANDSHAKE_TIMEOUT_MILLIS = 45_000L; //Handshake-Timeout
|
static final long HANDSHAKE_TIMEOUT_MILLIS = 45_000L; //Handshake-Timeout
|
||||||
static final long APPLICATION_HEARTBEAT_AFTER_MILLIS = 90_000L; //Application-Heartbeat
|
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 long INBOUND_STALE_AFTER_MILLIS = 210_000L; //Stale-Timeout - time without rxed data
|
||||||
static final List<Long> RECONNECT_DELAYS_MILLIS =
|
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
|
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.lastInboundMillis.set(now);
|
||||||
session.lastProgressMillis.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)) {
|
if ("CK".equals(opcode)) {
|
||||||
sendHeartbeat(session);
|
sendHeartbeat(session);
|
||||||
}
|
}
|
||||||
@@ -244,7 +258,15 @@ final class On4KstConnectionManager {
|
|||||||
new LinkedBlockingQueue<>();
|
new LinkedBlockingQueue<>();
|
||||||
LinkedBlockingQueue<ChatMessage> transmitQueue =
|
LinkedBlockingQueue<ChatMessage> transmitQueue =
|
||||||
new LinkedBlockingQueue<>();
|
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(
|
ReadThread readThread = new ReadThread(
|
||||||
token, socket, receiveQueue, this::isActiveSession,
|
token, socket, receiveQueue, this::isActiveSession,
|
||||||
@@ -252,8 +274,7 @@ final class On4KstConnectionManager {
|
|||||||
failure -> onConnectionFailure(token, failure));
|
failure -> onConnectionFailure(token, failure));
|
||||||
WriteThread writeThread = new WriteThread(
|
WriteThread writeThread = new WriteThread(
|
||||||
token, socket, transmitQueue,
|
token, socket, transmitQueue,
|
||||||
controller.getChatPreferences().getLoginChatCategoryMain()
|
mainCategory,
|
||||||
.getCategoryNumber(),
|
|
||||||
this::isActiveSession,
|
this::isActiveSession,
|
||||||
failure -> onConnectionFailure(token, failure),
|
failure -> onConnectionFailure(token, failure),
|
||||||
controller::onOn4KstOutboundFrameRejected);
|
controller::onOn4KstOutboundFrameRejected);
|
||||||
@@ -535,6 +556,29 @@ final class On4KstConnectionManager {
|
|||||||
session.transmitQueue.offer(heartbeat);
|
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) {
|
private void onConnectionFailure(long sessionId, Throwable failure) {
|
||||||
scheduler.execute(() -> failSession(sessionId, failure));
|
scheduler.execute(() -> failSession(sessionId, failure));
|
||||||
}
|
}
|
||||||
@@ -634,18 +678,47 @@ final class On4KstConnectionManager {
|
|||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
long inboundIdle = now - session.lastInboundMillis.get();
|
long lastInboundMillis = session.lastInboundMillis.get();
|
||||||
if (inboundIdle > INBOUND_STALE_AFTER_MILLIS) {
|
long inboundIdle = now - lastInboundMillis;
|
||||||
failSession(session.id,
|
IdleAction idleAction = determineIdleAction(
|
||||||
new SocketException("No ON4KST data received for "
|
inboundIdle,
|
||||||
+ inboundIdle / 1_000L + " seconds"));
|
session.lastHeartbeatMillis.get() >= lastInboundMillis,
|
||||||
|
session.connectionProbe.isOutstanding());
|
||||||
|
|
||||||
|
if (session.lastInboundMillis.get() != lastInboundMillis) {
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
if (inboundIdle > APPLICATION_HEARTBEAT_AFTER_MILLIS
|
switch (idleAction) {
|
||||||
&& session.lastHeartbeatMillis.get()
|
case TIMEOUT -> {
|
||||||
< session.lastInboundMillis.get()) {
|
if (session.lastInboundMillis.get() != lastInboundMillis) {
|
||||||
sendHeartbeat(session);
|
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) {
|
} catch (RuntimeException exception) {
|
||||||
LOGGER.log(Level.WARNING,
|
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() {
|
private void validateConfiguration() {
|
||||||
ChatPreferences preferences = controller.getChatPreferences();
|
ChatPreferences preferences = controller.getChatPreferences();
|
||||||
On4KstProtocol.login(
|
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) {
|
private String describeFailure(Throwable failure) {
|
||||||
if (failure == null) {
|
if (failure == null) {
|
||||||
return "unknown error";
|
return "unknown error";
|
||||||
@@ -808,12 +894,15 @@ final class On4KstConnectionManager {
|
|||||||
private final Socket socket;
|
private final Socket socket;
|
||||||
private final LinkedBlockingQueue<ChatMessage> receiveQueue;
|
private final LinkedBlockingQueue<ChatMessage> receiveQueue;
|
||||||
private final LinkedBlockingQueue<ChatMessage> transmitQueue;
|
private final LinkedBlockingQueue<ChatMessage> transmitQueue;
|
||||||
|
private final int mainCategory;
|
||||||
private final long connectedMillis = System.currentTimeMillis();
|
private final long connectedMillis = System.currentTimeMillis();
|
||||||
private final AtomicLong lastInboundMillis =
|
private final AtomicLong lastInboundMillis =
|
||||||
new AtomicLong(connectedMillis);
|
new AtomicLong(connectedMillis);
|
||||||
private final AtomicLong lastProgressMillis =
|
private final AtomicLong lastProgressMillis =
|
||||||
new AtomicLong(connectedMillis);
|
new AtomicLong(connectedMillis);
|
||||||
private final AtomicLong lastHeartbeatMillis = new AtomicLong();
|
private final AtomicLong lastHeartbeatMillis = new AtomicLong();
|
||||||
|
private final ConnectionProbeState connectionProbe =
|
||||||
|
new ConnectionProbeState();
|
||||||
private final Map<Integer, Map<String, ChatMember>> initialMembers =
|
private final Map<Integer, Map<String, ChatMember>> initialMembers =
|
||||||
new ConcurrentHashMap<>();
|
new ConcurrentHashMap<>();
|
||||||
|
|
||||||
@@ -831,12 +920,45 @@ final class On4KstConnectionManager {
|
|||||||
long id,
|
long id,
|
||||||
Socket socket,
|
Socket socket,
|
||||||
LinkedBlockingQueue<ChatMessage> receiveQueue,
|
LinkedBlockingQueue<ChatMessage> receiveQueue,
|
||||||
LinkedBlockingQueue<ChatMessage> transmitQueue
|
LinkedBlockingQueue<ChatMessage> transmitQueue,
|
||||||
|
int mainCategory
|
||||||
) {
|
) {
|
||||||
this.id = id;
|
this.id = id;
|
||||||
this.socket = socket;
|
this.socket = socket;
|
||||||
this.receiveQueue = receiveQueue;
|
this.receiveQueue = receiveQueue;
|
||||||
this.transmitQueue = transmitQueue;
|
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);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@@ -54,6 +54,27 @@ final class On4KstProtocol {
|
|||||||
+ "|0|";
|
+ "|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. */
|
/** Builds a category-qualified locator command after validating Maidenhead syntax. */
|
||||||
static String setLocator(int category, String locator) {
|
static String setLocator(int category, String locator) {
|
||||||
return command(category, "/SETLOC " + locator(locator));
|
return command(category, "/SETLOC " + locator(locator));
|
||||||
@@ -189,4 +210,4 @@ final class On4KstProtocol {
|
|||||||
}
|
}
|
||||||
return category;
|
return category;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -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;
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user