refactor(client): speak WebSocket through HttpClient instead of a hand-rolled framer

libsql-client/pom.xml

@@ -18,6 +18,10 @@
             <groupId>com.github.magnusp.libsql</groupId>
             <artifactId>libsql-hrana-codec</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.apache.httpcomponents.client5</groupId>
+            <artifactId>httpclient5-websocket</artifactId>
+        </dependency>
         <dependency>
             <groupId>org.junit.jupiter</groupId>
             <artifactId>junit-jupiter</artifactId>

libsql-itests/pom.xml

@@ -60,6 +60,15 @@
             <plugin>
                 <groupId>org.apache.maven.plugins</groupId>
                 <artifactId>maven-failsafe-plugin</artifactId>
+                <configuration>
+                    <systemPropertyVariables>
+                        <!-- Trust only the self-signed test certificate in front of sqld. -->
+                        <javax.net.ssl.trustStore>
+                            ${project.build.testOutputDirectory}/tls/truststore.p12
+                        </javax.net.ssl.trustStore>
+                        <javax.net.ssl.trustStorePassword>changeit</javax.net.ssl.trustStorePassword>
+                    </systemPropertyVariables>
+                </configuration>
             </plugin>
         </plugins>
     </build>

libsql-itests/src/test/java/com/github/magnusp/libsql/itest/SqldContainerBase.java

@@ -1,18 +1,24 @@
 package com.github.magnusp.libsql.itest;
 
-import org.junit.jupiter.api.BeforeAll;
 import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.Network;
 import org.testcontainers.containers.wait.strategy.Wait;
 import org.testcontainers.junit.jupiter.Container;
 import org.testcontainers.junit.jupiter.Testcontainers;
+import org.testcontainers.utility.DockerImageName;
+import org.testcontainers.utility.MountableFile;
 
 import java.time.Duration;
 
 @Testcontainers
 public abstract class SqldContainerBase {
 
     static final String LIBSQL_IMAGE = requireLibsqlImage();
+    private static final String NGINX_IMAGE = "nginx:1.27-alpine";
     private static final int PORT = 8080;
+    private static final int TLS_PORT = 8443;
+
+    private static final Network NETWORK = Network.newNetwork();
 
     private static String requireLibsqlImage() {
         String image = System.getProperty("libsql.image");
@@ -26,6 +32,8 @@ private static String requireLibsqlImage() {
 
     @Container
     public static final GenericContainer<?> LIBSQL_CONTAINER = new GenericContainer<>(LIBSQL_IMAGE)
+            .withNetwork(NETWORK)
+            .withNetworkAliases("sqld")
             .withExposedPorts(PORT)
             .withEnv("SQLD_NODE", "primary")
             .withEnv("SQLD_DISABLE_INTELLIGENT_THROTTLING", "true")
@@ -35,18 +43,35 @@ private static String requireLibsqlImage() {
                     .forStatusCode(405)
                     .withStartupTimeout(Duration.ofSeconds(60)));
 
-    @BeforeAll
-    public static void startContainer() {
+    /**
+     * A TLS terminator in front of sqld, so the {@code wss://} tests exercise a real handshake.
+     * Started on demand: only the WebSocket tests need it.
+     */
+    private static final GenericContainer<?> TLS_PROXY = new GenericContainer<>(DockerImageName.parse(NGINX_IMAGE))
+            .withNetwork(NETWORK)
+            .withExposedPorts(TLS_PORT)
+            .withCopyFileToContainer(MountableFile.forClasspathResource("tls/nginx.conf"), "/etc/nginx/nginx.conf")
+            .withCopyFileToContainer(
+                    MountableFile.forClasspathResource("tls/server.crt.pem"), "/etc/nginx/tls/server.crt.pem")
+            .withCopyFileToContainer(
+                    MountableFile.forClasspathResource("tls/server.key.pem"), "/etc/nginx/tls/server.key.pem")
+            .waitingFor(Wait.forListeningPort().withStartupTimeout(Duration.ofSeconds(60)));
+
+    static String getWebSocketUrl() {
+        return "ws://" + LIBSQL_CONTAINER.getHost() + ":" + LIBSQL_CONTAINER.getMappedPort(PORT);
+    }
+
+    static synchronized String getSecureWebSocketUrl() {
         if (!LIBSQL_CONTAINER.isRunning()) {
             LIBSQL_CONTAINER.start();
         }
+        if (!TLS_PROXY.isRunning()) {
+            TLS_PROXY.start();
+        }
+        return "wss://" + TLS_PROXY.getHost() + ":" + TLS_PROXY.getMappedPort(TLS_PORT);
     }
 
-    public static String getHttpUrl() {
+    static String getHttpUrl() {
         return "http://" + LIBSQL_CONTAINER.getHost() + ":" + LIBSQL_CONTAINER.getMappedPort(PORT);
     }
-
-    public static String getWebSocketUrl() {
-        return "ws://" + LIBSQL_CONTAINER.getHost() + ":" + LIBSQL_CONTAINER.getMappedPort(PORT);
-    }
 }

libsql-itests/src/test/java/com/github/magnusp/libsql/itest/WebSocketProtobufTransportIT.java

@@ -1,30 +1,38 @@
 package com.github.magnusp.libsql.itest;
 
-import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.MethodSource;
 
 import java.sql.Connection;
 import java.sql.DriverManager;
 import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.sql.Statement;
+import java.util.stream.Stream;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 /**
  * Drives the JDBC driver over Hrana 3 Protobuf on a WebSocket ({@code hrana3-protobuf}
- * subprotocol). The observable JDBC behavior must match the JSON transport.
+ * subprotocol). The observable JDBC behavior must match the JSON transport, over {@code ws://} and
+ * {@code wss://} alike.
  */
 public class WebSocketProtobufTransportIT extends SqldContainerBase {
 
-    private static Connection connect() throws SQLException {
-        return DriverManager.getConnection(
-                "jdbc:libsql:" + getWebSocketUrl() + "?encoding=protobuf");
+    static Stream<String> schemes() {
+        return Stream.of("ws", "wss");
     }
 
-    @Test
-    void selectReturnsTypedValuesIncluding64BitIntegerAndBlob() throws SQLException {
-        try (Connection connection = connect();
+    private static Connection connect(String scheme) throws SQLException {
+        String url = "wss".equals(scheme) ? getSecureWebSocketUrl() : getWebSocketUrl();
+        return DriverManager.getConnection("jdbc:libsql:" + url + "?encoding=protobuf");
+    }
+
+    @ParameterizedTest
+    @MethodSource("schemes")
+    void selectReturnsTypedValuesIncluding64BitIntegerAndBlob(String scheme) throws SQLException {
+        try (Connection connection = connect(scheme);
                 Statement statement = connection.createStatement();
                 ResultSet resultSet = statement.executeQuery(
                         "SELECT 9007199254740993 AS big, -9223372036854775808 AS min,"
@@ -40,28 +48,30 @@ void selectReturnsTypedValuesIncluding64BitIntegerAndBlob() throws SQLException
         }
     }
 
-    @Test
-    void transactionCommitsAndRollsBackOnOneStream() throws SQLException {
-        try (Connection connection = connect()) {
+    @ParameterizedTest
+    @MethodSource("schemes")
+    void transactionCommitsAndRollsBackOnOneStream(String scheme) throws SQLException {
+        try (Connection connection = connect(scheme)) {
             try (Statement statement = connection.createStatement()) {
-                statement.execute("CREATE TABLE ws_proto_tx (id INTEGER PRIMARY KEY, name TEXT)");
+                statement.execute(
+                        "CREATE TABLE ws_proto_tx_" + scheme + " (id INTEGER PRIMARY KEY, name TEXT)");
             }
 
             connection.setAutoCommit(false);
             try (Statement statement = connection.createStatement()) {
-                statement.execute("INSERT INTO ws_proto_tx VALUES (1, 'committed')");
+                statement.execute("INSERT INTO ws_proto_tx_" + scheme + " VALUES (1, 'committed')");
             }
             connection.commit();
 
             try (Statement statement = connection.createStatement()) {
-                statement.execute("INSERT INTO ws_proto_tx VALUES (2, 'rolled back')");
+                statement.execute("INSERT INTO ws_proto_tx_" + scheme + " VALUES (2, 'rolled back')");
             }
             connection.rollback();
             connection.setAutoCommit(true);
 
             try (Statement statement = connection.createStatement();
                     ResultSet resultSet = statement.executeQuery(
-                            "SELECT id, name FROM ws_proto_tx ORDER BY id")) {
+                            "SELECT id, name FROM ws_proto_tx_" + scheme + " ORDER BY id")) {
                 assertThat(resultSet.next()).isTrue();
                 assertThat(resultSet.getInt(1)).isEqualTo(1);
                 assertThat(resultSet.getString(2)).isEqualTo("committed");
@@ -70,9 +80,10 @@ void transactionCommitsAndRollsBackOnOneStream() throws SQLException {
         }
     }
 
-    @Test
-    void failingStatementMapsToSqlExceptionAndLeavesTheConnectionUsable() throws SQLException {
-        try (Connection connection = connect(); Statement statement = connection.createStatement()) {
+    @ParameterizedTest
+    @MethodSource("schemes")
+    void failingStatementMapsToSqlExceptionAndLeavesTheConnectionUsable(String scheme) throws SQLException {
+        try (Connection connection = connect(scheme); Statement statement = connection.createStatement()) {
             assertThatThrownBy(() -> statement.executeQuery("SELECT * FROM ws_proto_missing_table"))
                     .isInstanceOf(SQLException.class)
                     .hasMessageContaining("no such table");

libsql-itests/src/test/java/com/github/magnusp/libsql/itest/WebSocketTransportIT.java

@@ -4,6 +4,8 @@
 import com.github.magnusp.libsql.jdbc.LibsqlUrlParser;
 
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.MethodSource;
 
 import java.sql.Connection;
 import java.sql.DatabaseMetaData;
@@ -12,23 +14,34 @@
 import java.sql.SQLException;
 import java.sql.Statement;
 import java.util.Properties;
+import java.util.stream.Stream;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 /**
  * Drives the JDBC driver over Hrana 3 JSON on a WebSocket ({@code hrana3} subprotocol), which is
  * the same server and the same protocol structures as the HTTP transport but a different envelope.
+ * Every case runs over plain {@code ws://} and over {@code wss://} through a TLS terminator.
  */
 public class WebSocketTransportIT extends SqldContainerBase {
 
-    private static Connection connect() throws SQLException {
-        return DriverManager.getConnection("jdbc:libsql:" + getWebSocketUrl());
+    static Stream<String> schemes() {
+        return Stream.of("ws", "wss");
     }
 
-    @Test
-    void selectReturnsTypedValues() throws SQLException {
-        try (Connection connection = connect();
+    private static String url(String scheme) {
+        return "wss".equals(scheme) ? getSecureWebSocketUrl() : getWebSocketUrl();
+    }
+
+    private static Connection connect(String scheme) throws SQLException {
+        return DriverManager.getConnection("jdbc:libsql:" + url(scheme));
+    }
+
+    @ParameterizedTest
+    @MethodSource("schemes")
+    void selectReturnsTypedValues(String scheme) throws SQLException {
+        try (Connection connection = connect(scheme);
                 Statement statement = connection.createStatement();
                 ResultSet resultSet = statement.executeQuery(
                         "SELECT 1 AS i, 1.5 AS f, 'hello' AS t, x'010203' AS b, NULL AS n")) {
@@ -42,27 +55,30 @@ void selectReturnsTypedValues() throws SQLException {
         }
     }
 
-    @Test
-    void transactionCommitsAndRollsBackOnOneStream() throws SQLException {
-        try (Connection connection = connect()) {
+    @ParameterizedTest
+    @MethodSource("schemes")
+    void transactionCommitsAndRollsBackOnOneStream(String scheme) throws SQLException {
+        try (Connection connection = connect(scheme)) {
             try (Statement statement = connection.createStatement()) {
-                statement.execute("CREATE TABLE ws_tx (id INTEGER PRIMARY KEY, name TEXT)");
+                statement.execute(
+                        "CREATE TABLE ws_tx_" + scheme + " (id INTEGER PRIMARY KEY, name TEXT)");
             }
 
             connection.setAutoCommit(false);
             try (Statement statement = connection.createStatement()) {
-                statement.execute("INSERT INTO ws_tx VALUES (1, 'committed')");
+                statement.execute("INSERT INTO ws_tx_" + scheme + " VALUES (1, 'committed')");
             }
             connection.commit();
 
             try (Statement statement = connection.createStatement()) {
-                statement.execute("INSERT INTO ws_tx VALUES (2, 'rolled back')");
+                statement.execute("INSERT INTO ws_tx_" + scheme + " VALUES (2, 'rolled back')");
             }
             connection.rollback();
             connection.setAutoCommit(true);
 
             try (Statement statement = connection.createStatement();
-                    ResultSet resultSet = statement.executeQuery("SELECT id, name FROM ws_tx ORDER BY id")) {
+                    ResultSet resultSet = statement.executeQuery(
+                            "SELECT id, name FROM ws_tx_" + scheme + " ORDER BY id")) {
                 assertThat(resultSet.next()).isTrue();
                 assertThat(resultSet.getInt(1)).isEqualTo(1);
                 assertThat(resultSet.getString(2)).isEqualTo("committed");
@@ -71,9 +87,10 @@ void transactionCommitsAndRollsBackOnOneStream() throws SQLException {
         }
     }
 
-    @Test
-    void failingStatementMapsToSqlExceptionAndLeavesTheConnectionUsable() throws SQLException {
-        try (Connection connection = connect(); Statement statement = connection.createStatement()) {
+    @ParameterizedTest
+    @MethodSource("schemes")
+    void failingStatementMapsToSqlExceptionAndLeavesTheConnectionUsable(String scheme) throws SQLException {
+        try (Connection connection = connect(scheme); Statement statement = connection.createStatement()) {
             assertThatThrownBy(() -> statement.executeQuery("SELECT * FROM ws_missing_table"))
                     .isInstanceOf(SQLException.class)
                     .hasMessageContaining("no such table");
@@ -85,22 +102,25 @@ void failingStatementMapsToSqlExceptionAndLeavesTheConnectionUsable() throws SQL
         }
     }
 
-    @Test
-    void metadataAndBatchWorkOverWebSocket() throws SQLException {
-        try (Connection connection = connect()) {
+    @ParameterizedTest
+    @MethodSource("schemes")
+    void metadataAndBatchWorkOverWebSocket(String scheme) throws SQLException {
+        try (Connection connection = connect(scheme)) {
             DatabaseMetaData metaData = connection.getMetaData();
             assertThat(metaData.getDatabaseProductName()).isEqualTo("SQLite (libSQL)");
-            assertThat(metaData.getURL()).isEqualTo(getWebSocketUrl());
+            assertThat(metaData.getURL()).isEqualTo(url(scheme));
 
             try (Statement statement = connection.createStatement()) {
-                statement.execute("CREATE TABLE ws_batch (id INTEGER PRIMARY KEY, name TEXT)");
-                statement.addBatch("INSERT INTO ws_batch VALUES (1, 'a')");
-                statement.addBatch("INSERT INTO ws_batch VALUES (2, 'b')");
+                statement.execute(
+                        "CREATE TABLE ws_batch_" + scheme + " (id INTEGER PRIMARY KEY, name TEXT)");
+                statement.addBatch("INSERT INTO ws_batch_" + scheme + " VALUES (1, 'a')");
+                statement.addBatch("INSERT INTO ws_batch_" + scheme + " VALUES (2, 'b')");
                 assertThat(statement.executeBatch()).containsExactly(1, 1);
             }
 
             try (Statement statement = connection.createStatement();
-                    ResultSet resultSet = statement.executeQuery("SELECT COUNT(*) FROM ws_batch")) {
+                    ResultSet resultSet = statement.executeQuery(
+                            "SELECT COUNT(*) FROM ws_batch_" + scheme)) {
                 assertThat(resultSet.next()).isTrue();
                 assertThat(resultSet.getInt(1)).isEqualTo(2);
             }

libsql-itests/src/test/resources/tls/README.md

@@ -0,0 +1,19 @@
+# TLS fixtures for the `wss://` integration tests
+
+`server.crt.pem`, `server.key.pem`, and `truststore.p12` belong to a self-signed
+certificate for `localhost` and `127.0.0.1` that exists only so the test suite can
+put a TLS terminator in front of sqld. The key is public and worthless.
+
+Regenerate them when they expire (2036-10-21) with:
+
+```bash
+openssl req -x509 -newkey rsa:2048 -sha256 -days 3650 -nodes \
+  -keyout server.key.pem -out server.crt.pem \
+  -subj "/CN=localhost" -addext "subjectAltName=DNS:localhost,IP:127.0.0.1"
+openssl pkcs12 -export -in server.crt.pem -inkey server.key.pem \
+  -name libsql-test-tls -out truststore.p12 -passout pass:changeit
+```
+
+`libsql-itests/pom.xml` points the failsafe JVM at `truststore.p12` through
+`javax.net.ssl.trustStore`, so the driver trusts this certificate and nothing else
+during the integration tests.

libsql-itests/src/test/resources/tls/nginx.conf

@@ -0,0 +1,28 @@
+worker_processes 1;
+
+events {
+    worker_connections 128;
+}
+
+http {
+    map $http_upgrade $connection_upgrade {
+        default upgrade;
+        '' close;
+    }
+
+    server {
+        listen 8443 ssl;
+        ssl_certificate /etc/nginx/tls/server.crt.pem;
+        ssl_certificate_key /etc/nginx/tls/server.key.pem;
+
+        location / {
+            proxy_pass http://sqld:8080;
+            proxy_http_version 1.1;
+            proxy_set_header Upgrade $http_upgrade;
+            proxy_set_header Connection $connection_upgrade;
+            proxy_set_header Host $host;
+            proxy_read_timeout 3600s;
+            proxy_send_timeout 3600s;
+        }
+    }
+}