refactor: parse localnodes responses with streaming parser

src/main/java/com/scylladb/alternator/internal/AlternatorLiveNodes.java

@@ -39,8 +39,6 @@
 public class AlternatorLiveNodes extends Thread {
   private static final long DEFAULT_SHUTDOWN_TIMEOUT_MS = 5_000;
 
-  private final String alternatorScheme;
-  private final int alternatorPort;
   private final AtomicReference<List<URI>> liveNodes;
   private final List<URI> initialNodes;
   private final AtomicInteger nextLiveNodeIndex;
@@ -50,6 +48,7 @@ public class AlternatorLiveNodes extends Thread {
   private final SdkHttpClient pollingHttpClient;
   private final boolean ownsPollingClient;
   private final AtomicLong lastActivityTime = new AtomicLong(0);
+  private final LocalNodesResponseParser localNodesResponseParser;
 
   private static Logger logger = Logger.getLogger(AlternatorLiveNodes.class.getName());
 
@@ -359,8 +358,8 @@ private AlternatorLiveNodes(
     if (seedHosts == null || seedHosts.isEmpty()) {
       throw new RuntimeException("config must contain at least one seed host");
     }
-    this.alternatorScheme = config.getScheme();
-    this.alternatorPort = config.getPort();
+    this.localNodesResponseParser =
+        new LocalNodesResponseParser(config.getScheme(), config.getPort());
     this.initialNodes = hostsToUris(seedHosts);
     this.liveNodes = new AtomicReference<>();
     this.nextLiveNodeIndex = new AtomicInteger(0);
@@ -450,18 +449,15 @@ public ValidationError(String message, Throwable cause) {
 
   private void validateConfig() throws ValidationError {
     try {
-      // Make sure that `alternatorScheme` and `alternatorPort` are correct values
+      // Make sure that the configured scheme and port are valid values.
       this.hostToURI("1.1.1.1");
     } catch (MalformedURLException | URISyntaxException e) {
       throw new ValidationError("failed to validate configuration", e);
     }
   }
 
   private URI hostToURI(String host) throws URISyntaxException, MalformedURLException {
-    URI uri = new URI(alternatorScheme, null, host, alternatorPort, null, null, null);
-    // Make sure that URI to URL conversion works
-    uri.toURL();
-    return uri;
+    return localNodesResponseParser.hostToURI(host);
   }
 
   private List<URI> hostsToUris(List<String> hosts) {
@@ -695,37 +691,12 @@ private List<URI> getNodes(URI uri) throws IOException {
   }
 
   private List<URI> parseLocalNodesResponse(String responseStr) {
-    // response looks like: ["127.0.0.2","127.0.0.3","127.0.0.1"]
-    responseStr = responseStr.trim();
-    if (!responseStr.startsWith("[") || !responseStr.endsWith("]")) {
+    try {
+      return localNodesResponseParser.parse(responseStr);
+    } catch (LocalNodesResponseParser.InvalidLocalNodesResponseException e) {
       logger.log(Level.WARNING, "Malformed /localnodes response: " + responseStr);
       return Collections.emptyList();
     }
-
-    String responseBody = responseStr.substring(1, responseStr.length() - 1).trim();
-    if (responseBody.isEmpty()) {
-      return Collections.emptyList();
-    }
-
-    String[] list = responseBody.split(",");
-    List<URI> newHosts = new ArrayList<>();
-    for (String host : list) {
-      host = host.trim();
-      if (host.isEmpty()) {
-        continue;
-      }
-      if (host.length() < 2 || !host.startsWith("\"") || !host.endsWith("\"")) {
-        logger.log(Level.WARNING, "Malformed host entry in /localnodes response: " + host);
-        continue;
-      }
-      host = host.substring(1, host.length() - 1);
-      try {
-        newHosts.add(this.hostToURI(host));
-      } catch (URISyntaxException | MalformedURLException e) {
-        logger.log(Level.WARNING, "Invalid host: " + host, e);
-      }
-    }
-    return newHosts;
   }
 
   /**

src/main/java/com/scylladb/alternator/internal/LocalNodesResponseParser.java

@@ -0,0 +1,165 @@
+package com.scylladb.alternator.internal;
+
+import java.io.IOException;
+import java.net.MalformedURLException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+final class LocalNodesResponseParser {
+  private final String alternatorScheme;
+  private final int alternatorPort;
+
+  private static final Logger logger = Logger.getLogger(LocalNodesResponseParser.class.getName());
+
+  LocalNodesResponseParser(String alternatorScheme, int alternatorPort) {
+    this.alternatorScheme = alternatorScheme;
+    this.alternatorPort = alternatorPort;
+  }
+
+  List<URI> parse(String responseBody) throws InvalidLocalNodesResponseException {
+    List<URI> nodes = new ArrayList<>();
+    for (String host : new JsonStringArrayParser(responseBody).parse()) {
+      try {
+        nodes.add(hostToURI(host));
+      } catch (URISyntaxException | MalformedURLException e) {
+        logger.log(Level.WARNING, "Invalid host: " + host, e);
+      }
+    }
+    return nodes;
+  }
+
+  URI hostToURI(String host) throws URISyntaxException, MalformedURLException {
+    URI uri = new URI(alternatorScheme, null, host, alternatorPort, null, null, null);
+    uri.toURL();
+    return uri;
+  }
+
+  static class InvalidLocalNodesResponseException extends IOException {
+    private InvalidLocalNodesResponseException(String message) {
+      super(message);
+    }
+  }
+
+  private static final class JsonStringArrayParser {
+    private final String body;
+    private int pos = 0;
+
+    private JsonStringArrayParser(String body) {
+      this.body = body != null ? body : "";
+    }
+
+    private List<String> parse() throws InvalidLocalNodesResponseException {
+      List<String> values = new ArrayList<>();
+      expect('[');
+      skipWhitespace();
+      if (peek(']')) {
+        pos++;
+        expectEnd();
+        return values;
+      }
+
+      while (true) {
+        values.add(parseString());
+        skipWhitespace();
+        if (peek(']')) {
+          pos++;
+          expectEnd();
+          return values;
+        }
+        expect(',');
+      }
+    }
+
+    private String parseString() throws InvalidLocalNodesResponseException {
+      expect('"');
+      StringBuilder value = new StringBuilder();
+      while (pos < body.length()) {
+        char ch = body.charAt(pos++);
+        if (ch == '"') {
+          return value.toString();
+        }
+        if (ch != '\\') {
+          value.append(ch);
+          continue;
+        }
+        value.append(parseEscapedCharacter());
+      }
+      throw invalid("unterminated /localnodes JSON response");
+    }
+
+    private char parseEscapedCharacter() throws InvalidLocalNodesResponseException {
+      if (pos >= body.length()) {
+        throw invalid("invalid escape in /localnodes JSON response");
+      }
+      char escaped = body.charAt(pos++);
+      switch (escaped) {
+        case '"':
+        case '\\':
+        case '/':
+          return escaped;
+        case 'b':
+          return '\b';
+        case 'f':
+          return '\f';
+        case 'n':
+          return '\n';
+        case 'r':
+          return '\r';
+        case 't':
+          return '\t';
+        case 'u':
+          return parseUnicodeEscape();
+        default:
+          throw invalid("unsupported escape in /localnodes JSON response");
+      }
+    }
+
+    private char parseUnicodeEscape() throws InvalidLocalNodesResponseException {
+      if (pos + 4 > body.length()) {
+        throw invalid("invalid unicode escape in /localnodes JSON response");
+      }
+      int codePoint = 0;
+      for (int i = 0; i < 4; i++) {
+        int digit = Character.digit(body.charAt(pos++), 16);
+        if (digit < 0) {
+          throw invalid("invalid unicode escape in /localnodes JSON response");
+        }
+        codePoint = (codePoint << 4) + digit;
+      }
+      return (char) codePoint;
+    }
+
+    private void expect(char expected) throws InvalidLocalNodesResponseException {
+      skipWhitespace();
+      if (pos >= body.length() || body.charAt(pos) != expected) {
+        throw invalid("invalid /localnodes JSON response");
+      }
+      pos++;
+    }
+
+    private void expectEnd() throws InvalidLocalNodesResponseException {
+      skipWhitespace();
+      if (pos != body.length()) {
+        throw invalid("invalid trailing data in /localnodes JSON response");
+      }
+    }
+
+    private boolean peek(char expected) {
+      return pos < body.length() && body.charAt(pos) == expected;
+    }
+
+    private void skipWhitespace() {
+      while (pos < body.length() && Character.isWhitespace(body.charAt(pos))) {
+        pos++;
+      }
+    }
+
+    private InvalidLocalNodesResponseException invalid(String message) {
+      return new InvalidLocalNodesResponseException(message);
+    }
+  }
+}

src/test/java/com/scylladb/alternator/internal/AlternatorLiveNodesClusterDiscoveryTest.java

@@ -272,6 +272,28 @@ public void testScopedLocalNodesQueryValuesAreEncodedOnce() throws Exception {
     assertFalse(encodedQuery.contains("%25"));
   }
 
+  @Test
+  public void testLocalNodesParserHandlesWhitespaceAndEscapedStrings() throws Exception {
+    Map<String, String> responses = new HashMap<>();
+    responses.put("seed.example.com", " [ \"node1.example.com\" , \"node\\u0032.example.com\" ] ");
+    DiscoveryHttpClient httpClient = new DiscoveryHttpClient(responses);
+
+    AlternatorConfig config =
+        AlternatorConfig.builder()
+            .withSeedHost("seed.example.com")
+            .withScheme("http")
+            .withPort(8000)
+            .withRoutingScope(ClusterScope.create())
+            .build();
+
+    AlternatorLiveNodes liveNodes = new AlternatorLiveNodes(config, httpClient);
+    liveNodes.updateLiveNodes();
+
+    assertEquals(
+        new LinkedHashSet<>(Arrays.asList("node1.example.com", "node2.example.com")),
+        hostSet(liveNodes.getLiveNodes()));
+  }
+
   @Test
   public void testCustomScopeLocalNodesQueryValuesAreEncodedWhenNeeded() throws Exception {
     Map<String, String> responses = new HashMap<>();

src/test/java/com/scylladb/alternator/internal/LocalNodesResponseParserTest.java

@@ -0,0 +1,66 @@
+package com.scylladb.alternator.internal;
+
+import static org.junit.Assert.*;
+
+import java.net.URI;
+import java.util.List;
+import org.junit.Test;
+
+public class LocalNodesResponseParserTest {
+  private final LocalNodesResponseParser parser = new LocalNodesResponseParser("http", 8000);
+
+  @Test
+  public void parsesEmptyArrays() throws Exception {
+    assertTrue(parser.parse("[]").isEmpty());
+    assertTrue(parser.parse(" [ ] ").isEmpty());
+  }
+
+  @Test
+  public void parsesValidMultiHostResponses() throws Exception {
+    List<URI> nodes = parser.parse("[\"node1.example.com\",\"node2.example.com\"]");
+
+    assertHosts(nodes, "node1.example.com", "node2.example.com");
+  }
+
+  @Test
+  public void parsesEscapedHostStrings() throws Exception {
+    List<URI> nodes =
+        parser.parse(" [ \"node\\u0031.example.com\" , \"node\\u0032.example.com\" ] ");
+
+    assertHosts(nodes, "node1.example.com", "node2.example.com");
+  }
+
+  @Test
+  public void skipsInvalidHostEntries() throws Exception {
+    List<URI> nodes = parser.parse("[\"node1.example.com\",\"bad host\",\"node2.example.com\"]");
+
+    assertHosts(nodes, "node1.example.com", "node2.example.com");
+  }
+
+  @Test
+  public void rejectsMalformedBodies() throws Exception {
+    assertMalformed("");
+    assertMalformed("not-json");
+    assertMalformed("[");
+    assertMalformed("[bad]");
+    assertMalformed("[\"node1.example.com\",bad]");
+  }
+
+  private void assertMalformed(String body) throws Exception {
+    try {
+      parser.parse(body);
+      fail("Expected malformed /localnodes body: " + body);
+    } catch (LocalNodesResponseParser.InvalidLocalNodesResponseException e) {
+      // expected
+    }
+  }
+
+  private static void assertHosts(List<URI> nodes, String... expectedHosts) {
+    assertEquals(expectedHosts.length, nodes.size());
+    for (int i = 0; i < expectedHosts.length; i++) {
+      assertEquals(expectedHosts[i], nodes.get(i).getHost());
+      assertEquals("http", nodes.get(i).getScheme());
+      assertEquals(8000, nodes.get(i).getPort());
+    }
+  }
+}