sec-beidou-rtcm/src/main/java/com/imdroid/sideslope/ntrip/RtkrcvConfigService.java

@@ -54,7 +54,8 @@ public class RtkrcvConfigService {
             replaced = replaceValueLine(replaced, "inpstr1-path", nz(profile.getInpstr1Path()));
             replaced = replaceValueLine(replaced, "inpstr2-path", nz(profile.getInpstr2Path()));
             replaced = replaceValueLine(replaced, "inpstr3-path", nz(profile.getInpstr3Path()));
-            replaced = replaceValueLine(replaced, "outstr1-path", nz(profile.getOutstr1Path()));
+            String resolvedOutPath = resolveOutPath(profile);
+            replaced = replaceValueLine(replaced, "outstr1-path", resolvedOutPath);
             // If local tcp endpoints are used (e.g., 127.0.0.1:port), force type to tcpcli
             if (looksLikeTcpEndpoint(profile.getInpstr1Path())) {
                 replaced = replaceValueLine(replaced, "inpstr1-type", "tcpcli");
@@ -62,7 +63,7 @@ public class RtkrcvConfigService {
                 replaced = replaceValueLine(replaced, "misc-timeout", "300000");
                 replaced = replaceValueLine(replaced, "misc-reconnect", "3000");
             }
-            if (looksLikeTcpEndpoint(profile.getOutstr1Path())) {
+            if (looksLikeTcpEndpoint(resolvedOutPath)) {
                 replaced = replaceValueLine(replaced, "outstr1-type", "tcpcli");
                 replaced = replaceValueLine(replaced, "misc-timeout", "300000");
                 replaced = replaceValueLine(replaced, "misc-reconnect", "3000");
@@ -106,6 +107,27 @@ public class RtkrcvConfigService {
         return s == null ? "" : s;
     }
 
+    private String resolveOutPath(RtkrcvProfile profile) {
+        String candidate = nz(profile.getOutstr1Path());
+        if (!candidate.isEmpty()) {
+            return candidate;
+        }
+        String inPath = nz(profile.getInpstr1Path());
+        if (looksLikeTcpEndpoint(inPath)) {
+            String[] parts = inPath.split(":", 2);
+            if (parts.length == 2) {
+                String host = parts[0].trim();
+                try {
+                    int inPort = Integer.parseInt(parts[1].trim());
+                    return host + ":" + (inPort + 1);
+                } catch (NumberFormatException ignore) {
+                    // fall through and return empty candidate
+                }
+            }
+        }
+        return candidate;
+    }
+
     private boolean looksLikeTcpEndpoint(String path) {
         if (path == null) return false;
         String p = path.trim();

sec-beidou-rtcm/src/main/java/com/imdroid/sideslope/rtkcluster/RtkClusterService.java

@@ -50,6 +50,7 @@ public class RtkClusterService implements ApplicationRunner {
     private RtkrcvConfigService configService;
 
     private final Map<String, DeviceEndpoint> endpoints = new ConcurrentHashMap<>();
+    private final Map<String, DeviceEndpoint> outEndpoints = new ConcurrentHashMap<>();
     private final Map<String, Process> processes = new ConcurrentHashMap<>();
 
     private final ExecutorService worker = Executors.newCachedThreadPool(r -> {
@@ -66,10 +67,11 @@ public class RtkClusterService implements ApplicationRunner {
             LOGGER.info("No rtkrcv_profile records found. RtkCluster idle.");
             return;
         }
-        int slot = 1;
+        int slot = 0;
         for (RtkrcvProfile profile : profiles) {
             try {
-                int port = basePort + slot++;
+                int port = basePort + slot;
+                slot += 2; // reserve a pair (IN port, OUT port)
                 bootstrapDevice(profile, port);
             } catch (Exception e) {
                 LOGGER.error("Bootstrap device {} failed: {}", profile.getDeviceId(), e.getMessage(), e);
@@ -81,13 +83,17 @@ public class RtkClusterService implements ApplicationRunner {
         String deviceId = profile.getDeviceId();
         // 1) Start endpoint server (if not exists)
         endpoints.computeIfAbsent(deviceId, k -> new DeviceEndpoint(workPort));
+        outEndpoints.computeIfAbsent(deviceId, k -> new DeviceEndpoint(workPort + 1, true));
         DeviceEndpoint ep = endpoints.get(deviceId);
-        ep.ensureStarted();
+        DeviceEndpoint outEp = outEndpoints.get(deviceId);
+        ep.ensureInMode();
+        outEp.ensureOutMode();
 
         // 2) Update profile inp/out to local cluster port
-        String localPath = "127.0.0.1:" + workPort;
+        String localInPath = "127.0.0.1:" + workPort;
-        profile.setInpstr1Path(localPath);
+        String localOutPath = "127.0.0.1:" + (workPort + 1);
-        profile.setOutstr1Path(localPath);
+        profile.setInpstr1Path(localInPath);
+        profile.setOutstr1Path(localOutPath);
         profileMapper.updateById(profile);
 
         // 3) Generate config and start rtkrcv
@@ -103,7 +109,7 @@ public class RtkClusterService implements ApplicationRunner {
         session.setUpdatedAt(LocalDateTime.now());
         sessionMapper.insert(session);
 
-        //startRtkrcv(deviceId, conf.getParent().toString(), conf.toString());
+        startRtkrcv(deviceId, conf.getParent().toString(), conf.toString());
     }
 
     private void startRtkrcv(String deviceId, String workDir, String confPath) {
@@ -174,6 +180,7 @@ public class RtkClusterService implements ApplicationRunner {
 
     static class DeviceEndpoint {
         private final int port;
+        private final boolean outMode;
         private final ExecutorService exec;
         private volatile boolean started = false;
         private volatile ServerSocket server;
@@ -182,7 +189,12 @@ public class RtkClusterService implements ApplicationRunner {
         private final java.util.concurrent.LinkedBlockingDeque<byte[]> rtcmQueue = new java.util.concurrent.LinkedBlockingDeque<>(1024);
 
         DeviceEndpoint(int port) {
+            this(port, false);
+        }
+
+        DeviceEndpoint(int port, boolean outMode) {
             this.port = port;
+            this.outMode = outMode;
             this.exec = Executors.newCachedThreadPool(r -> {
                 Thread t = new Thread(r, "rtkcluster-ep-" + this.port);
                 t.setDaemon(true);
@@ -193,11 +205,28 @@ public class RtkClusterService implements ApplicationRunner {
         synchronized void ensureStarted() {
             if (started) return;
             exec.submit(this::acceptLoop);
+            if (!outMode) {
                 exec.submit(this::dequeueLoop);
+            }
             started = true;
         }
 
+        void ensureInMode() {
+            if (outMode) {
+                throw new IllegalStateException("Endpoint is configured as OUT mode: " + port);
+            }
+            ensureStarted();
+        }
+
+        void ensureOutMode() {
+            if (!outMode) {
+                throw new IllegalStateException("Endpoint is configured as IN mode: " + port);
+            }
+            ensureStarted();
+        }
+
         void enqueueRtcm(byte[] data) {
+            if (outMode) return;
             if (data == null || data.length == 0) return;
             if (!rtcmQueue.offerLast(data)) {
                 // queue full: drop oldest to keep stream fresh
@@ -211,7 +240,7 @@ public class RtkClusterService implements ApplicationRunner {
                 ss.setReuseAddress(true);
                 ss.bind(new InetSocketAddress("127.0.0.1", port));
                 this.server = ss;
-                LOGGER.info("RtkCluster device endpoint listening on 127.0.0.1:{}", port);
+                LOGGER.info("RtkCluster {} endpoint listening on 127.0.0.1:{}", outMode ? "OUT" : "IN", port);
                 while (true) {
                     Socket s = ss.accept();
                     classifyConnection(s);
@@ -223,45 +252,41 @@ public class RtkClusterService implements ApplicationRunner {
 
         private void classifyConnection(Socket s) {
             exec.submit(() -> {
+                boolean assigned = false;
                 try {
-                    // short probe to classify role (increase to 1000ms to reduce misclassification)
-                    s.setSoTimeout(1000);
-                    InputStream in = s.getInputStream();
-                    byte[] probe = new byte[256];
-                    int n = 0;
-                    try { n = in.read(probe); } catch (IOException ignore) {}
-                    // restore to blocking mode for steady-state
-                    s.setSoTimeout(0);
-                    if (n > 0 && isLikelyText(probe, n)) {
-                        // OUT connection (NMEA etc.)
-                        closeQuietly(outConn);
                     s.setTcpNoDelay(true);
                     s.setKeepAlive(true);
-                        outConn = s;
+                    if (outMode) {
-                        LOGGER.debug("Endpoint {} OUT connected", port);
+                        Socket previous = outConn;
-                        pumpOut(outConn, probe, n);
+                        if (isSocketAlive(previous)) {
+                            LOGGER.info("Endpoint {} replacing existing OUT connection", port);
+                            closeQuietly(previous);
                         } else {
-                        // IN connection (RTCM sink)
+                            LOGGER.debug("Endpoint {} OUT connected", port);
-                        closeQuietly(inConn);
-                        s.setTcpNoDelay(true);
-                        s.setKeepAlive(true);
-                        inConn = s;
-                        LOGGER.debug("Endpoint {} IN connected", port);
                         }
-                } catch (IOException e) {
+                        outConn = s;
-                    LOGGER.warn("classifyConnection error: {}", e.getMessage());
+                        assigned = true;
-                    closeQuietly(s);
+                        pumpOut(s, null, 0);
-                }
+                        return;
-            });
                     }
 
-        private boolean isLikelyText(byte[] buf, int n) {
+                    Socket previous = inConn;
-            int printable = 0;
+                    if (isSocketAlive(previous)) {
-            for (int i = 0; i < n; i++) {
+                        LOGGER.info("Endpoint {} replacing existing IN connection", port);
-                int b = buf[i] & 0xFF;
+                        closeQuietly(previous);
-                if (b >= 32 && b <= 126) printable++;
+                    } else {
+                        LOGGER.debug("Endpoint {} IN connected", port);
                     }
-            return printable >= Math.max(1, n - 4); // tolerate some non-printables
+                    inConn = s;
+                    assigned = true;
+                } catch (IOException e) {
+                    LOGGER.warn("classifyConnection error: {}", e.getMessage());
+                } finally {
+                    if (!assigned) {
+                        closeQuietly(s);
+                    }
+                }
+            });
         }
 
         private void pumpOut(Socket s, byte[] firstBuf, int firstLen) {
@@ -270,7 +295,7 @@ public class RtkClusterService implements ApplicationRunner {
                     byte[] buf = new byte[2048];
                     int read;
                     // deliver first classified bytes if any
-                    if (firstLen > 0) {
+                    if (firstBuf != null && firstLen > 0) {
                         String preview = new String(firstBuf, 0, firstLen, StandardCharsets.US_ASCII).replaceAll("\n", "\\n");
                         LOGGER.info("[OUT:{}] {}", port, preview);
                     }
@@ -292,8 +317,7 @@ public class RtkClusterService implements ApplicationRunner {
             while (true) {
                 try {
                     Socket sink = inConn;
-                    if (sink == null || sink.isClosed()) {
+                    if (!isSocketAlive(sink)) {
-                        // avoid consuming queue when no sink, preventing starvation
                         Thread.sleep(20);
                         continue;
                     }
@@ -303,7 +327,12 @@ public class RtkClusterService implements ApplicationRunner {
                         os.write(data);
                         os.flush();
                     } catch (IOException e) {
-                        LOGGER.warn("Write RTCM failed on {}: {}", port, e.getMessage());
+                        String msg = e.getMessage();
+                        if (msg != null && msg.toLowerCase().contains("closed")) {
+                            LOGGER.debug("Write RTCM failed on {}: {}", port, msg);
+                        } else {
+                            LOGGER.warn("Write RTCM failed on {}: {}", port, msg);
+                        }
                         closeQuietly(sink);
                         inConn = null;
                     }
@@ -314,6 +343,10 @@ public class RtkClusterService implements ApplicationRunner {
             }
         }
 
+        private boolean isSocketAlive(Socket s) {
+            return s != null && s.isConnected() && !s.isClosed() && !s.isInputShutdown() && !s.isOutputShutdown();
+        }
+
         private void closeQuietly(Socket s) {
             if (s == null) return;
             try { s.close(); } catch (IOException ignore) {}