diff --git a/pom.xml b/pom.xml index 5cfd89a..a167eb8 100644 --- a/pom.xml +++ b/pom.xml @@ -445,6 +445,7 @@ java.sql java.net.http jdk.crypto.ec + jdk.net ${main.class} ${project.build.directory}/modules diff --git a/src/main/java/kst4contest/controller/ChatController.java b/src/main/java/kst4contest/controller/ChatController.java index 2ec6213..bb9dcca 100644 --- a/src/main/java/kst4contest/controller/ChatController.java +++ b/src/main/java/kst4contest/controller/ChatController.java @@ -35,6 +35,8 @@ import java.util.function.Consumer; import java.util.function.Predicate; import java.nio.charset.StandardCharsets; import kst4contest.logic.FrequencyTextParser; +import java.util.logging.Level; +import java.util.logging.Logger; @@ -57,6 +59,9 @@ public class ChatController implements ThreadStatusCallback, PstRotatorEventList * */ + private static final Logger LOGGER = + Logger.getLogger(ChatController.class.getName()); + private static final boolean DEBUG_BAND_UPGRADE_HINT = true; //for new band hint @@ -86,6 +91,9 @@ public class ChatController implements ThreadStatusCallback, PstRotatorEventList boolean disconnected; boolean disconnectionPerformedByUser = false; + private final On4KstConnectionManager on4KstConnectionManager = + new On4KstConnectionManager(this); + public boolean isDisconnectionPerformedByUser() { return disconnectionPerformedByUser; @@ -140,6 +148,15 @@ public class ChatController implements ThreadStatusCallback, PstRotatorEventList return disconnected; } + /** + * Returns the authoritative ON4KST lifecycle state. + * + * @return current connection, authentication or synchronization state + */ + public On4KstConnectionState getOn4KstConnectionState() { + return on4KstConnectionManager.getState(); + } + public void setDisconnected(boolean disconnected) { this.disconnected = disconnected; } @@ -161,6 +178,112 @@ public class ChatController implements ThreadStatusCallback, PstRotatorEventList } else System.out.println("ERRRRRRRRRRRRRRRRRRRRRRRRRRRÖRRRRRRRRRRRRRRRRRRR"); } + /** + * Publishes one authoritative connection-state transition to legacy controller + * flags, generic worker status listeners and the dedicated UI callback. + * + *

The compatibility flags remain derived values. No caller may set them to + * infer socket health; only the connection manager owns that decision.

+ * + * @param state new lifecycle state + * @param detail human-readable progress or failure reason + * @param critical whether the transition should be emphasized as an error + */ + void updateOn4KstConnectionState( + On4KstConnectionState state, + String detail, + boolean critical + ) { + setConnectedAndLoggedIn(state == On4KstConnectionState.ONLINE); + setConnectedAndNOTLoggedIn( + state.isConnectionAttemptActive() + && state != On4KstConnectionState.ONLINE); + setDisconnected(state == On4KstConnectionState.DISCONNECTED); + + ThreadStateMessage status = new ThreadStateMessage( + "ON4KST", state.isConnectionAttemptActive(), detail, critical); + status.setRunningInformationTextDescription(state.name()); + onThreadStatus("ON4KST", status); + + if (statusListener != null) { + statusListener.onConnectionStateChanged(state, detail); + } + } + + /** + * Installs all resources belonging to one successfully opened connection + * generation. + * + *

The session-id check prevents a slow connection attempt from overwriting a + * newer socket and its queues.

+ */ + synchronized void installOn4KstSession( + long connectionSessionId, + Socket sessionSocket, + LinkedBlockingQueue receiveQueue, + LinkedBlockingQueue transmitQueue, + ReadThread sessionReadThread, + WriteThread sessionWriteThread, + MessageBusManagementThread sessionMessageProcessor + ) { + if (!on4KstConnectionManager.isActiveSession(connectionSessionId)) { + return; + } + this.socket = sessionSocket; + this.messageRXBus = receiveQueue; + this.messageTXBus = transmitQueue; + this.readThread = sessionReadThread; + this.writeThread = sessionWriteThread; + this.messageProcessor = sessionMessageProcessor; + } + + void onOn4KstLogstat(long connectionSessionId, String[] fields) { + on4KstConnectionManager.onLogstat(connectionSessionId, fields); + } + + void stageInitialOn4KstChatMember( + long connectionSessionId, + ChatMember member + ) { + on4KstConnectionManager.stageInitialChatMember( + connectionSessionId, member); + } + + void onOn4KstInitialUserListCompleted( + long connectionSessionId, + ChatCategory category + ) { + on4KstConnectionManager.onInitialUserListCompleted( + connectionSessionId, category); + } + + void onOn4KstConnectionOnline() { + scheduleBeaconTimer(INITIAL_BEACON_DELAY_MILLIS); + } + + void onOn4KstConnectionLost() { + stopBeaconTimer(); + } + + void onOn4KstOutboundFrameRejected(String reason) { + onOn4KstConnectionWarning("Outbound frame rejected locally: " + reason); + } + + /** + * Receives a non-fatal ON4KST diagnostic, writes it to the persistent warning + * log and forwards it to the status UI. + * + * @param warning sanitized diagnostic text; passwords and raw chat frames must + * never be included + */ + void onOn4KstConnectionWarning(String warning) { + LOGGER.log(Level.WARNING, "ON4KST warning: {0}", warning); + ThreadStateMessage status = new ThreadStateMessage( + "ON4KST", true, warning, false); + status.setRunningInformationTextDescription("WARNING"); + onThreadStatus("ON4KST", status); + } + /******************************************************************************** * PSTRotator controlling @@ -573,6 +696,16 @@ public class ChatController implements ThreadStatusCallback, PstRotatorEventList */ public void disconnect(String action) { + /* + * New connections are owned by the session manager. Keep the historic cleanup + * below as a compatibility fallback, but do not let it manipulate resources + * belonging to a replacement session. + */ + if (on4KstConnectionManager != null) { + disconnectManaged(action); + return; + } + // stopContextLoop(); //stops thread for calculating sked priorities stopScoreScheduler(); @@ -757,10 +890,8 @@ public class ChatController implements ThreadStatusCallback, PstRotatorEventList } } catch (IOException e) { - // TODO Auto-generated catch block e.printStackTrace(); } catch (Exception e2) { - // TODO Auto-generated catch block e2.printStackTrace(); } @@ -768,6 +899,69 @@ public class ChatController implements ThreadStatusCallback, PstRotatorEventList } + private void disconnectManaged(String action) { + setDisconnectionPerformedByUser(true); + on4KstConnectionManager.stopByUser(); + clearActiveChatMembers(); + runOnFxThread(() -> lst_clusterMemberList.clear()); + + stopBeaconTimer(); + stopScoreScheduler(); + stopDxClusterServer(); + cancelTimer(userActualizationtimer); + userActualizationtimer = null; + cancelTimer(ASQueryTimer); + ASQueryTimer = null; + + stopUdpReader(readUDPbyUCXThread, + chatPreferences.getLogsynch_ucxUDPWkdCallListenerPort()); + readUDPbyUCXThread = null; + stopUdpReader(readUDPByWintestThread, + chatPreferences.getLogsynch_wintestNetworkPort()); + stopWintestUdpListener(); + stopUdpReader(airScoutUDPReaderThread, + chatPreferences.getAirScout_asCommunicationPort()); + airScoutUDPReaderThread = null; + + if (ApplicationConstants.DISCSTRING_DISCONNECT_AND_CLOSE.equals(action)) { + if (dbHandler != null) { + dbHandler.closeDBConnection(); + } + if (rotatorClient != null) { + rotatorClient.stopRotor(); + rotatorClient.stop(); + rotatorClient = null; + } + } + } + + private void cancelTimer(Timer timer) { + if (timer != null) { + timer.cancel(); + timer.purge(); + } + } + + private void stopUdpReader(Thread readerThread, int listenerPort) { + if (readerThread != null) { + readerThread.interrupt(); + } + try (DatagramSocket datagramSocket = new DatagramSocket()) { + datagramSocket.setBroadcast(true); + byte[] poison = ApplicationConstants.DISCONNECT_RDR_POISONPILL.getBytes( + StandardCharsets.UTF_8); + DatagramPacket packet = new DatagramPacket( + poison, poison.length, + InetAddress.getByName("255.255.255.255"), listenerPort); + datagramSocket.send(packet); + } catch (IOException exception) { + System.out.println("[ChatController, warning]: could not wake UDP reader on port " + + listenerPort + ": " + exception.getMessage()); + } + } + + + // private ObservableList activeSkeds = FXCollections.observableArrayList(); // public ObservableList getActiveSkeds() { // return activeSkeds; @@ -1416,6 +1610,10 @@ private ObservableList private final List pendingChatMessages = new ArrayList<>(); private boolean chatMessageFlushScheduled = false; + private static final int RECENT_INBOUND_MESSAGE_KEYS_MAX = 5_000; + private final LinkedHashMap recentInboundMessageKeys = + new LinkedHashMap<>(); + /* * Same idea for DXCluster messages. */ @@ -1572,6 +1770,55 @@ private ObservableList runOnFxThread(() -> lst_chatMemberList.clear()); } + /** + * Atomically replaces one category after its terminating UE frame arrived. + * Partial UA0 snapshots are never exposed to the TableView. + * + * @param connectionSessionId session that produced the completed snapshot + * @param category chat category whose members are being replaced + * @param completeMembers validated members staged before the UE terminator + */ + public void replaceActiveChatMembersForCategory( + long connectionSessionId, + ChatCategory category, + Collection completeMembers + ) { + if (category == null + || !on4KstConnectionManager.isActiveSession(connectionSessionId)) { + return; + } + + int categoryNumber = category.getCategoryNumber(); + List safeMembers = completeMembers == null + ? List.of() : new ArrayList<>(completeMembers); + for (ChatMember member : safeMembers) { + initializeFrequencyFromStationNameIfUnambiguous(member); + } + + activeChatMembersByCallAndCategory.entrySet().removeIf(entry -> { + ChatMember member = entry.getValue(); + return member != null && member.getChatCategory() != null + && member.getChatCategory().getCategoryNumber() == categoryNumber; + }); + for (ChatMember member : safeMembers) { + String key = buildActiveChatMemberKey(member); + if (key != null) { + activeChatMembersByCallAndCategory.put(key, member); + } + } + + runOnFxThread(() -> { + if (!on4KstConnectionManager.isActiveSession(connectionSessionId)) { + return; + } + lst_chatMemberList.removeIf(member -> member != null + && member.getChatCategory() != null + && member.getChatCategory().getCategoryNumber() == categoryNumber); + lst_chatMemberList.addAll(safeMembers); + fireUserListUpdate("Complete ON4KST user list received"); + }); + } + /** * Resolves a member from the thread-safe active model. This avoids reading the * TableView backing list from MessageBusManagementThread. @@ -1913,6 +2160,9 @@ private ObservableList if (message == null) { return; } + if (isDuplicateInboundMessage(message)) { + return; + } synchronized (pendingChatMessagesLock) { pendingChatMessages.add(message); @@ -1927,6 +2177,47 @@ private ObservableList Platform.runLater(this::flushPendingChatMessagesToUi); } + /** + * Suppresses the small replay overlap deliberately requested after a reconnect. + * + *

The manager asks for messages beginning one timestamp before the last known + * message so that a boundary message cannot be lost. This bounded key cache + * removes the expected duplicate without growing for the lifetime of the + * application.

+ */ + private boolean isDuplicateInboundMessage(ChatMessage message) { + String timestamp = message.getMessageGeneratedTime(); + if (timestamp == null || timestamp.isBlank()) { + return false; + } + + String sender = message.getSender() == null + ? "" : String.valueOf(message.getSender().getCallSign()); + String receiver = message.getReceiver() == null + ? "" : String.valueOf(message.getReceiver().getCallSign()); + int category = message.getChatCategory() == null + ? -1 : message.getChatCategory().getCategoryNumber(); + String key = category + "|" + timestamp + "|" + sender + "|" + + receiver + "|" + String.valueOf(message.getMessageText()); + + synchronized (recentInboundMessageKeys) { + if (recentInboundMessageKeys.containsKey(key)) { + return true; + } + recentInboundMessageKeys.put(key, Boolean.TRUE); + while (recentInboundMessageKeys.size() + > RECENT_INBOUND_MESSAGE_KEYS_MAX) { + Iterator iterator = recentInboundMessageKeys.keySet().iterator(); + if (!iterator.hasNext()) { + break; + } + iterator.next(); + iterator.remove(); + } + return false; + } + } + private void flushPendingChatMessagesToUi() { List batch; @@ -2615,6 +2906,10 @@ private ObservableList */ public void execute() throws InterruptedException, IOException { + if (on4KstConnectionManager != null) { + executeManaged(); + return; + } chatController = this; @@ -2837,6 +3132,50 @@ private ObservableList } + private synchronized void executeManaged() throws IOException { + chatController = this; + setDisconnectionPerformedByUser(false); + startScoreScheduler(); + + if (readUDPbyUCXThread == null || !readUDPbyUCXThread.isAlive()) { + readUDPbyUCXThread = new ReadUDPbyUCXMessageThread( + chatPreferences.getLogsynch_ucxUDPWkdCallListenerPort(), + this, this); + readUDPbyUCXThread.setName("readUDPbyUCXThread"); + readUDPbyUCXThread.start(); + } + + if (chatPreferences.isLogsynch_wintestNetworkListenerEnabled()) { + startWintestUdpListener(); + } + + if (airScoutUDPReaderThread == null || !airScoutUDPReaderThread.isAlive()) { + airScoutUDPReaderThread = new ReadUDPbyAirScoutMessageThread( + chatPreferences.getAirScout_asCommunicationPort(), + this, this); + airScoutUDPReaderThread.setName("airscoutudpreaderThread"); + airScoutUDPReaderThread.start(); + } + + cancelTimer(userActualizationtimer); + userActualizationtimer = new Timer("UserActualizationTimer", true); + userActualizationtimer.schedule( + new UserActualizationTask(this), 4_000L, 60_000L); + + cancelTimer(ASQueryTimer); + ASQueryTimer = new Timer("AirScoutQueryTimer", true); + ASQueryTimer.schedule( + new AirScoutPeriodicalAPReflectionInquirerTask(this), + 10_000L, 60_000L); + + if (chatPreferences.isStn_pstRotatorEnabled() && rotatorClient == null) { + initRotor(); + } + startDxClusterServerIfEnabled(); + on4KstConnectionManager.start(); + } + + /** * Returns the background reachability service used by the station table. @@ -2964,6 +3303,11 @@ private ObservableList */ public void initialize23001() throws InterruptedException, IOException { + if (on4KstConnectionManager != null) { + on4KstConnectionManager.start(); + return; + } + messageTXBus.clear(); ChatMessage message = new ChatMessage(); diff --git a/src/main/java/kst4contest/controller/MessageBusManagementThread.java b/src/main/java/kst4contest/controller/MessageBusManagementThread.java index 204afbf..f2e81b8 100644 --- a/src/main/java/kst4contest/controller/MessageBusManagementThread.java +++ b/src/main/java/kst4contest/controller/MessageBusManagementThread.java @@ -19,6 +19,8 @@ import kst4contest.locatorUtils.DirectionUtils; import kst4contest.locatorUtils.Location; import kst4contest.model.*; import kst4contest.logic.FrequencyTextParser; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.function.LongPredicate; /** * @@ -37,6 +39,11 @@ public class MessageBusManagementThread extends Thread { private PrintWriter writer; // private Socket socket; private ChatController client; + + private final long connectionSessionId; + private final LinkedBlockingQueue receiveQueue; + private final LongPredicate connectionSessionIsActive; + // private File fileLogRAW; // private TimerTask userActualizationTask; // Is used as a temporary userout-print // private TimerTask userActualizationTask; //kst4contest.test 4 23001 @@ -118,9 +125,22 @@ public class MessageBusManagementThread extends Thread { } public MessageBusManagementThread(ChatController client, ThreadStatusCallback callBack) { + this(client, callBack, 0L, client.getMessageRXBus(), ignored -> true); + } + + public MessageBusManagementThread( + ChatController client, + ThreadStatusCallback callBack, + long connectionSessionId, + LinkedBlockingQueue receiveQueue, + LongPredicate connectionSessionIsActive + ) { this.callBackToController = callBack; this.client = client; + this.connectionSessionId = connectionSessionId; + this.receiveQueue = receiveQueue; + this.connectionSessionIsActive = connectionSessionIsActive; ThreadStateMessage threadStateMessage = new ThreadStateMessage(this.ThreadNickName, true, "initialized", false); callBackToController.onThreadStatus(ThreadNickName,threadStateMessage); @@ -522,7 +542,8 @@ public class MessageBusManagementThread extends Thread { messageToProcess.setMessageText(reduce); - if (messageToProcess.getMessageText().isEmpty()) { + if (messageToProcess.getMessageText() == null + || messageToProcess.getMessageText().isEmpty()) { // System.out.println("[MSGBUSMGTT:] ###################### no processable data"); } else { @@ -754,6 +775,50 @@ public class MessageBusManagementThread extends Thread { return fallbackReceiver; } + private boolean validateInboundUserFrame(String[] fields) { + if (fields == null || fields.length < 6) { + logRejectedInboundUserFrame( + "Ignoring truncated ON4KST user frame", fields); + return false; + } + if (fields[2] == null || fields[2].isBlank()) { + logRejectedInboundUserFrame( + "Ignoring ON4KST user frame with empty callsign", fields); + return false; + } + try { + On4KstProtocol.category(Integer.parseInt(fields[1])); + On4KstProtocol.locator(fields[4]); + Integer.parseInt(fields[5]); + return true; + } catch (IllegalArgumentException invalidUser) { + logRejectedInboundUserFrame( + "Ignoring malformed ON4KST user '" + fields[2] + + "': " + invalidUser.getMessage(), fields); + return false; + } + } + + /** + * Records a rejected user frame without writing the complete raw frame or user + * name to the diagnostic log. + * + *

The opcode, category and callsign are sufficient to identify the offending + * list position. Omitting the remaining fields avoids unnecessary disclosure of + * free-form profile text.

+ */ + private void logRejectedInboundUserFrame(String reason, String[] fields) { + String opcode = fields != null && fields.length > 0 ? fields[0] : "UNKNOWN"; + String category = fields != null && fields.length > 1 ? fields[1] : "UNKNOWN"; + String callsign = fields != null && fields.length > 2 ? fields[2] : "UNKNOWN"; + client.onOn4KstConnectionWarning( + reason + "; opcode=" + opcode + + ", category=" + category + + ", callsign=" + callsign + + ", fieldCount=" + (fields == null ? 0 : fields.length)); + } + + /** * Processes received messages via port 23001 (improved telnet Interface) * @@ -808,23 +873,33 @@ public class MessageBusManagementThread extends Thread { * here we have a helper list for identifying questions for my qrg which can be autoanswered later */ - if (messageToProcess.getMessageText().isEmpty()) { -// System.out.println("[MSGBUSMGTT:] no processable data"); - + if (messageToProcess.getMessageText() == null + || messageToProcess.getMessageText().isEmpty()) { + // No processable data. } else { - if (messageToProcess.getMessageText().contains(SRVR_LOGSTAT)) { - String logstatMessage[]; - logstatMessage = messageToProcess.getMessageText().split("\\|"); - if (logstatMessage[1].contains(SRVR_LOGINOK)) { - this.client.setConnectedAndLoggedIn(true); - } else { - this.client.setConnectedAndNOTLoggedIn(true); - this.client.setConnectedAndLoggedIn(false); - } + if (messageToProcess.getMessageText().startsWith(SRVR_LOGSTAT + "|")) { + String[] logstatMessage = + messageToProcess.getMessageText().split("\\|", -1); + this.client.onOn4KstLogstat( + connectionSessionId, + logstatMessage); } - String splittedMessageLine[] = messageToProcess.getMessageText().split("\\|"); + String[] splittedMessageLine = + messageToProcess.getMessageText().split("\\|"); + + String opcode = splittedMessageLine.length == 0 + ? "" + : splittedMessageLine[0]; + + if ((INITIALUSERLISTENTRY.equals(opcode) + || USERENTEREDCHAT.equals(opcode) + || USERENTEREDCHAT2.equals(opcode)) + && !validateInboundUserFrame(splittedMessageLine)) { + return; + } +// String splittedMessageLine[] = messageToProcess.getMessageText().split("\\|"); /** * Initializes the Userlist if entry fits UA0 @@ -832,7 +907,7 @@ public class MessageBusManagementThread extends Thread { * * */ - if (splittedMessageLine[0].contains(INITIALUSERLISTENTRY)) { + if (splittedMessageLine[0].equals(INITIALUSERLISTENTRY)) { // System.out.println("MSGBUS: User detected"); ChatMember newMember = new ChatMember(); @@ -853,7 +928,9 @@ public class MessageBusManagementThread extends Thread { if (!client.getChatPreferences().getStn_loginCallSign().equals(newMember.getCallSign())) { - this.client.addOrUpdateActiveChatMember(newMember); // the own call will not be in the list + this.client.stageInitialOn4KstChatMember( + connectionSessionId, + newMember); // this.client.getReachabilityService().ensureAutoTropoMarginCalculated(newMember); // Reachability is calculated on demand only: map click, selected station, or manual request. } @@ -877,7 +954,8 @@ public class MessageBusManagementThread extends Thread { * UA2|2|W5ADD|Parker|EM40WL|2| * */ - if (splittedMessageLine[0].contains(USERENTEREDCHAT) || splittedMessageLine[0].contains(USERENTEREDCHAT2)) { + if (splittedMessageLine[0].equals(USERENTEREDCHAT) + || splittedMessageLine[0].equals(USERENTEREDCHAT2)) { // System.out.println("MSGBUS: User detected"); @@ -1640,18 +1718,29 @@ public class MessageBusManagementThread extends Thread { /** * Userinfo-update: UE|2|22562| */ - if (splittedMessageLine[0].contains(SRVR_USERLISTEND)) { + if (SRVR_USERLISTEND.equals(opcode)) { + if (splittedMessageLine.length < 2) { + System.out.println( + "[MSGBUSMGT, Warning:] Ignoring malformed UE frame: " + + messageToProcess.getMessageText()); + return; + } - // No worthy information, count of users - } else + this.client.onOn4KstInitialUserListCompleted( + connectionSessionId, + util_getChatCategoryByCategoryNrString( + splittedMessageLine[1])); - if (splittedMessageLine[0].contains(SRVR_DXCEND)) { + } else if (SRVR_DXCEND.equals(opcode)) { - // No worthy information, count of users - } else + // DF marks the end of the initial DX-cluster data. + // The frame contains no data that needs to be published. + + } else if (SRVR_COMMUNICATIONK.equals(opcode)) { + + // CK is a regular server delimiter/acknowledgement. + // It is intentionally accepted without further processing. - if (splittedMessageLine[0].contains(SRVR_COMMUNICATIONK)) { - // No worthy information, end of srvrmsgs } else //-> LOGSTAT|114|Wrong password!| @@ -1931,13 +2020,17 @@ public class MessageBusManagementThread extends Thread { while (true) { try { - messageTextRaw = client.getMessageRXBus().take(); + messageTextRaw = receiveQueue.take(); - if (messageTextRaw.getMessageText().equals(ApplicationConstants.DISCONNECT_RDR_POISONPILL) && messageTextRaw.getMessageSenderName().equals(ApplicationConstants.DISCONNECT_RDR_POISONPILL)) { - client.getMessageRXBus().clear(); + if (ApplicationConstants.DISCONNECT_RDR_POISONPILL.equals(messageTextRaw.getMessageText()) + && ApplicationConstants.DISCONNECT_RDR_POISONPILL.equals(messageTextRaw.getMessageSenderName())) { + receiveQueue.clear(); break; } else { + if (!connectionSessionIsActive.test(connectionSessionId)) { + break; + } messageLine = messageTextRaw.getMessageText(); /*********************************************** diff --git a/src/main/java/kst4contest/controller/On4KstConnectionManager.java b/src/main/java/kst4contest/controller/On4KstConnectionManager.java new file mode 100644 index 0000000..ecbac75 --- /dev/null +++ b/src/main/java/kst4contest/controller/On4KstConnectionManager.java @@ -0,0 +1,837 @@ +package kst4contest.controller; + +import java.io.IOException; +import java.net.InetSocketAddress; +import java.net.Socket; +import java.net.SocketException; +import java.time.Duration; +import java.time.LocalDateTime; +import java.time.ZoneOffset; +import java.time.format.DateTimeFormatter; +import java.time.format.DateTimeParseException; +import java.util.ArrayList; +import java.util.Collection; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; +import java.util.logging.Level; +import java.util.logging.Logger; + +import jdk.net.ExtendedSocketOptions; +import kst4contest.ApplicationConstants; +import kst4contest.model.ChatCategory; +import kst4contest.model.ChatMember; +import kst4contest.model.ChatMessage; +import kst4contest.model.ChatPreferences; + +/** + * Owns the complete lifecycle of the single ON4KST TCP session. + * + *

Every reader, writer, queue and parser belongs to an immutable session id. + * A delayed failure from an old socket can therefore never close or consume data + * from its replacement.

+ */ +final class On4KstConnectionManager { + private static final Logger LOGGER = + Logger.getLogger(On4KstConnectionManager.class.getName()); + private static final DateTimeFormatter LIVE_MESSAGE_TIMESTAMP = + DateTimeFormatter.ofPattern("yyyyMMddHHmmss"); + + static final int CONNECT_TIMEOUT_MILLIS = 10_000; //TCP-Connect-Timeout + 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 + static final long INBOUND_STALE_AFTER_MILLIS = 210_000L; //Stale-Timeout - time without rxed data + static final List RECONNECT_DELAYS_MILLIS = + List.of(2_000L, 5_000L, 10_000L, 20_000L, 30_000L); //Reconnect-Backoff if no connection possible + + private final ChatController controller; + private final ScheduledExecutorService scheduler; + private final AtomicLong generation = new AtomicLong(); + private final AtomicLong lastReceivedMessageTimestamp = new AtomicLong(); + + private volatile Session activeSession; + private volatile On4KstConnectionState state = + On4KstConnectionState.DISCONNECTED; + private volatile boolean stopRequested = true; + private int reconnectAttempt; + + On4KstConnectionManager(ChatController controller) { + this.controller = controller; + this.scheduler = Executors.newSingleThreadScheduledExecutor(runnable -> { + Thread thread = new Thread(runnable, "On4KstConnectionSupervisor"); + thread.setDaemon(true); + return thread; + }); + this.scheduler.scheduleAtFixedRate( + this::monitorActiveSession, 5L, 5L, TimeUnit.SECONDS); + LOGGER.fine("ON4KST connection supervisor initialized"); + } + + /** + * Returns the last lifecycle state published by the connection supervisor. + * + * @return current immutable connection-state value + */ + On4KstConnectionState getState() { + return state; + } + + /** + * Verifies that a callback still belongs to the currently installed session. + * + *

Every reconnect receives a new id. Late EOF, write or parser callbacks from + * an obsolete socket therefore become harmless instead of closing the replacement + * connection.

+ * + * @param sessionId id captured by the calling worker + * @return {@code true} only for the current, open and non-stopped session + */ + boolean isActiveSession(long sessionId) { + Session session = activeSession; + return session != null + && session.id == sessionId + && !session.closed + && !stopRequested; + } + + /** + * Starts a non-blocking connection attempt. + * + *

Configuration is validated before a socket is opened. A duplicate Connect + * action is ignored while another attempt or session is active. Connection work + * runs on the supervisor executor, so an unreachable server cannot block the + * JavaFX application thread.

+ */ + void start() { + long token; + synchronized (this) { + if (!stopRequested && state.isConnectionAttemptActive()) { + return; + } + + try { + validateConfiguration(); + } catch (IllegalArgumentException invalidConfiguration) { + stopRequested = true; + transition(On4KstConnectionState.DISCONNECTED, + "Invalid ON4KST configuration: " + + invalidConfiguration.getMessage(), true); + return; + } + + stopRequested = false; + reconnectAttempt = 0; + token = generation.incrementAndGet(); + transition(On4KstConnectionState.CONNECTING, + "Opening ON4KST connection", false); + } + + scheduler.execute(() -> openConnection(token)); + } + + /** + * Stops the current session and invalidates every scheduled callback or reconnect + * belonging to it. + */ + void stopByUser() { + Session oldSession; + synchronized (this) { + stopRequested = true; + generation.incrementAndGet(); + transition(On4KstConnectionState.STOPPING, + "Disconnecting from ON4KST", false); + oldSession = activeSession; + activeSession = null; + } + + closeSession(oldSession); + controller.onOn4KstConnectionLost(); + transition(On4KstConnectionState.DISCONNECTED, + "Disconnected by user", false); + } + + /** + * Records one received protocol line as proof of application-level liveness. + * + *

TCP's {@code isConnected()} only states that a connection once succeeded. + * It does not prove that the peer is still reachable. Updating the inbound + * timestamp here gives the monitor a meaningful end-to-end signal.

+ * + * @param sessionId immutable source-session id + * @param line complete protocol line received from ON4KST + */ + void onInboundActivity(long sessionId, String line) { + Session session = activeSession; + if (session == null || session.id != sessionId || session.closed) { + return; + } + + long now = System.currentTimeMillis(); + session.lastInboundMillis.set(now); + session.lastProgressMillis.set(now); + + String opcode = opcode(line); + if ("CK".equals(opcode)) { + sendHeartbeat(session); + } + + if (!session.loginSent + && line != null + && line.toLowerCase(Locale.ROOT).contains("login")) { + scheduler.execute(() -> sendLogin(sessionId)); + } + + if ("CH".equals(opcode) || "CR".equals(opcode)) { + recordHistoryTimestamp(line); + } + } + + void onLogstat(long sessionId, String[] fields) { + String[] copy = fields == null ? new String[0] : fields.clone(); + scheduler.execute(() -> handleLogstat(sessionId, copy)); + } + + void stageInitialChatMember(long sessionId, ChatMember member) { + Session session = activeSession; + if (session == null || session.id != sessionId || member == null + || member.getChatCategory() == null || member.getCallSign() == null) { + return; + } + + int category = member.getChatCategory().getCategoryNumber(); + session.initialMembers + .computeIfAbsent(category, ignored -> new ConcurrentHashMap<>()) + .put(member.getCallSign().trim().toUpperCase(Locale.ROOT), member); + session.lastProgressMillis.set(System.currentTimeMillis()); + } + + void onInitialUserListCompleted(long sessionId, ChatCategory category) { + if (category == null) { + return; + } + scheduler.execute(() -> completeInitialUserList( + sessionId, category.getCategoryNumber())); + } + + private void openConnection(long token) { + if (!mayOpen(token)) { + return; + } + + LOGGER.log(Level.INFO, + "Opening ON4KST TCP session {0}", token); + Socket socket = new Socket(); + try { + ChatPreferences preferences = controller.getChatPreferences(); + socket.connect(new InetSocketAddress( + preferences.getStn_on4kstServersDns(), + preferences.getStn_on4kstServersPort()), + CONNECT_TIMEOUT_MILLIS); + configureSocket(socket); + LOGGER.log(Level.INFO, + "ON4KST TCP session {0} connected to {1}", + new Object[] {token, socket.getRemoteSocketAddress()}); + + LinkedBlockingQueue receiveQueue = + new LinkedBlockingQueue<>(); + LinkedBlockingQueue transmitQueue = + new LinkedBlockingQueue<>(); + Session session = new Session(token, socket, receiveQueue, transmitQueue); + + ReadThread readThread = new ReadThread( + token, socket, receiveQueue, this::isActiveSession, + line -> onInboundActivity(token, line), + failure -> onConnectionFailure(token, failure)); + WriteThread writeThread = new WriteThread( + token, socket, transmitQueue, + controller.getChatPreferences().getLoginChatCategoryMain() + .getCategoryNumber(), + this::isActiveSession, + failure -> onConnectionFailure(token, failure), + controller::onOn4KstOutboundFrameRejected); + MessageBusManagementThread messageProcessor = + new MessageBusManagementThread( + controller, controller, token, receiveQueue, + this::isActiveSession); + + session.readThread = readThread; + session.writeThread = writeThread; + session.messageProcessor = messageProcessor; + + synchronized (this) { + if (!mayOpen(token)) { + closeSession(session); + return; + } + activeSession = session; + controller.installOn4KstSession( + token, socket, receiveQueue, transmitQueue, + readThread, writeThread, messageProcessor); + transition(On4KstConnectionState.WAITING_FOR_LOGIN_PROMPT, + "TCP connected; waiting for ON4KST login prompt", false); + } + + messageProcessor.start(); + writeThread.start(); + readThread.start(); + scheduler.schedule( + () -> sendLogin(token), LOGIN_FALLBACK_MILLIS, + TimeUnit.MILLISECONDS); + } catch (Exception exception) { + try { + socket.close(); + } catch (IOException ignored) { + // The original connection exception is more useful. + } + scheduler.execute(() -> handleOpenFailure(token, exception)); + } + } + + private boolean mayOpen(long token) { + return !stopRequested && generation.get() == token; + } + + private void sendLogin(long sessionId) { + Session session = activeSession; + if (session == null || session.id != sessionId || session.loginSent + || session.closed || stopRequested) { + return; + } + + try { + ChatPreferences preferences = controller.getChatPreferences(); + int mainCategory = preferences.getLoginChatCategoryMain() + .getCategoryNumber(); + long historyFrom = Math.max( + 0L, lastReceivedMessageTimestamp.get() - 1L); + String login = On4KstProtocol.login( + preferences.getStn_loginCallSign(), + preferences.getStn_loginPassword(), + mainCategory, + "KST4Contest v" + ApplicationConstants.APPLICATION_CURRENT_VERSION, + historyFrom); + + session.loginSent = true; + session.lastProgressMillis.set(System.currentTimeMillis()); + LOGGER.log(Level.INFO, + "Sending ON4KST login for session {0}, main category {1}", + new Object[] {sessionId, mainCategory}); + transition(On4KstConnectionState.AUTHENTICATING, + "ON4KST login sent", false); + sendControl(session, login); + } catch (IllegalArgumentException invalidConfiguration) { + failPermanently(session, + "Invalid ON4KST login configuration: " + + invalidConfiguration.getMessage()); + } + } + + private void handleLogstat(long sessionId, String[] fields) { + Session session = activeSession; + if (session == null || session.id != sessionId || session.closed) { + return; + } + + String code = fields.length > 1 ? fields[1] : ""; + if (!"100".equals(code)) { + String serverText = fields.length > 2 ? fields[2] : "Login rejected"; + failPermanently(session, + "ON4KST login rejected (" + code + "): " + serverText); + return; + } + + if (session.authenticated) { + return; + } + session.authenticated = true; + session.lastProgressMillis.set(System.currentTimeMillis()); + LOGGER.log(Level.INFO, + "ON4KST login accepted for session {0}", sessionId); + + int mainCategory = controller.getChatPreferences() + .getLoginChatCategoryMain().getCategoryNumber(); + transition(On4KstConnectionState.SYNCING_MAIN_CHAT, + "Login accepted; loading main chat", false); + sendControl(session, On4KstProtocol.settingsDone(mainCategory)); + } + + /** + * Publishes the initial user snapshot for one chat category exactly once. + * + *

ON4KST can send further {@code UE} frames after live user updates or + * after commands such as {@code SETNAME} and {@code BACK}. Those frames do + * not announce a new, empty snapshot. Treating them as another initial-list + * completion would remove the already published members because the staging + * map was consumed by the first {@code UE} frame.

+ * + *

The completed-category set is updated before the staging map is removed. + * This makes the operation idempotent even if completion callbacks should + * later be invoked from more than one thread. A genuinely empty initial list + * remains valid: the first {@code UE} for a category is always processed, + * even when no preceding valid {@code UA0} frame was staged.

+ * + * @param sessionId immutable id of the socket session that received the frame + * @param categoryNumber numeric ON4KST category terminated by {@code UE} + */ + private void completeInitialUserList(long sessionId, int categoryNumber) { + Session session = activeSession; + if (session == null || session.id != sessionId || session.closed) { + return; + } + + if (!session.completedInitialUserLists.add(categoryNumber)) { + LOGGER.log(Level.FINE, + "ON4KST session {0}: ignoring duplicate user-list end " + + "marker for category {1}; the initial snapshot " + + "has already been published", + new Object[] {sessionId, categoryNumber}); + return; + } + + Map staged = + session.initialMembers.remove(categoryNumber); + Collection completeMembers = staged == null + ? List.of() + : new ArrayList<>(staged.values()); + + LOGGER.log(Level.INFO, + "ON4KST session {0}: complete user list for category {1} " + + "contains {2} valid users", + new Object[] { + sessionId, + categoryNumber, + completeMembers.size() + }); + + controller.replaceActiveChatMembersForCategory( + sessionId, + new ChatCategory(categoryNumber), + completeMembers); + + ChatPreferences preferences = controller.getChatPreferences(); + int mainCategory = + preferences.getLoginChatCategoryMain().getCategoryNumber(); + + if (categoryNumber == mainCategory && !session.mainListComplete) { + session.mainListComplete = true; + configureMainChat(session); + + if (hasDistinctSecondChat(preferences)) { + int secondCategory = + preferences.getLoginChatCategorySecond() + .getCategoryNumber(); + + transition( + On4KstConnectionState.SYNCING_SECOND_CHAT, + "Main chat ready; loading second chat", + false); + + sendControl( + session, + On4KstProtocol.addChat( + secondCategory, + Math.max( + 0L, + lastReceivedMessageTimestamp.get() - 1L))); + } else { + markOnline(session); + } + return; + } + + if (hasDistinctSecondChat(preferences) + && categoryNumber + == preferences.getLoginChatCategorySecond() + .getCategoryNumber() + && !session.secondListComplete) { + session.secondListComplete = true; + configureSecondChat(session); + markOnline(session); + } + } + + private void configureMainChat(Session session) { + ChatPreferences preferences = controller.getChatPreferences(); + int category = preferences.getLoginChatCategoryMain().getCategoryNumber(); + sendControl(session, On4KstProtocol.setLocator( + category, preferences.getStn_loginLocatorMainCat())); + if (preferences.getStn_loginNameMainCat() != null + && !preferences.getStn_loginNameMainCat().isBlank()) { + sendControl(session, On4KstProtocol.setName( + category, preferences.getStn_loginNameMainCat())); + } + sendControl(session, On4KstProtocol.back(category)); + + String secondLocator = preferences.getStn_loginLocatorSecondCat(); + String mainLocator = preferences.getStn_loginLocatorMainCat(); + if (preferences.isLoginToSecondChatEnabled() + && secondLocator != null && !secondLocator.isBlank() + && !secondLocator.equalsIgnoreCase(mainLocator)) { + controller.onOn4KstConnectionWarning( + "ON4KST uses one locator per TCP session. The second-chat locator '" + + secondLocator + "' is ignored; using '" + mainLocator + "'."); + } + } + + private void configureSecondChat(Session session) { + ChatPreferences preferences = controller.getChatPreferences(); + int category = preferences.getLoginChatCategorySecond().getCategoryNumber(); + if (preferences.getStn_loginNameSecondCat() != null + && !preferences.getStn_loginNameSecondCat().isBlank()) { + sendControl(session, On4KstProtocol.setName( + category, preferences.getStn_loginNameSecondCat())); + } + sendControl(session, On4KstProtocol.back(category)); + } + + private void markOnline(Session session) { + if (!isActiveSession(session.id)) { + return; + } + reconnectAttempt = 0; + session.online = true; + session.lastProgressMillis.set(System.currentTimeMillis()); + LOGGER.log(Level.INFO, + "ON4KST session {0} is authenticated and synchronized", + session.id); + transition(On4KstConnectionState.ONLINE, + "ON4KST session is authenticated and synchronized", false); + controller.onOn4KstConnectionOnline(); + } + + private void sendControl(Session session, String frame) { + if (session == null || !isActiveSession(session.id)) { + return; + } + ChatMessage message = new ChatMessage(); + message.setMessageDirectedToServer(true); + message.setMessageText(frame); + session.transmitQueue.offer(message); + } + + private void sendHeartbeat(Session session) { + if (session == null || !isActiveSession(session.id)) { + return; + } + long now = System.currentTimeMillis(); + session.lastHeartbeatMillis.set(now); + LOGGER.log(Level.FINE, + "Sending application heartbeat for ON4KST session {0}", + session.id); + ChatMessage heartbeat = new ChatMessage(); + heartbeat.setMessageDirectedToServer(true); + heartbeat.setMessageText(""); + session.transmitQueue.offer(heartbeat); + } + + private void onConnectionFailure(long sessionId, Throwable failure) { + scheduler.execute(() -> failSession(sessionId, failure)); + } + + private void failSession(long sessionId, Throwable failure) { + Session failedSession; + synchronized (this) { + failedSession = activeSession; + if (failedSession == null || failedSession.id != sessionId + || failedSession.closed) { + return; + } + activeSession = null; + failedSession.closed = true; + } + + closeSession(failedSession); + controller.onOn4KstConnectionLost(); + if (stopRequested) { + transition(On4KstConnectionState.DISCONNECTED, + "ON4KST connection stopped", false); + return; + } + + scheduleReconnect(failure); + } + + private void handleOpenFailure(long token, Throwable failure) { + if (!mayOpen(token)) { + return; + } + controller.onOn4KstConnectionLost(); + scheduleReconnect(failure); + } + + private void scheduleReconnect(Throwable failure) { + if (stopRequested) { + transition(On4KstConnectionState.DISCONNECTED, + "ON4KST connection stopped", false); + return; + } + + String reason = describeFailure(failure); + LOGGER.log(Level.WARNING, + "ON4KST connection lost; automatic reconnect scheduled", failure); + long delay = RECONNECT_DELAYS_MILLIS.get(Math.min( + reconnectAttempt, RECONNECT_DELAYS_MILLIS.size() - 1)); + reconnectAttempt++; + transition(On4KstConnectionState.RECONNECT_WAIT, + "Connection lost (" + reason + "); reconnecting in " + + Duration.ofMillis(delay).toSeconds() + " s", true); + + long nextToken = generation.incrementAndGet(); + scheduler.schedule(() -> { + if (!mayOpen(nextToken)) { + return; + } + transition(On4KstConnectionState.CONNECTING, + "Reconnecting to ON4KST", false); + openConnection(nextToken); + }, delay, TimeUnit.MILLISECONDS); + } + + private void failPermanently(Session session, String reason) { + if (session == null || !isActiveSession(session.id)) { + return; + } + stopRequested = true; + generation.incrementAndGet(); + activeSession = null; + session.closed = true; + closeSession(session); + controller.onOn4KstConnectionLost(); + LOGGER.log(Level.WARNING, reason); + transition(On4KstConnectionState.DISCONNECTED, reason, true); + } + + private void monitorActiveSession() { + try { + Session session = activeSession; + if (session == null || session.closed || stopRequested) { + return; + } + + if (session.socket.isClosed()) { + failSession(session.id, + new SocketException("Socket is closed")); + return; + } + + long now = System.currentTimeMillis(); + if (!session.online + && now - session.lastProgressMillis.get() + > HANDSHAKE_TIMEOUT_MILLIS) { + failSession(session.id, + new SocketException("ON4KST handshake timed out")); + 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")); + return; + } + + if (inboundIdle > APPLICATION_HEARTBEAT_AFTER_MILLIS + && session.lastHeartbeatMillis.get() + < session.lastInboundMillis.get()) { + sendHeartbeat(session); + } + } catch (RuntimeException exception) { + LOGGER.log(Level.WARNING, + "ON4KST connection monitor failed", exception); + } + } + + private void validateConfiguration() { + ChatPreferences preferences = controller.getChatPreferences(); + On4KstProtocol.login( + preferences.getStn_loginCallSign(), + preferences.getStn_loginPassword(), + preferences.getLoginChatCategoryMain().getCategoryNumber(), + "KST4Contest v" + ApplicationConstants.APPLICATION_CURRENT_VERSION, + 0L); + On4KstProtocol.locator(preferences.getStn_loginLocatorMainCat()); + if (preferences.getStn_loginNameMainCat() != null + && !preferences.getStn_loginNameMainCat().isBlank()) { + On4KstProtocol.field( + preferences.getStn_loginNameMainCat(), "main chat name"); + } + + if (preferences.isLoginToSecondChatEnabled()) { + if (preferences.getLoginChatCategorySecond() == null) { + throw new IllegalArgumentException("Second chat has no category"); + } + On4KstProtocol.category( + preferences.getLoginChatCategorySecond().getCategoryNumber()); + if (preferences.getLoginChatCategorySecond().getCategoryNumber() + == preferences.getLoginChatCategoryMain().getCategoryNumber()) { + controller.onOn4KstConnectionWarning( + "Second ON4KST chat equals the main chat and will not be added twice."); + } + if (preferences.getStn_loginNameSecondCat() != null + && !preferences.getStn_loginNameSecondCat().isBlank()) { + On4KstProtocol.field( + preferences.getStn_loginNameSecondCat(), "second chat name"); + } + } + } + + private boolean hasDistinctSecondChat(ChatPreferences preferences) { + return preferences.isLoginToSecondChatEnabled() + && preferences.getLoginChatCategorySecond() != null + && preferences.getLoginChatCategorySecond().getCategoryNumber() + != preferences.getLoginChatCategoryMain().getCategoryNumber(); + } + + private void configureSocket(Socket socket) throws IOException { + socket.setTcpNoDelay(true); + socket.setKeepAlive(true); + + try { + socket.setOption(ExtendedSocketOptions.TCP_KEEPIDLE, 45); + socket.setOption(ExtendedSocketOptions.TCP_KEEPINTERVAL, 15); + socket.setOption(ExtendedSocketOptions.TCP_KEEPCOUNT, 3); + } catch (UnsupportedOperationException | IOException exception) { + LOGGER.log(Level.INFO, + "Platform does not support configurable TCP keepalive; " + + "application heartbeat remains active", exception); + } + } + + private void closeSession(Session session) { + if (session == null) { + return; + } + LOGGER.log(Level.FINE, + "Closing ON4KST session {0}", session.id); + session.closed = true; + if (session.readThread != null) { + session.readThread.interrupt(); + } + if (session.writeThread != null) { + session.writeThread.interrupt(); + } + if (session.messageProcessor != null) { + session.messageProcessor.interrupt(); + } + try { + session.socket.close(); + } catch (IOException exception) { + LOGGER.log(Level.FINE, "Error closing obsolete ON4KST socket", exception); + } + } + + private void transition( + On4KstConnectionState newState, + String detail, + boolean critical + ) { + On4KstConnectionState previousState = state; + state = newState; + Level level = critical + || newState == On4KstConnectionState.DISCONNECTED + || newState == On4KstConnectionState.RECONNECT_WAIT + ? Level.WARNING : Level.INFO; + LOGGER.log(level, + "ON4KST state {0} -> {1}; detail: {2}", + new Object[] {previousState, newState, detail}); + controller.updateOn4KstConnectionState(newState, detail, critical); + } + + private void recordHistoryTimestamp(String line) { + long timestamp = parseMessageTimestamp(line); + if (timestamp > 0L) { + lastReceivedMessageTimestamp.accumulateAndGet(timestamp, Math::max); + } + } + + static long parseMessageTimestamp(String line) { + String[] fields = line == null ? new String[0] : line.split("\\|", -1); + if (fields.length < 3) { + return 0L; + } + + try { + long numeric = Long.parseLong(fields[2]); + long now = System.currentTimeMillis() / 1_000L; + if (numeric > 0L && numeric <= now + 86_400L) { + return numeric; + } + } catch (NumberFormatException ignored) { + return 0L; + } + + try { + return LocalDateTime.parse(fields[2], LIVE_MESSAGE_TIMESTAMP) + .toEpochSecond(ZoneOffset.UTC); + } catch (DateTimeParseException ignored) { + return 0L; + } + } + + 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"; + } + String message = failure.getMessage(); + return message == null || message.isBlank() + ? failure.getClass().getSimpleName() : message; + } + + private static final class Session { + private final Set completedInitialUserLists = + ConcurrentHashMap.newKeySet(); + + private final long id; + private final Socket socket; + private final LinkedBlockingQueue receiveQueue; + private final LinkedBlockingQueue transmitQueue; + 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 Map> initialMembers = + new ConcurrentHashMap<>(); + + private volatile ReadThread readThread; + private volatile WriteThread writeThread; + private volatile MessageBusManagementThread messageProcessor; + private volatile boolean loginSent; + private volatile boolean authenticated; + private volatile boolean mainListComplete; + private volatile boolean secondListComplete; + private volatile boolean online; + private volatile boolean closed; + + private Session( + long id, + Socket socket, + LinkedBlockingQueue receiveQueue, + LinkedBlockingQueue transmitQueue + ) { + this.id = id; + this.socket = socket; + this.receiveQueue = receiveQueue; + this.transmitQueue = transmitQueue; + } + } +} \ No newline at end of file diff --git a/src/main/java/kst4contest/controller/On4KstConnectionState.java b/src/main/java/kst4contest/controller/On4KstConnectionState.java new file mode 100644 index 0000000..d4798aa --- /dev/null +++ b/src/main/java/kst4contest/controller/On4KstConnectionState.java @@ -0,0 +1,50 @@ +package kst4contest.controller; + +/** + * Observable lifecycle of the ON4KST TCP session. + * + *

A connected TCP socket is deliberately not synonymous with an authenticated + * chat session. The intermediate states make that distinction visible to the UI + * and prevent application messages from being sent in the wrong protocol context.

+ */ +public enum On4KstConnectionState { + DISCONNECTED, + CONNECTING, + WAITING_FOR_LOGIN_PROMPT, + AUTHENTICATING, + SYNCING_MAIN_CHAT, + SYNCING_SECOND_CHAT, + ONLINE, + RECONNECT_WAIT, + STOPPING; + + /** + * Returns whether the complete application-level ON4KST handshake has finished. + * + * @return {@code true} only after authentication and all requested user lists + * have been synchronized + */ + public boolean isOnline() { + return this == ONLINE; + } + + /** + * Returns whether a connection attempt or usable session is currently owned by + * the connection manager. + * + *

This is intentionally broader than {@link #isOnline()}. The UI uses it to + * prevent a second Connect action while authentication, synchronization or a + * scheduled reconnect is already in progress.

+ * + * @return {@code true} while connecting, synchronizing, online or waiting for + * an automatic reconnect + */ + public boolean isConnectionAttemptActive() { + return switch (this) { + case CONNECTING, WAITING_FOR_LOGIN_PROMPT, AUTHENTICATING, + SYNCING_MAIN_CHAT, SYNCING_SECOND_CHAT, ONLINE, + RECONNECT_WAIT -> true; + default -> false; + }; + } +} \ No newline at end of file diff --git a/src/main/java/kst4contest/controller/On4KstProtocol.java b/src/main/java/kst4contest/controller/On4KstProtocol.java new file mode 100644 index 0000000..932575c --- /dev/null +++ b/src/main/java/kst4contest/controller/On4KstProtocol.java @@ -0,0 +1,192 @@ +package kst4contest.controller; + +import java.util.Locale; +import java.util.regex.Pattern; + +/** + * Builds ON4KST port-23001 frames and rejects values that could break framing or + * put the server into an invalid chat context. + * + *

All outbound protocol construction is concentrated here. User-controlled + * values may therefore never introduce a field separator or a second line, and + * category and locator validation happens before the frame reaches the socket.

+ */ +final class On4KstProtocol { + private static final Pattern LOCATOR_6 = + Pattern.compile("^[A-Ra-r]{2}[0-9]{2}[A-Xa-x]{2}$"); + + private On4KstProtocol() { + } + + /** + * Builds the initial authenticated login frame. + * + * @param callsign login callsign + * @param password ON4KST password; never logged by this class + * @param category primary chat category + * @param clientName client identification sent to the server + * @param lastMessageTimestamp earliest history timestamp to request + * @return validated frame without CR/LF terminator + */ + static String login( + String callsign, + String password, + int category, + String clientName, + long lastMessageTimestamp + ) { + return "LOGINC|" + field(callsign, "callsign") + + "|" + password(password) + + "|" + category(category) + + "|" + field(clientName, "client name") + + "|25|0|1|" + Math.max(0L, lastMessageTimestamp) + "|0|"; + } + + /** Builds the settings-complete frame for the supplied chat category. */ + static String settingsDone(int category) { + return "SDONE|" + category(category) + "|"; + } + + /** Builds the frame used to add a distinct second chat to the same session. */ + static String addChat(int category, long lastMessageTimestamp) { + return "ACHAT|" + category(category) + + "|25|10|2|" + Math.max(0L, lastMessageTimestamp) + + "|0|"; + } + + /** Builds a category-qualified locator command after validating Maidenhead syntax. */ + static String setLocator(int category, String locator) { + return command(category, "/SETLOC " + locator(locator)); + } + + /** Builds a category-qualified chat-name command. */ + static String setName(int category, String name) { + return command(category, "/SETNAME " + field(name, "chat name")); + } + + /** Builds the command that changes the operator state back to available. */ + static String back(int category) { + return command(category, "/BACK"); + } + + /** + * Wraps one validated slash command in an ON4KST message frame. + * + * @return frame without CR/LF terminator + */ + static String command(int category, String command) { + return "MSG|" + category(category) + "|0|" + + messageText(command) + "|0|"; + } + + /** + * Wraps one operator chat message in a category-qualified ON4KST frame. + * + * @return frame without CR/LF terminator + */ + static String chatMessage(int category, String text) { + return "MSG|" + category(category) + "|0|" + + messageText(text) + "|0|"; + } + + /** + * Removes trailing line terminators from a legacy raw frame while rejecting an + * embedded line break that could inject a second server command. + * + * @param frame legacy raw frame, possibly with trailing CR/LF + * @return exactly one normalized protocol line + * @throws IllegalArgumentException if the value is {@code null} or contains an + * embedded line break + */ + static String normalizeRawFrame(String frame) { + if (frame == null) { + throw new IllegalArgumentException("ON4KST frame must not be null"); + } + + int end = frame.length(); + while (end > 0) { + char last = frame.charAt(end - 1); + if (last != '\r' && last != '\n') { + break; + } + end--; + } + + String normalized = frame.substring(0, end); + if (normalized.indexOf('\r') >= 0 || normalized.indexOf('\n') >= 0) { + throw new IllegalArgumentException( + "ON4KST frame contains an embedded line break"); + } + return normalized; + } + + /** + * Validates and normalizes a six-character Maidenhead locator. + * + * @return upper-case locator + */ + static String locator(String locator) { + String normalized = field(locator, "locator").toUpperCase(Locale.ROOT); + if (!LOCATOR_6.matcher(normalized).matches()) { + throw new IllegalArgumentException( + "Locator must be a six-character Maidenhead locator: " + normalized); + } + return normalized; + } + + /** Rejects message text containing an ON4KST field or line delimiter. */ + static String messageText(String text) { + String value = field(text, "message text"); + if (value.indexOf('|') >= 0) { + throw new IllegalArgumentException( + "Message text contains the ON4KST field separator '|'"); + } + return value; + } + + /** + * Validates one required, non-password protocol field. + * + * @param value field value + * @param label diagnostic label used in validation errors + * @return trimmed value + */ + static String field(String value, String label) { + if (value == null || value.isBlank()) { + throw new IllegalArgumentException(label + " must not be empty"); + } + if (value.indexOf('|') >= 0 + || value.indexOf('\r') >= 0 + || value.indexOf('\n') >= 0) { + throw new IllegalArgumentException( + label + " contains an ON4KST frame delimiter"); + } + return value.trim(); + } + + private static String password(String value) { + if (value == null || value.isEmpty()) { + throw new IllegalArgumentException("password must not be empty"); + } + if (value.indexOf('|') >= 0 + || value.indexOf('\r') >= 0 + || value.indexOf('\n') >= 0) { + throw new IllegalArgumentException( + "password contains an ON4KST frame delimiter"); + } + return value; + } + + /** + * Validates the category range supported by ON4KST. + * + * @return the unchanged category for convenient inline use + */ + static int category(int category) { + if (category < 1 || category > 12) { + throw new IllegalArgumentException( + "Unsupported ON4KST chat category: " + category); + } + return category; + } +} \ No newline at end of file diff --git a/src/main/java/kst4contest/controller/On4KstProtocolTest.java b/src/main/java/kst4contest/controller/On4KstProtocolTest.java new file mode 100644 index 0000000..f42b121 --- /dev/null +++ b/src/main/java/kst4contest/controller/On4KstProtocolTest.java @@ -0,0 +1,69 @@ +package kst4contest.controller; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; + +import org.junit.jupiter.api.Test; + +import java.time.LocalDateTime; +import java.time.ZoneOffset; + +class On4KstProtocolTest { + @Test + void buildsLoginWithReplayOverlap() { + assertEquals( + "LOGINC|DL1ABC|secret|2|KST4Contest v1.2.3|25|0|1|12344|0|", + On4KstProtocol.login( + "DL1ABC", "secret", 2, + "KST4Contest v1.2.3", 12_344L)); + } + + @Test + void buildsContextSafeSecondChatFrames() { + assertEquals("SDONE|2|", On4KstProtocol.settingsDone(2)); + assertEquals("ACHAT|3|25|10|2|100|0|", + On4KstProtocol.addChat(3, 100L)); + assertEquals("MSG|2|0|/SETLOC JO31AA|0|", + On4KstProtocol.setLocator(2, "jo31aa")); + assertEquals("MSG|3|0|/SETNAME 10G 10368.200|0|", + On4KstProtocol.setName(3, "10G 10368.200")); + } + + @Test + void stripsOnlyTrailingLineEndings() { + assertEquals("CK|", On4KstProtocol.normalizeRawFrame("CK|\r\n")); + assertThrows(IllegalArgumentException.class, + () -> On4KstProtocol.normalizeRawFrame("CK|\rBROKEN")); + } + + @Test + void rejectsValuesThatCouldCreateASecondProtocolFrame() { + assertThrows(IllegalArgumentException.class, + () -> On4KstProtocol.chatMessage(2, "hello|0|")); + assertThrows(IllegalArgumentException.class, + () -> On4KstProtocol.chatMessage(2, "hello\r\nQUIT|")); + assertThrows(IllegalArgumentException.class, + () -> On4KstProtocol.login( + "DL1ABC", "bad|password", 2, "client", 0L)); + } + + @Test + void rejectsInvalidLocatorAndCategoryBeforeTheyReachTheServer() { + assertThrows(IllegalArgumentException.class, + () -> On4KstProtocol.setLocator(2, "JO31")); + assertThrows(IllegalArgumentException.class, + () -> On4KstProtocol.settingsDone(99)); + } + + @Test + void convertsBothHistoryAndLiveMessageTimestampsForReconnect() { + assertEquals(1_186_819_108L, + On4KstConnectionManager.parseMessageTimestamp( + "CR|2|1186819108|EA6VQ|Gabriel|0|msg|0|")); + assertEquals( + LocalDateTime.of(2026, 8, 13, 12, 34, 56) + .toEpochSecond(ZoneOffset.UTC), + On4KstConnectionManager.parseMessageTimestamp( + "CH|2|20260813123456|DL1ABC|Op|0|msg|0|")); + } +} \ No newline at end of file diff --git a/src/main/java/kst4contest/controller/On4KstSocketThreadTest.java b/src/main/java/kst4contest/controller/On4KstSocketThreadTest.java new file mode 100644 index 0000000..2c3e818 --- /dev/null +++ b/src/main/java/kst4contest/controller/On4KstSocketThreadTest.java @@ -0,0 +1,100 @@ +package kst4contest.controller; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +import java.io.BufferedReader; +import java.io.OutputStreamWriter; +import java.io.InputStreamReader; +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 On4KstSocketThreadTest { + @Test + @Timeout(5) + void eofIsReportedImmediately() throws Exception { + try (ServerSocket server = new ServerSocket(0)) { + CompletableFuture serverDone = CompletableFuture.runAsync(() -> { + try (Socket accepted = server.accept(); + OutputStreamWriter out = new OutputStreamWriter( + accepted.getOutputStream(), StandardCharsets.UTF_8)) { + out.write("CK|\r\n"); + out.flush(); + } catch (Exception exception) { + throw new RuntimeException(exception); + } + }); + + try (Socket client = new Socket("127.0.0.1", server.getLocalPort())) { + LinkedBlockingQueue queue = new LinkedBlockingQueue<>(); + AtomicBoolean active = new AtomicBoolean(true); + CompletableFuture failure = new CompletableFuture<>(); + ReadThread reader = new ReadThread( + 7L, client, queue, ignored -> active.get(), ignored -> { }, + failure::complete); + reader.start(); + + assertEquals("CK|", queue.poll(2, TimeUnit.SECONDS).getMessageText()); + failure.get(2, TimeUnit.SECONDS); + active.set(false); + reader.join(Duration.ofSeconds(2).toMillis()); + } + serverDone.get(2, TimeUnit.SECONDS); + } + } + + @Test + @Timeout(5) + void writerUsesOneExactCrLfPerFrameIncludingHeartbeat() throws Exception { + try (ServerSocket server = new ServerSocket(0)) { + CompletableFuture firstLine = new CompletableFuture<>(); + CompletableFuture secondLine = new CompletableFuture<>(); + CompletableFuture serverDone = CompletableFuture.runAsync(() -> { + try (Socket accepted = server.accept(); + BufferedReader in = new BufferedReader(new InputStreamReader( + accepted.getInputStream(), StandardCharsets.UTF_8))) { + firstLine.complete(in.readLine()); + secondLine.complete(in.readLine()); + } catch (Exception exception) { + throw new RuntimeException(exception); + } + }); + + try (Socket client = new Socket("127.0.0.1", server.getLocalPort())) { + LinkedBlockingQueue 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("SDONE|2|\r")); + assertEquals("", firstLine.get(2, TimeUnit.SECONDS)); + assertEquals("SDONE|2|", secondLine.get(2, TimeUnit.SECONDS)); + + active.set(false); + writer.interrupt(); + writer.join(Duration.ofSeconds(2).toMillis()); + } + serverDone.get(2, TimeUnit.SECONDS); + } + } + + private ChatMessage serverFrame(String text) { + ChatMessage message = new ChatMessage(); + message.setMessageDirectedToServer(true); + message.setMessageText(text); + return message; + } +} \ No newline at end of file diff --git a/src/main/java/kst4contest/controller/ReadThread.java b/src/main/java/kst4contest/controller/ReadThread.java index f26dd2d..87e5bef 100644 --- a/src/main/java/kst4contest/controller/ReadThread.java +++ b/src/main/java/kst4contest/controller/ReadThread.java @@ -1,119 +1,130 @@ package kst4contest.controller; -import java.io.*; -import java.net.*; +import java.io.BufferedReader; +import java.io.EOFException; +import java.io.IOException; +import java.io.InputStreamReader; +import java.net.Socket; import java.nio.charset.StandardCharsets; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.function.Consumer; +import java.util.function.LongPredicate; import java.util.logging.Level; import java.util.logging.Logger; import kst4contest.model.ChatMessage; - + /** - * This thread is responsible for reading telnet servers input at port 23001 and printing it - * to the console. - * It runs in an infinite loop until the client disconnects from the server. + * Reads exactly one immutable ON4KST connection session. * - * @author www.codejava.net + *

EOF is a connection-loss event, not an empty chat message. Every line is + * associated with the session id captured by this reader, so a delayed exception + * from an obsolete socket cannot affect a newer reconnect.

*/ public class ReadThread extends Thread { private static final Logger LOGGER = Logger.getLogger(ReadThread.class.getName()); - private BufferedReader reader; - private Socket socket; - private ChatController client; - public boolean accidentalDisconnected; - - - - public boolean isAccidentalDisconnected() { - return accidentalDisconnected; - } - public void setAccidentalDisconnected(boolean accidentalDisconnected) { - this.accidentalDisconnected = accidentalDisconnected; - } + private final long sessionId; + private final Socket socket; + private final LinkedBlockingQueue receiveQueue; + private final LongPredicate sessionIsActive; + private final Consumer inboundActivity; + private final Consumer connectionFailure; + private final BufferedReader reader; - // private boolean readingFinished = true; //kst4contest.test 4 23001 - private boolean readingFinished = true; - - InputStream input; - - public ReadThread(Socket socket, ChatController client) { + /** + * Compatibility constructor for the pre-session controller path. + * + * @deprecated new connections should be created by + * {@link On4KstConnectionManager} + */ + @Deprecated + public ReadThread(Socket socket, ChatController client) throws IOException { + this(0L, socket, client.getMessageRXBus(), ignored -> true, + ignored -> { }, ignored -> { }); + } + + /** + * Creates the reader for one connection generation. + * + * @param sessionId immutable id of the owning socket session + * @param socket connected ON4KST socket + * @param receiveQueue private receive queue belonging to this session + * @param sessionIsActive guard against callbacks from an obsolete session + * @param inboundActivity callback used for liveness and protocol progress + * @param connectionFailure callback for EOF, I/O and unexpected runtime errors + * @throws IOException if the socket input stream cannot be opened + */ + public ReadThread( + long sessionId, + Socket socket, + LinkedBlockingQueue receiveQueue, + LongPredicate sessionIsActive, + Consumer inboundActivity, + Consumer connectionFailure + ) throws IOException { + this.sessionId = sessionId; this.socket = socket; - this.client = client; - - try { - input = socket.getInputStream(); - reader = new BufferedReader(new InputStreamReader(input, StandardCharsets.UTF_8)); - - } catch (IOException ex) { - LOGGER.log(Level.SEVERE, "Error getting input stream", ex); - } + this.receiveQueue = receiveQueue; + this.sessionIsActive = sessionIsActive; + this.inboundActivity = inboundActivity; + this.connectionFailure = connectionFailure; + this.reader = new BufferedReader(new InputStreamReader( + socket.getInputStream(), StandardCharsets.UTF_8)); } - + + @Override public void run() { - Thread.currentThread().setName("ReadFromTelnetThread"); - - ChatMessage message; //bugfix leak, moved out of while - while (true) { - -// System.out.println("rdth"); - - try { - - String response = reader.readLine(); - message = new ChatMessage(); - message.setMessageText(response); - -// message.setDirectedToServer(false); -// message.setDirectedToServer(false); -// message.setDirectedToServer(false); - - - if (response != null) { - client.getMessageRXBus().put(message); -// System.out.println("[RT]: read message and added it to msgrxqueue --- " + response + " ---"); - } else { - System.out.println("[RT]: read message responsed a nullstring, do nothing, buffersize = " + socket.getReceiveBufferSize() + ", reader ready? " - + reader.ready()); -// reader = new BufferedReader(new InputStreamReader(input)); -// response = reader.readLine(); - this.client.getSocket().close(); - this.interrupt(); + Thread.currentThread().setName("ReadFromOn4Kst-" + sessionId); + LOGGER.log(Level.FINE, + "ON4KST reader started for session {0}", sessionId); + try { + while (!isInterrupted() && sessionIsActive.test(sessionId)) { + String response = reader.readLine(); + if (response == null) { + throw new EOFException("ON4KST closed the TCP connection"); } - - } - catch (Exception sexc) { - LOGGER.log(Level.SEVERE, "[ReadThread] Socket closed unexpectedly", sexc); - try { - this.client.getSocket().close(); - this.interrupt(); - break; - } catch (IOException e) { - LOGGER.log(Level.SEVERE, "[ReadThread] Error closing socket", e); - } + inboundActivity.accept(response); + if (!sessionIsActive.test(sessionId)) { + break; + } + ChatMessage message = new ChatMessage(); + message.setMessageText(response); + receiveQueue.put(message); } - + } catch (InterruptedException interrupted) { + Thread.currentThread().interrupt(); + } catch (IOException exception) { + if (sessionIsActive.test(sessionId)) { + LOGGER.log(Level.FINE, + "ON4KST read failed for session " + sessionId, + exception); + connectionFailure.accept(exception); + } + } catch (RuntimeException exception) { + if (sessionIsActive.test(sessionId)) { + LOGGER.log(Level.SEVERE, "Unexpected ON4KST reader failure", exception); + connectionFailure.accept(exception); + } + } finally { + LOGGER.log(Level.FINE, + "ON4KST reader stopped for session {0}", sessionId); } } - + + /** + * Interrupts the read loop and closes the session socket. + * + * @return always {@code true} after a successful close + * @throws IOException if closing the reader or socket fails + */ public boolean terminateConnection() throws IOException { - this.reader.close(); - this.input.close(); - this.socket.close(); - - return true; + interrupt(); + reader.close(); + socket.close(); + return true; } - - public boolean isReadingFinished() { - return readingFinished; - } - - public void setReadingFinished(boolean readingReady) { - this.readingFinished = readingReady; - } - - } \ No newline at end of file diff --git a/src/main/java/kst4contest/controller/StatusUpdateListener.java b/src/main/java/kst4contest/controller/StatusUpdateListener.java index b866203..45ba4cf 100644 --- a/src/main/java/kst4contest/controller/StatusUpdateListener.java +++ b/src/main/java/kst4contest/controller/StatusUpdateListener.java @@ -17,4 +17,20 @@ public interface StatusUpdateListener { void onUserListUpdated(String reason); // new: userlist-update -} + /** + * Called whenever the authoritative ON4KST session changes lifecycle state. + * + *

The callback may originate from a background connection supervisor. A UI + * implementation must marshal control changes onto its application thread.

+ * + * @param state new connection, authentication or synchronization state + * @param detail human-readable progress or failure reason + */ + default void onConnectionStateChanged( + On4KstConnectionState state, + String detail + ) { + // Optional for non-UI listeners. + } + +} \ No newline at end of file diff --git a/src/main/java/kst4contest/controller/WriteThread.java b/src/main/java/kst4contest/controller/WriteThread.java index ec76443..8f583c2 100644 --- a/src/main/java/kst4contest/controller/WriteThread.java +++ b/src/main/java/kst4contest/controller/WriteThread.java @@ -1,289 +1,173 @@ package kst4contest.controller; -import java.io.*; -import java.net.*; -import java.nio.charset.Charset; +import java.io.BufferedWriter; +import java.io.IOException; +import java.io.OutputStreamWriter; +import java.net.Socket; import java.nio.charset.StandardCharsets; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.function.Consumer; +import java.util.function.LongPredicate; +import java.util.logging.Level; +import java.util.logging.Logger; import kst4contest.ApplicationConstants; +import kst4contest.model.ChatCategory; import kst4contest.model.ChatMessage; /** - * This thread is responsible for sending content to the chat. As we only use - * the tx function, there is no content in run() method - * + * Serializes and writes exactly one immutable ON4KST connection session. * + *

The writer owns one private queue and appends exactly one CR/LF terminator + * per frame. All category selection and delimiter validation happens before bytes + * reach the socket.

*/ public class WriteThread extends Thread { - private PrintWriter writer; - private Socket socket; - private ChatController client; - private OutputStream output; - - private ChatMessage messageToBeSend; - - public WriteThread(Socket socket, ChatController client) throws InterruptedException { - this.socket = socket; - this.client = client; - - try { - output = socket.getOutputStream(); - - writer = new PrintWriter(output, true, StandardCharsets.UTF_8); - - } catch (IOException ex) { - System.out.println("Error getting output stream: " + ex.getMessage()); - ex.printStackTrace(); - } - } + private static final Logger LOGGER = + Logger.getLogger(WriteThread.class.getName()); + private final long sessionId; + private final Socket socket; + private final LinkedBlockingQueue transmitQueue; + private final LongPredicate sessionIsActive; + private final Consumer connectionFailure; + private final Consumer rejectedFrame; + private final BufferedWriter writer; + private final int defaultCategory; /** - * This method is used to send a message to the server, raw formatted. E.g. for - * the keepalive message. This method sends only in the main message-Category. To send it in a category - * "defined by Chatmessage", use txByRxmsgCatOrigin(Chatmessage "toBeSend") - * - * @param messageToServer - * @throws InterruptedException - */ - public void tx(ChatMessage messageToServer) throws InterruptedException { - -// writer.println(messageToServer.getMessage()); //kst4contest.test 4 23001 -// writer.flush(); //kst4contest.test 4 23001 - System.out.println(messageToServer.getMessageText() + "< sended to the writer"); - writer.println(messageToServer.getMessageText()); - - } - - - /** - * This method is used to send a message directly to a receiver in a special chatcategory. The receivers category - * will be read out of the Chatmessage.getChatCategory method. The message text will be modified to fit kst - * messageformat + * Compatibility constructor for the pre-session controller path. * - * @param messageToServer - * @throws InterruptedException + * @deprecated new connections should be created by + * {@link On4KstConnectionManager} */ - public void txByRxmsgCatOrigin(ChatMessage messageToServer) throws InterruptedException { - -// writer.println(messageToServer.getMessage()); //kst4contest.test 4 23001 -// writer.flush(); //kst4contest.test 4 23001 - - String originalMessageText = messageToServer.getMessageText() + ""; - - String newMessageText = ""; - - newMessageText = ("MSG|" + messageToServer.getChatCategory().getCategoryNumber() - + "|0|" + originalMessageText + "|0|"); //original before 1.26 - - - System.out.println(newMessageText + "< sended to the writer (DIRECTED REPLY)"); - writer.println(newMessageText); - + @Deprecated + public WriteThread(Socket socket, ChatController client) throws IOException { + this(0L, socket, client.getMessageTXBus(), + client.getChatPreferences().getLoginChatCategoryMain().getCategoryNumber(), + ignored -> true, + ignored -> { }, System.out::println); } /** - * This method gets a textmessage to the chat and adds some characters to hit - * the neccessarry format to send a message in the on4kst chat either to another - * station or to the public. - * - * @param messageToServer - * @throws InterruptedException + * Creates the writer for one connection generation. + * + * @param sessionId immutable id of the owning socket session + * @param socket connected ON4KST socket + * @param transmitQueue private transmit queue belonging to this session + * @param defaultCategory fallback category for unqualified chat messages + * @param sessionIsActive guard against writes from an obsolete session + * @param connectionFailure callback for socket and unexpected runtime errors + * @param rejectedFrame callback for locally rejected protocol content + * @throws IOException if the socket output stream cannot be opened */ - public void txKSTFormatted(ChatMessage messageToServer) throws InterruptedException { + public WriteThread( + long sessionId, + Socket socket, + LinkedBlockingQueue transmitQueue, + int defaultCategory, + LongPredicate sessionIsActive, + Consumer connectionFailure, + Consumer rejectedFrame + ) throws IOException { + this.sessionId = sessionId; + this.socket = socket; + this.transmitQueue = transmitQueue; + this.defaultCategory = defaultCategory; + this.sessionIsActive = sessionIsActive; + this.connectionFailure = connectionFailure; + this.rejectedFrame = rejectedFrame; + this.writer = new BufferedWriter(new OutputStreamWriter( + socket.getOutputStream(), StandardCharsets.UTF_8)); + } -// writer.println(messageToServer.getMessageText()); - messageToBeSend = messageToServer; + @Override + public void run() { + Thread.currentThread().setName("WriteToOn4Kst-" + sessionId); + LOGGER.log(Level.FINE, + "ON4KST writer started for session {0}", sessionId); try { - - messageToBeSend = client.getMessageTXBus().take(); -// this.client.getmesetChatsetServerready(true); - - } catch (InterruptedException e) { - e.printStackTrace(); - } - - String messageLine = messageToBeSend.getMessageText(); - - if (messageToBeSend.isMessageDirectedToServer()) { - /** - * We have to check if we only commands the server (keepalive) or want do talk - * to the community - */ - - try { - tx(messageToBeSend); - System.out.println("BUS: tx: " + messageToBeSend.getMessageText()); - - } catch (InterruptedException e) { - // TODO Auto-generated catch block - e.printStackTrace(); - } - - } else { - - ChatMessage ownMSG = new ChatMessage(); - -// ownMSG.setMessageText( -// "MSG|" + this.client.getCategory().getCategoryNumber() + "|0|" + messageLine + "|0|"); - - ownMSG.setMessageText("MSG|" + this.client.getChatPreferences().getLoginChatCategoryMain().getCategoryNumber() - + "|0|" + messageLine + "|0|"); //original before 1.26 - - try { - tx(ownMSG); - System.out.println("BUS: tx: " + ownMSG.getMessageText()); - - } catch (InterruptedException e) { - // TODO Auto-generated catch block - e.printStackTrace(); - } - } - - if (messageToBeSend.equals("/QUIT")) { - try { - this.client.getReadThread().terminateConnection(); - this.client.getReadThread().interrupt(); - this.client.getWriteThread().terminateConnection(); - this.client.getWriteThread().interrupt(); - this.interrupt(); - - } catch (IOException e) { - e.printStackTrace(); - } - } - - } - - public boolean terminateConnection() throws IOException { - - this.output.close(); - this.socket.close(); - - return true; - } - - public void run() { - Thread.currentThread().setName("WriteToTelnetThread"); - - while (true) { - try { - messageToBeSend = client.getMessageTXBus().take(); - - if (messageToBeSend.getMessageText().equals(ApplicationConstants.DISCONNECT_RDR_POISONPILL) - && messageToBeSend.getMessageSenderName().equals(ApplicationConstants.DISCONNECT_RDR_POISONPILL)) { - client.getMessageRXBus().clear(); - this.interrupt(); + while (!isInterrupted() && sessionIsActive.test(sessionId)) { + ChatMessage message = transmitQueue.take(); + if (isPoisonPill(message)) { + break; + } + if (!sessionIsActive.test(sessionId)) { break; - } else { - String messageLine = messageToBeSend.getMessageText(); - - if (messageToBeSend.isMessageDirectedToServer()) { - /** - * We have to check if we only commands the server (keepalive) or want do talk - * to the community - */ - - try { - tx(messageToBeSend); - System.out.println("BUS: tx: " + messageToBeSend.getMessageText()); - - } catch (InterruptedException e) { - // TODO Auto-generated catch block - e.printStackTrace(); - } - - } else { //message is not directed to the server, it´s directed to all or to a station - - if (messageToBeSend.getChatCategory() == this.client.getChatCategoryMain() || messageToBeSend.getChatCategory() == this.client.getChatCategorySecondChat()) { - - txByRxmsgCatOrigin(messageToBeSend); - - } else { //default bhv if destination cat is not detectable - - - ChatMessage ownMSG = new ChatMessage(); - - ownMSG.setMessageText( - "MSG|" + this.client.getChatPreferences().getLoginChatCategoryMain().getCategoryNumber() + "|0|" - + messageLine + "|0|"); - - try { - tx(ownMSG); - System.out.println("WT: tx (raw): " + ownMSG.getMessageText()); - - } catch (InterruptedException e) { - e.printStackTrace(); - } - } - } } - System.out.println("WritheTh: got message out of the queue: " + messageToBeSend.getMessageText()); - -// this.client.getmesetChatsetServerready(true); - - } catch (InterruptedException e) { - e.printStackTrace(); - client.getMessageTXBus().clear(); - } - -// String messageLine = messageTextRaw.getMessageText(); -// -// if (messageTextRaw.isMessageDirectedToServer()) { -// /** -// * We have to check if we only commands the server (keepalive) or want do talk -// * to the community -// */ -// -// try { -// tx(messageTextRaw); -// System.out.println("BUS: tx: " + messageTextRaw.getMessageText()); -// -// } catch (InterruptedException e) { -// // TODO Auto-generated catch block -// e.printStackTrace(); -// } -// -// } else { -// -// ChatMessage ownMSG = new ChatMessage(); -// -//// ownMSG.setMessageText( -//// "MSG|" + this.client.getCategory().getCategoryNumber() + "|0|" + messageLine + "|0|"); -// -// ownMSG.setMessageText( -// "MSG|" + this.client.getChatPreferences().getLoginChatCategory().getCategoryNumber() + "|0|" -// + messageLine + "|0|"); -// -// try { -// tx(ownMSG); -// System.out.println("BUS: tx: " + ownMSG.getMessageText()); -// -// } catch (InterruptedException e) { -// // TODO Auto-generated catch block -// e.printStackTrace(); -// } -// } + try { + writeFrame(formatFrame(message)); + } catch (IllegalArgumentException invalidFrame) { + LOGGER.log(Level.FINE, + "Rejected outbound ON4KST frame in session {0}: {1}", + new Object[] {sessionId, invalidFrame.getMessage()}); + rejectedFrame.accept(invalidFrame.getMessage()); + } } -// if (messageTextRaw.equals("/QUIT")) { -// try { -// this.client.getReadThread().terminateConnection(); -// this.client.getReadThread().interrupt(); -// this.client.getWriteThread().terminateConnection(); -// this.client.getWriteThread().interrupt(); -// this.interrupt(); -// -// } catch (IOException e) { -// // TODO Auto-generated catch block -// e.printStackTrace(); -// } -// } + } catch (InterruptedException interrupted) { + Thread.currentThread().interrupt(); + } catch (IOException exception) { + if (sessionIsActive.test(sessionId)) { + LOGGER.log(Level.FINE, + "ON4KST write failed for session " + sessionId, + exception); + connectionFailure.accept(exception); + } + } catch (RuntimeException exception) { + if (sessionIsActive.test(sessionId)) { + LOGGER.log(Level.SEVERE, + "Unexpected ON4KST writer failure for session " + + sessionId, + exception); + connectionFailure.accept(exception); + } + } finally { + LOGGER.log(Level.FINE, + "ON4KST writer stopped for session {0}", sessionId); + } + } - -// while (true) { -// -// } + private String formatFrame(ChatMessage message) { + if (message == null) { + throw new IllegalArgumentException("Cannot send an empty ON4KST message"); + } + if (message.isMessageDirectedToServer()) { + return On4KstProtocol.normalizeRawFrame(message.getMessageText()); + } + + ChatCategory category = message.getChatCategory(); + return On4KstProtocol.chatMessage( + category == null ? defaultCategory : category.getCategoryNumber(), + message.getMessageText()); + } + + private void writeFrame(String frame) throws IOException { + writer.write(frame); + writer.write("\r\n"); + writer.flush(); + } + + private boolean isPoisonPill(ChatMessage message) { + return message != null + && ApplicationConstants.DISCONNECT_RDR_POISONPILL.equals( + message.getMessageText()) + && ApplicationConstants.DISCONNECT_RDR_POISONPILL.equals( + message.getMessageSenderName()); + } + + /** + * Interrupts the write loop and closes the session socket. + * + * @return always {@code true} after a successful close + * @throws IOException if closing the writer or socket fails + */ + public boolean terminateConnection() throws IOException { + interrupt(); + writer.close(); + socket.close(); + return true; } } + diff --git a/src/main/java/kst4contest/controller/keepAliveMessageSenderTask.java b/src/main/java/kst4contest/controller/keepAliveMessageSenderTask.java index 974e92d..1396663 100644 --- a/src/main/java/kst4contest/controller/keepAliveMessageSenderTask.java +++ b/src/main/java/kst4contest/controller/keepAliveMessageSenderTask.java @@ -4,6 +4,13 @@ import java.util.TimerTask; import kst4contest.model.ChatMessage; +/** + * Enqueues the empty application-level reply expected by ON4KST keepalive + * handling. + * + *

The writer owns line termination and appends exactly one CR/LF sequence. + * Supplying an empty payload here avoids the historic double-terminator frame.

+ */ public class keepAliveMessageSenderTask extends TimerTask { private ChatController client; @@ -16,13 +23,14 @@ public class keepAliveMessageSenderTask extends TimerTask { @Override public void run() { - + Thread.currentThread().setName("KeepAliveMessageSenderTask"); - + // System.out.println("[keepalive: ] Thread runned now"); - + ChatMessage keepAliveMSG = new ChatMessage(); - keepAliveMSG.setMessageText("\r"); + // WriteThread appends exactly one CRLF. An empty frame is the protocol reply. + keepAliveMSG.setMessageText(""); keepAliveMSG.setMessageDirectedToServer(true); // System.out.println(new Utils4KST().time_generateCurrentMMDDhhmmTimeString() + " [keepaliveTask]: Sending keepalive: " @@ -33,4 +41,4 @@ public class keepAliveMessageSenderTask extends TimerTask { this.client.getMessageTXBus().add(keepAliveMSG); } -} +} \ No newline at end of file diff --git a/src/main/java/kst4contest/view/Kst4ContestApplication.java b/src/main/java/kst4contest/view/Kst4ContestApplication.java index d619f56..b5dca5a 100644 --- a/src/main/java/kst4contest/view/Kst4ContestApplication.java +++ b/src/main/java/kst4contest/view/Kst4ContestApplication.java @@ -1,5 +1,6 @@ package kst4contest.view; import kst4contest.utils.VersionUtils; +import kst4contest.controller.On4KstConnectionState; import javafx.scene.image.Image; import java.io.File; @@ -81,6 +82,10 @@ public class Kst4ContestApplication extends Application implements StatusUpdateL // Enables optional color highlighting of the QRA/grid cell. // The status text itself remains visible independently of this flag. + + private static final Logger LOGGER = Logger.getLogger( + Kst4ContestApplication.class.getName()); + private boolean gridSquareHighlightEnabled = false; @@ -113,6 +118,11 @@ public class Kst4ContestApplication extends Application implements StatusUpdateL private StationMapView stationMapView; //view class for the avl stn map private StationMapBridge stationMapBridge; //bridge for mapping actions between map and view + private final Button btnConnectionStateIndicator = new Button("LINK"); + private final Tooltip tipConnectionStateIndicator = new Tooltip(); + private On4KstConnectionState lastDisplayedConnectionState; + private String lastDisplayedConnectionDetail = ""; + private final Button btnBandUpgradeIndicator = new Button("BAND+"); private final Tooltip tipBandUpgradeIndicator = new Tooltip(); private Timeline bandUpgradeBlinkTimeline; @@ -5320,8 +5330,8 @@ public class Kst4ContestApplication extends Application implements StatusUpdateL menuItemOptionsAwayBack.setDisable(false); menuItemOptionsSetFrequencyAsName.setDisable(false); - chatcontroller.setConnectedAndLoggedIn(true); - chatcontroller.setDisconnected(false); +// chatcontroller.setConnectedAndLoggedIn(true); +// chatcontroller.setDisconnected(false); } catch (InterruptedException | IOException e) { e.printStackTrace(); @@ -5643,6 +5653,121 @@ public class Kst4ContestApplication extends Application implements StatusUpdateL btnSkedWarnIndicator.setTooltip(tipSkedWarnIndicator); } + private void initConnectionStateIndicatorButton() { + LOGGER.fine("Initializing compact ON4KST connection-state indicator"); + + btnConnectionStateIndicator.setMouseTransparent(true); + btnConnectionStateIndicator.setFocusTraversable(false); + btnConnectionStateIndicator.setMnemonicParsing(false); + btnConnectionStateIndicator.setMinSize(44, 22); + btnConnectionStateIndicator.setPrefSize(44, 22); + btnConnectionStateIndicator.setMaxSize(44, 22); + btnConnectionStateIndicator.setTooltip(tipConnectionStateIndicator); + btnConnectionStateIndicator.setAccessibleText( + "ON4KST connection state"); + FlowPane.setMargin( + btnConnectionStateIndicator, new Insets(2, 4, 2, 4)); + + updateConnectionStateIndicator( + chatcontroller.getOn4KstConnectionState(), + "No ON4KST connection"); + } + + private void updateConnectionStateIndicator( + On4KstConnectionState state, + String detail + ) { + if (!Platform.isFxApplicationThread()) { + LOGGER.warning( + "Connection indicator update arrived outside the JavaFX thread; " + + "rescheduling it safely"); + Platform.runLater(() -> updateConnectionStateIndicator(state, detail)); + return; + } + + On4KstConnectionState effectiveState = state == null + ? On4KstConnectionState.DISCONNECTED : state; + String stateDetail = detail == null || detail.isBlank() + ? effectiveState.name() : detail; + logConnectionIndicatorTransition(effectiveState, stateDetail); + + tipConnectionStateIndicator.setText( + "ON4KST link: " + effectiveState.name() + "\n" + stateDetail); + btnConnectionStateIndicator.setAccessibleHelp(stateDetail); + + String commonStyle = + "-fx-font-size: 10px;" + + "-fx-font-weight: bold;" + + "-fx-padding: 1 5 1 5;" + + "-fx-background-radius: 6;" + + "-fx-border-radius: 6;" + + "-fx-border-width: 2;"; + + switch (effectiveState) { + case ONLINE -> { + btnConnectionStateIndicator.setText("LINK"); + btnConnectionStateIndicator.setStyle( + commonStyle + + "-fx-background-color: #238636;" + + "-fx-border-color: #56d364;" + + "-fx-text-fill: white;" + + "-fx-effect: dropshadow(three-pass-box, " + + "rgba(35,134,54,0.55), 5, 0.25, 0, 0);"); + } + + case CONNECTING, WAITING_FOR_LOGIN_PROMPT, AUTHENTICATING, + SYNCING_MAIN_CHAT, SYNCING_SECOND_CHAT, STOPPING -> { + btnConnectionStateIndicator.setText("LINK…"); + btnConnectionStateIndicator.setStyle( + commonStyle + + "-fx-background-color: #ffb300;" + + "-fx-border-color: #ffe082;" + + "-fx-text-fill: #1b1b1b;" + + "-fx-effect: dropshadow(three-pass-box, " + + "rgba(255,179,0,0.65), 6, 0.3, 0, 0);"); + } + + case DISCONNECTED, RECONNECT_WAIT -> { + btnConnectionStateIndicator.setText("LINK!"); + btnConnectionStateIndicator.setStyle( + commonStyle + + "-fx-background-color: #d50000;" + + "-fx-border-color: #ff6b6b;" + + "-fx-text-fill: white;" + + "-fx-effect: dropshadow(three-pass-box, " + + "rgba(255,0,0,0.95), 10, 0.55, 0, 0);"); + } + } + } + + private void logConnectionIndicatorTransition( + On4KstConnectionState newState, + String newDetail + ) { + boolean stateChanged = newState != lastDisplayedConnectionState; + boolean detailChanged = !Objects.equals( + newDetail, lastDisplayedConnectionDetail); + if (!stateChanged && !detailChanged) { + return; + } + + On4KstConnectionState previousState = lastDisplayedConnectionState; + Level logLevel = switch (newState) { + case DISCONNECTED, RECONNECT_WAIT -> Level.WARNING; + default -> Level.INFO; + }; + + LOGGER.log(logLevel, + "ON4KST connection indicator: {0} -> {1}; detail: {2}", + new Object[] { + previousState == null ? "UNINITIALIZED" : previousState, + newState, + newDetail + }); + + lastDisplayedConnectionState = newState; + lastDisplayedConnectionDetail = newDetail; + } private void showBlinkingSkedWarnIndicator(String text) { @@ -6590,6 +6715,9 @@ public class Kst4ContestApplication extends Application implements StatusUpdateL flwpne_StatusBar.getChildren().add(mainScreenMenuBar); bPaneChatWindow.setTop(flwpne_StatusBar); + initConnectionStateIndicatorButton(); + flwpne_StatusBar.getChildren().add(btnConnectionStateIndicator); + initSkedWarnIndicatorButton(); chatcontroller.lastUiReminderEventProperty().addListener( @@ -6949,9 +7077,25 @@ public class Kst4ContestApplication extends Application implements StatusUpdateL } else { - chatState = "DISCONNECTED!"; + On4KstConnectionState connectionState = + chatcontroller.getOn4KstConnectionState(); + + chatState = switch (connectionState) { + case RECONNECT_WAIT -> + "CONNECTION LOST – reconnect scheduled"; + case CONNECTING -> + "Connecting to ON4KST…"; + case WAITING_FOR_LOGIN_PROMPT, AUTHENTICATING -> + "Connected – authenticating with ON4KST…"; + case SYNCING_MAIN_CHAT, SYNCING_SECOND_CHAT -> + "Connected – synchronizing ON4KST chat data…"; + default -> + "DISCONNECTED!"; + }; + chatcontroller.getChatPreferences().setChatState(chatState); } + if (chatcontroller.isDisconnected()) { chatState = "DISCONNECTED!"; chatcontroller.getChatPreferences().setChatState(chatState); @@ -11578,8 +11722,8 @@ public class Kst4ContestApplication extends Application implements StatusUpdateL txtFldstn_maxQRBDefault.setDisable(true); btnOptionspnlConnect.setDisable(true); btnOptionspnlDisconnect.setDisable(false); - chatcontroller.setConnectedAndLoggedIn(true); - chatcontroller.setDisconnected(false); +// chatcontroller.setConnectedAndLoggedIn(true); +// chatcontroller.setDisconnected(false); station_chkBxEnableSecondChat.setDisable(true); stn_choiceBxChatChategorySecond.setDisable(true); } @@ -11882,7 +12026,8 @@ public class Kst4ContestApplication extends Application implements StatusUpdateL String logDir = Path.of(System.getProperty("user.home"), ".praktiKST").toString(); new File(logDir).mkdirs(); FileHandler fileHandler = new FileHandler(logDir + "/kst4contest-errors.log", true); - fileHandler.setLevel(Level.SEVERE); + // Connection loss/reconnect diagnostics are warnings, not fatal crashes. + fileHandler.setLevel(Level.WARNING); fileHandler.setFormatter(new SimpleFormatter()); Logger rootLogger = Logger.getLogger(""); rootLogger.addHandler(fileHandler); @@ -11892,17 +12037,77 @@ public class Kst4ContestApplication extends Application implements StatusUpdateL } @Override - public void onThreadStatusChanged(String key, ThreadStateMessage threadStateMessage) { + public void onThreadStatusChanged( + String key, + ThreadStateMessage threadStateMessage + ) { + if (threadStateMessage == null) { + LOGGER.log(Level.WARNING, + "Ignoring empty thread-status update from source {0}", + key == null ? "UNKNOWN" : key); + return; + } + if (key == null || key.isBlank()) { + LOGGER.log(Level.WARNING, + "Ignoring thread-status update without a source key. Detail: {0}", + threadStateMessage.getRunningInformation()); + return; + } - Platform.runLater(() -> { - updateStatusButton(key, threadStateMessage); - }); + if ("ON4KST".equalsIgnoreCase(key)) { + String detail = threadStateMessage.getRunningInformation(); + Platform.runLater(() -> updateConnectionStateIndicator( + chatcontroller.getOn4KstConnectionState(), detail)); + return; + } + Platform.runLater(() -> updateStatusButton(key, threadStateMessage)); maybeShowBandUpgradeIndicator(key, threadStateMessage); - //if we receive a threadstatemessage for sked warning, enable the sked warning + } - } + public void onConnectionStateChanged( + On4KstConnectionState state, + String detail + ) { + On4KstConnectionState effectiveState = state; + if (effectiveState == null) { + LOGGER.log(Level.WARNING, + "Received ON4KST connection callback without a state; " + + "treating it as disconnected. Detail: {0}", + detail); + effectiveState = On4KstConnectionState.DISCONNECTED; + } + + On4KstConnectionState stateToDisplay = effectiveState; + Platform.runLater(() -> { + boolean active = stateToDisplay.isConnectionAttemptActive(); + boolean online = stateToDisplay.isOnline(); + updateConnectionStateIndicator(stateToDisplay, detail); + + if (menuItemFileConnect != null) { + menuItemFileConnect.setDisable(active); + } + if (menuItemFileDisconnect != null) { + menuItemFileDisconnect.setDisable(!active); + } + if (menuItemOptionsAwayBack != null) { + menuItemOptionsAwayBack.setDisable(!online); + } + if (menuItemOptionsSetFrequencyAsName != null) { + menuItemOptionsSetFrequencyAsName.setDisable(!online); + } + if (btnOptionspnlConnect != null) { + btnOptionspnlConnect.setDisable(active); + } + if (sendButton != null) { + sendButton.setDisable(!online); + } + if (txt_chatMessageUserInput != null) { + txt_chatMessageUserInput.setDisable(!online); + } + }); + } /** diff --git a/src/main/java/module-info.java b/src/main/java/module-info.java index 4fc4c9c..dbe7810 100644 --- a/src/main/java/module-info.java +++ b/src/main/java/module-info.java @@ -10,6 +10,7 @@ module praktiKST { requires java.net.http; requires java.desktop; requires jdk.crypto.ec; + requires jdk.net; requires org.junit.jupiter.api; requires org.mockito; exports kst4contest.controller.interfaces;