Buggfix RELP

Author: anders-wartoft

src/main/java/nu/sitia/loggenerator/outputitems/RelpOutputItem.java

@@ -20,6 +20,7 @@
 import nu.sitia.loggenerator.Configuration;
 import java.io.*;
 import java.net.Socket;
+import java.net.SocketTimeoutException;
 import java.util.List;
 import java.util.logging.Logger;
 
@@ -121,14 +122,138 @@ public void send(List<String> toSend) {
         }
     }
 
+        /**
+     * Reconnect to the RELP server after disconnection
+     */
+    private void reconnect() throws IOException {
+        logger.info("Reconnecting to RELP server...");
+        try {
+            if (socket != null && !socket.isClosed()) {
+                socket.close();
+            }
+        } catch (IOException e) {
+            logger.warning("Error closing old socket: " + e.getMessage());
+        }
+        
+        socket = new Socket(hostname, port);
+        socket.setSoTimeout(5000);
+        outputStream = socket.getOutputStream();
+        inputReader = new BufferedReader(new InputStreamReader(socket.getInputStream()));
+        
+        // Send RELP OPEN command with proper offers
+        String openData = "relp_version=0\nrelp_software=LogGenerator\ncommands=syslog";
+        String openFrame = "1 open " + openData.length() + " " + openData + "\n";
+        logger.fine("Sending OPEN frame (reconnect): " + openFrame);
+        outputStream.write(openFrame.getBytes());
+        outputStream.flush();
+        
+        String response = readRelpResponse();
+        logger.fine("Received OPEN response (reconnect): " + response);
+        
+        if (response == null || !response.contains("rsp")) {
+            throw new RuntimeException("Failed to reconnect to RELP server. Response: " + response);
+        }
+        
+        transactionNr = 2;
+        connected = true;
+        logger.info("Reconnected to RELP server");
+    }
+
     /**
      * Send a single message via RELP protocol
      * @param message The message to send
      */
     private void sendRelpMessage(String message) {
+        sendRelpMessage(message, 0);
+    }
+    
+    /**
+     * Read a RELP response from the server
+     * @return The complete RELP response
+     * @throws IOException if reading fails
+     */
+    private String readRelpResponse() throws IOException {
+        // Read the response line: txnr command len [data]
+        String line = inputReader.readLine();
+        if (line == null) {
+            return null;
+        }
+        
+        logger.finest("Read RELP response line: " + line);
+        
+        // Parse the header: txnr command len [data]
+        String[] parts = line.split(" ", 4);
+        if (parts.length < 3) {
+            return line; // Return as-is if not proper RELP format
+        }
+        
+        try {
+            int dataLen = Integer.parseInt(parts[2]);
+            
+            if (dataLen == 0) {
+                // No data
+                return line;
+            }
+            
+            // Check if data is already on the same line
+            if (parts.length == 4 && parts[3].length() >= dataLen) {
+                return line;
+            }
+            
+            // Data is on subsequent lines or partially on this line
+            // Read exactly dataLen bytes
+            char[] dataBuffer = new char[dataLen];
+            int totalRead = 0;
+            
+            // If there's partial data on the header line, copy it first
+            if (parts.length == 4 && parts[3].length() > 0) {
+                int partialLen = Math.min(parts[3].length(), dataLen);
+                parts[3].getChars(0, partialLen, dataBuffer, 0);
+                totalRead = partialLen;
+            }
+            
+            // Read the remaining data
+            while (totalRead < dataLen) {
+                int read = inputReader.read(dataBuffer, totalRead, dataLen - totalRead);
+                if (read == -1) {
+                    break;
+                }
+                totalRead += read;
+            }
+            
+            // Reconstruct the full response
+            String data = new String(dataBuffer, 0, totalRead);
+            String result = parts[0] + " " + parts[1] + " " + parts[2] + " " + data;
+            logger.finest("Reconstructed RELP response: " + result);
+            return result;
+            
+        } catch (NumberFormatException e) {
+            return line; // Return as-is if len is not a number
+        }
+    }
+
+    /**
+     * Send a single message via RELP protocol with retry logic
+     * @param message The message to send
+     * @param retryCount Current retry count to prevent infinite recursion
+     */
+    private void sendRelpMessage(String message, int retryCount) {
+        if (retryCount > 3) {
+            throw new RuntimeException("Failed to send message after " + retryCount + " retries. Check rsyslog RELP configuration.");
+        }
+        
         try {
             if (!connected) {
-                throw new RuntimeException("Not connected to RELP server");
+                logger.fine("Not connected, attempting reconnect (retry " + retryCount + ")");
+                try {
+                    Thread.sleep(200);
+                    reconnect();
+                } catch (IOException e) {
+                    throw new RuntimeException("Failed to reconnect: " + e.getMessage(), e);
+                } catch (InterruptedException e) {
+                    Thread.currentThread().interrupt();
+                    throw new RuntimeException("Interrupted during reconnect: " + e.getMessage(), e);
+                }
             }
 
             String relpFrame = createRelpFrame("syslog", message);
@@ -137,20 +262,45 @@ private void sendRelpMessage(String message) {
             outputStream.flush();
 
             // Read response with timeout
-            String response = inputReader.readLine();
+            String response = readRelpResponse();
             if (response == null) {
                 connected = false;
-                throw new RuntimeException("Connection closed by server");
+                logger.fine("Connection closed by server, retrying (attempt " + (retryCount + 1) + ")");
+                sendRelpMessage(message, retryCount + 1);
+                return;
             }
-            if (!response.contains("rsp")) {
+            
+            logger.finer("Received RELP response: " + response);
+            
+            // Check if server is closing
+            if (response.contains("serverclose")) {
+                connected = false;
+                logger.fine("Server sent serverclose, retrying (attempt " + (retryCount + 1) + ")");
+                sendRelpMessage(message, retryCount + 1);
+                return;
+            }
+            
+            // Parse RELP response: txnr command len data
+            // Example: "2 rsp 2 OK" or "2 rsp 0 "
+            String[] parts = response.split(" ", 4);
+            if (parts.length < 2) {
                 throw new RuntimeException("Invalid response from RELP server: " + response);
             }
-            logger.finer("Received RELP response: " + response);
+            
+            String command = parts[1];
+            if (!"rsp".equalsIgnoreCase(command)) {
+                throw new RuntimeException("Invalid response from RELP server (expected 'rsp', got '" + command + "'): " + response);
+            }
+            
+            // Success - message sent
+            logger.finest("Message sent successfully");
+            
         } catch (IOException e) {
             connected = false;
             throw new RuntimeException("Failed to send RELP message: " + e.getMessage(), e);
         }
     }
+
 
     /**
      * Create a RELP frame
@@ -166,28 +316,43 @@ private String createRelpFrame(String command, String data) {
         return frame;
     }
 
-    @Override
+        @Override
     public void setup() throws RuntimeException {
         super.setup();
         try {
             socket = new Socket(hostname, port);
+            socket.setSoTimeout(5000); // 5 second timeout for reads
             outputStream = socket.getOutputStream();
             inputReader = new BufferedReader(new InputStreamReader(socket.getInputStream()));
 
-            // Send RELP OPEN command
-            String openFrame = "1 open 0 \n";
+            // Send RELP OPEN command with proper offers
+            // Format: txnr command len data\n
+            // Data format: relp_version=0\nrelp_software=LogGenerator\ncommands=syslog
+            String openData = "relp_version=0\nrelp_software=LogGenerator\ncommands=syslog";
+            String openFrame = "1 open " + openData.length() + " " + openData + "\n";
+            logger.fine("Sending OPEN frame: " + openFrame);
             outputStream.write(openFrame.getBytes());
             outputStream.flush();
 
-            // Read open response
-            String response = inputReader.readLine();
-            if (response == null || !response.contains("rsp")) {
-                throw new RuntimeException("Failed to open RELP connection");
+            // Read open response with timeout
+            String response = readRelpResponse();
+            logger.fine("Received OPEN response: " + response);
+            
+            if (response == null) {
+                throw new RuntimeException("No response from RELP server");
+            }
+            
+            // Check for valid response
+            if (!response.contains("rsp")) {
+                throw new RuntimeException("Failed to open RELP connection. Response: " + response);
             }
 
             connected = true;
             transactionNr = 2; // Next transaction number after OPEN
             logger.info("Connected to RELP server " + hostname + ":" + port);
+        } catch (SocketTimeoutException e) {
+            connected = false;
+            throw new RuntimeException("Timeout connecting to RELP server: " + e.getMessage(), e);
         } catch (IOException e) {
             connected = false;
             throw new RuntimeException("Failed to connect to RELP server: " + e.getMessage(), e);