diff --git a/libsql-client/src/main/java/com/github/magnusp/libsql/client/LibsqlClient.java b/libsql-client/src/main/java/com/github/magnusp/libsql/client/LibsqlClient.java new file mode 100644 index 0000000..cf1ecfb --- /dev/null +++ b/libsql-client/src/main/java/com/github/magnusp/libsql/client/LibsqlClient.java @@ -0,0 +1,37 @@ +package com.github.magnusp.libsql.client; + +import com.github.magnusp.libsql.hrana.codec.Batch; +import com.github.magnusp.libsql.hrana.codec.BatchResult; +import com.github.magnusp.libsql.hrana.codec.Stmt; +import com.github.magnusp.libsql.hrana.codec.StmtResult; + +/** + * The transport-independent client seam used by the JDBC driver. + * + *

Both transports expose the same two execution styles. A one-shot call opens a stream, + * runs the work and closes the stream in as few round trips as possible, so it leaves no server-side + * state behind and needs no explicit cleanup. An on-stream call reuses a lazily opened + * interactive stream, which is what a transaction needs: the server keeps the transaction open + * between calls. + * + *

Implementations track their own stream state. {@link #hasOpenStream()} is the single source of + * truth for whether a transaction begin is still pending. + */ +public interface LibsqlClient extends AutoCloseable { + + StmtResult executeOneShot(Stmt stmt) throws LibsqlException; + + StmtResult executeOnStream(Stmt stmt) throws LibsqlException; + + BatchResult executeBatchOneShot(Batch batch) throws LibsqlException; + + BatchResult executeBatchOnStream(Batch batch) throws LibsqlException; + + boolean hasOpenStream(); + + /** Closes the interactive stream, if one is open, and releases its server-side state. */ + void closeStream() throws LibsqlException; + + @Override + void close(); +} diff --git a/libsql-client/src/main/java/com/github/magnusp/libsql/client/LibsqlHttpClient.java b/libsql-client/src/main/java/com/github/magnusp/libsql/client/LibsqlHttpClient.java index d432fcc..3316814 100644 --- a/libsql-client/src/main/java/com/github/magnusp/libsql/client/LibsqlHttpClient.java +++ b/libsql-client/src/main/java/com/github/magnusp/libsql/client/LibsqlHttpClient.java @@ -1,5 +1,7 @@ package com.github.magnusp.libsql.client; +import com.github.magnusp.libsql.hrana.codec.Batch; +import com.github.magnusp.libsql.hrana.codec.BatchResult; import com.github.magnusp.libsql.hrana.codec.DirectHranaJsonCodec; import com.github.magnusp.libsql.hrana.codec.HranaError; import com.github.magnusp.libsql.hrana.codec.HranaJsonCodec; @@ -24,7 +26,7 @@ import java.util.concurrent.Executors; import java.util.concurrent.ThreadLocalRandom; -public class LibsqlHttpClient { +public class LibsqlHttpClient implements LibsqlClient { private static final ConcurrentHashMap HTTP_CLIENT_CACHE = new ConcurrentHashMap<>(); @@ -33,6 +35,12 @@ public class LibsqlHttpClient { private final HranaJsonCodec codec; private final URI pipelineUri; + /** Guards the interactive stream state below. */ + private final Object streamLock = new Object(); + + private String baton; + private URI baseUrl; + public LibsqlHttpClient(LibsqlClientConfig config) { this(config, new DirectHranaJsonCodec()); } @@ -59,6 +67,7 @@ private static HttpClient getOrCreateHttpClient(Duration connectTimeout) { .build()); } + @Override public StmtResult executeOneShot(Stmt stmt) { // One-shot: [execute, close] in one round trip, so no idle stream is left for the server to reap. PipelineReqBody req = new PipelineReqBody( @@ -86,6 +95,107 @@ public StmtResult executeOneShot(Stmt stmt) { }; } + @Override + public StmtResult executeOnStream(Stmt stmt) { + StreamResult result; + synchronized (streamLock) { + PipelineReqBody req = new PipelineReqBody(baton, List.of(new StreamRequest.Execute(stmt))); + PipelineRespBody resp = sendPipeline(req, baseUrl); + updateStreamState(resp); + result = firstResult(resp); + } + return switch (result) { + case StreamResult.Ok ok -> { + if (ok.response() instanceof StreamResponse.Execute exec) { + yield exec.result(); + } + throw new LibsqlException("Unexpected stream response: " + ok.response(), null, 200); + } + case StreamResult.Error err -> throw new LibsqlException(err.error()); + }; + } + + @Override + public BatchResult executeBatchOneShot(Batch batch) { + // One-shot: [batch, close] in one round trip, so no idle stream is left for the server to reap. + PipelineReqBody req = new PipelineReqBody( + null, + List.of( + new StreamRequest.BatchReq(batch), + StreamRequest.Close.INSTANCE + ) + ); + + PipelineRespBody resp = sendPipelineWithRetry(req, pipelineUri); + return batchResult(firstResult(resp)); + } + + @Override + public BatchResult executeBatchOnStream(Batch batch) { + StreamResult result; + synchronized (streamLock) { + PipelineReqBody req = new PipelineReqBody(baton, List.of(new StreamRequest.BatchReq(batch))); + PipelineRespBody resp = sendPipeline(req, baseUrl); + updateStreamState(resp); + result = firstResult(resp); + } + return batchResult(result); + } + + @Override + public boolean hasOpenStream() { + synchronized (streamLock) { + return baton != null; + } + } + + @Override + public void closeStream() { + synchronized (streamLock) { + if (baton == null) { + return; + } + try { + PipelineReqBody req = new PipelineReqBody(baton, List.of(StreamRequest.Close.INSTANCE)); + PipelineRespBody resp = sendPipeline(req, baseUrl); + updateStreamState(resp); + } finally { + baton = null; + } + } + } + + @Override + public void close() { + // HTTP keeps no connection-level state; a pipeline request owns its whole stream. + } + + private static StreamResult firstResult(PipelineRespBody resp) { + if (resp.results().isEmpty()) { + throw new LibsqlException("Empty result list in response", null, 200); + } + return resp.results().get(0); + } + + private static BatchResult batchResult(StreamResult result) { + return switch (result) { + case StreamResult.Ok ok -> { + if (ok.response() instanceof StreamResponse.BatchResp batchResp) { + yield batchResp.result(); + } + throw new LibsqlException("Unexpected stream response: " + ok.response(), null, 200); + } + case StreamResult.Error err -> throw new LibsqlException(err.error()); + }; + } + + private void updateStreamState(PipelineRespBody resp) { + this.baton = resp.baton(); + if (resp.baseUrl() != null && !resp.baseUrl().isBlank()) { + this.baseUrl = URI.create(resp.baseUrl()); + } + } + public PipelineRespBody sendPipeline(PipelineReqBody reqBody, URI targetUri) { return sendPipelineWithRetry(reqBody, targetUri != null ? targetUri : pipelineUri); } diff --git a/libsql-client/src/main/java/com/github/magnusp/libsql/client/LibsqlWebSocketClient.java b/libsql-client/src/main/java/com/github/magnusp/libsql/client/LibsqlWebSocketClient.java new file mode 100644 index 0000000..c2afe0b --- /dev/null +++ b/libsql-client/src/main/java/com/github/magnusp/libsql/client/LibsqlWebSocketClient.java @@ -0,0 +1,695 @@ +package com.github.magnusp.libsql.client; + +import com.github.magnusp.libsql.hrana.codec.Batch; +import com.github.magnusp.libsql.hrana.codec.BatchResult; +import com.github.magnusp.libsql.hrana.codec.DirectHranaJsonCodec; +import com.github.magnusp.libsql.hrana.codec.HranaWsCodec; +import com.github.magnusp.libsql.hrana.codec.Stmt; +import com.github.magnusp.libsql.hrana.codec.StmtResult; +import com.github.magnusp.libsql.hrana.codec.WsClientMsg; +import com.github.magnusp.libsql.hrana.codec.WsRequest; +import com.github.magnusp.libsql.hrana.codec.WsResponse; +import com.github.magnusp.libsql.hrana.codec.WsServerMsg; + +import javax.net.ssl.SSLParameters; +import javax.net.ssl.SSLSocket; +import javax.net.ssl.SSLSocketFactory; +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.net.InetSocketAddress; +import java.net.Socket; +import java.net.URI; +import java.nio.charset.StandardCharsets; +import java.security.MessageDigest; +import java.security.NoSuchAlgorithmException; +import java.security.SecureRandom; +import java.time.Duration; +import java.util.Base64; +import java.util.HashMap; +import java.util.Locale; +import java.util.Map; +import java.util.Objects; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ThreadLocalRandom; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicInteger; + +/** + * Hrana 3 client over a WebSocket, speaking the JSON encoding ({@code hrana3} subprotocol). + * + *

One WebSocket connection multiplexes any number of streams. Requests are matched to responses + * by {@code request_id}, so the client may send a whole one-shot sequence (open, execute, close) + * without waiting in between and still receive the responses in order. + * + *

All request methods block the calling thread until the matching response arrives. The caller + * (the JDBC connection) serializes calls, so a single interactive stream id is enough. + * + *

Transport note. The WebSocket is spoken directly over a socket rather than + * through {@link java.net.http.WebSocket}. sqld answers the upgrade with a {@code 101} that carries + * {@code content-length: 31} and no body, and the JDK client completes its opening handshake only + * after reading a full response body ({@code OpeningHandshake.send()} uses {@code + * BodyHandlers.ofString()}), so it hangs against every sqld version. The framing below is the RFC + * 6455 client subset Hrana needs: text/binary frames, fragmentation, ping/pong and close. + */ +public class LibsqlWebSocketClient implements LibsqlClient { + + private static final String SUBPROTOCOL = "hrana3"; + private static final String WEBSOCKET_GUID = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"; + + private static final int OP_CONTINUATION = 0x0; + private static final int OP_TEXT = 0x1; + private static final int OP_BINARY = 0x2; + private static final int OP_CLOSE = 0x8; + private static final int OP_PING = 0x9; + private static final int OP_PONG = 0xA; + private static final int CLOSE_NORMAL = 1000; + + private static final int MAX_FRAME_BYTES = 1024 * 1024 * 1024; + + private final LibsqlClientConfig config; + private final HranaWsCodec codec; + private final Duration timeout; + private final Transport transport; + + private final ConcurrentHashMap> pending = + new ConcurrentHashMap<>(); + private final CompletableFuture hello = new CompletableFuture<>(); + private final AtomicInteger nextRequestId = new AtomicInteger(1); + private final AtomicInteger nextStreamId = new AtomicInteger(1); + + /** Guarded by {@code this}: the id of the lazily opened interactive stream, or -1. */ + private int interactiveStreamId = -1; + + private boolean closed; + + public LibsqlWebSocketClient(LibsqlClientConfig config) { + this(config, new DirectHranaJsonCodec()); + } + + public LibsqlWebSocketClient(LibsqlClientConfig config, HranaWsCodec codec) { + this.config = Objects.requireNonNull(config, "config cannot be null"); + this.codec = Objects.requireNonNull(codec, "codec cannot be null"); + this.timeout = config.connectTimeout() != null + ? config.connectTimeout() + : LibsqlClientConfig.DEFAULT_CONNECT_TIMEOUT; + this.transport = new Transport(URI.create(config.url()), timeout); + + try { + send(new WsClientMsg.Hello(authToken())); + awaitHello(); + } catch (RuntimeException e) { + transport.abort(); + throw e; + } + } + + private String authToken() { + String token = config.authToken(); + return token == null || token.isBlank() ? null : token; + } + + @Override + public StmtResult executeOneShot(Stmt stmt) { + int streamId = nextStreamId.getAndIncrement(); + CompletableFuture open = sendRequest(new WsRequest.OpenStream(streamId)); + CompletableFuture execute = + sendRequest(new WsRequest.Execute(streamId, stmt)); + CompletableFuture close = sendRequest(new WsRequest.CloseStream(streamId)); + + try { + await(open); + WsResponse response = await(execute); + if (!(response instanceof WsResponse.Execute result)) { + throw new LibsqlException("Unexpected response to execute: " + response, null, 200); + } + return result.result(); + } finally { + // The stream is already open, so the close must be sent even if execute failed. Its + // outcome cannot change the result the caller sees. + awaitCloseQuietly(close); + } + } + + @Override + public BatchResult executeBatchOneShot(Batch batch) { + int streamId = nextStreamId.getAndIncrement(); + CompletableFuture open = sendRequest(new WsRequest.OpenStream(streamId)); + CompletableFuture execute = sendRequest(new WsRequest.Batch(streamId, batch)); + CompletableFuture close = sendRequest(new WsRequest.CloseStream(streamId)); + + try { + await(open); + WsResponse response = await(execute); + if (!(response instanceof WsResponse.Batch result)) { + throw new LibsqlException("Unexpected response to batch: " + response, null, 200); + } + return result.result(); + } finally { + awaitCloseQuietly(close); + } + } + + @Override + public synchronized StmtResult executeOnStream(Stmt stmt) { + int streamId = ensureStream(); + WsResponse response = await(sendRequest(new WsRequest.Execute(streamId, stmt))); + if (!(response instanceof WsResponse.Execute result)) { + throw new LibsqlException("Unexpected response to execute: " + response, null, 200); + } + return result.result(); + } + + @Override + public synchronized BatchResult executeBatchOnStream(Batch batch) { + int streamId = ensureStream(); + WsResponse response = await(sendRequest(new WsRequest.Batch(streamId, batch))); + if (!(response instanceof WsResponse.Batch result)) { + throw new LibsqlException("Unexpected response to batch: " + response, null, 200); + } + return result.result(); + } + + @Override + public synchronized boolean hasOpenStream() { + return interactiveStreamId >= 0; + } + + @Override + public synchronized void closeStream() { + if (interactiveStreamId < 0) { + return; + } + int streamId = interactiveStreamId; + interactiveStreamId = -1; + await(sendRequest(new WsRequest.CloseStream(streamId))); + } + + @Override + public synchronized void close() { + if (closed) { + return; + } + closed = true; + try { + closeStream(); + } catch (LibsqlException e) { + // Closing a connection is best effort: the peer may already be gone and the server + // reaps the stream when it notices. + } finally { + transport.close(CLOSE_NORMAL, "closing"); + } + } + + /** Callers hold {@code this}. */ + private int ensureStream() { + if (interactiveStreamId < 0) { + int streamId = nextStreamId.getAndIncrement(); + await(sendRequest(new WsRequest.OpenStream(streamId))); + interactiveStreamId = streamId; + } + return interactiveStreamId; + } + + private CompletableFuture sendRequest(WsRequest request) { + int requestId = nextRequestId.getAndIncrement(); + CompletableFuture future = new CompletableFuture<>(); + pending.put(requestId, future); + try { + send(new WsClientMsg.Request(requestId, request)); + } catch (RuntimeException e) { + pending.remove(requestId); + future.completeExceptionally(e); + } + return future; + } + + private void send(WsClientMsg message) { + byte[] payload; + try { + payload = codec.serialize(message); + } catch (IOException e) { + throw new LibsqlException("Failed to serialize Hrana message: " + e.getMessage(), e); + } + transport.send(payload, codec.isBinary()); + } + + private void awaitHello() { + try { + WsServerMsg message = hello.get(timeout.toMillis(), TimeUnit.MILLISECONDS); + if (message instanceof WsServerMsg.HelloError error) { + throw new LibsqlException(error.error()); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new LibsqlException("Interrupted while waiting for the Hrana hello response", e); + } catch (TimeoutException e) { + throw new LibsqlException( + "Timed out after " + timeout.toMillis() + " ms waiting for the Hrana hello response", + e); + } catch (ExecutionException e) { + throw unwrap("Hrana hello failed", e); + } + } + + private WsResponse await(CompletableFuture future) { + try { + return future.get(timeout.toMillis(), TimeUnit.MILLISECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new LibsqlException("Interrupted while waiting for a Hrana response", e); + } catch (TimeoutException e) { + throw new LibsqlException( + "Timed out after " + timeout.toMillis() + " ms waiting for a Hrana response", e); + } catch (ExecutionException e) { + throw unwrap("Hrana request failed", e); + } + } + + private void awaitCloseQuietly(CompletableFuture close) { + try { + close.get(timeout.toMillis(), TimeUnit.MILLISECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } catch (TimeoutException | ExecutionException e) { + // The caller's result is already decided; a failed close only means the server will + // reap the stream. The response future is removed by the reader or on error. + } + } + + private static LibsqlException unwrap(String message, ExecutionException e) { + Throwable cause = e.getCause() != null ? e.getCause() : e; + if (cause instanceof LibsqlException libsql) { + return libsql; + } + return new LibsqlException(message + ": " + cause.getMessage(), cause); + } + + private void dispatch(byte[] payload) { + WsServerMsg message; + try { + message = codec.deserialize(payload); + } catch (IOException e) { + failAll(new LibsqlException( + "Failed to decode Hrana WebSocket message: " + e.getMessage(), e)); + transport.abort(); + return; + } + + switch (message) { + case WsServerMsg.HelloOk ok -> hello.complete(ok); + case WsServerMsg.HelloError error -> hello.completeExceptionally(new LibsqlException(error.error())); + case WsServerMsg.ResponseOk ok -> complete(ok.requestId(), ok.response()); + case WsServerMsg.ResponseError error -> fail(error.requestId(), new LibsqlException(error.error())); + } + } + + private void complete(int requestId, WsResponse response) { + CompletableFuture future = pending.remove(requestId); + if (future == null) { + protocolViolation("unexpected response for request_id " + requestId); + return; + } + future.complete(response); + } + + private void fail(int requestId, LibsqlException error) { + CompletableFuture future = pending.remove(requestId); + if (future == null) { + protocolViolation("unexpected error response for request_id " + requestId); + return; + } + future.completeExceptionally(error); + } + + private void protocolViolation(String detail) { + failAll(new LibsqlException("Hrana WebSocket protocol violation: " + detail, "WS_PROTOCOL", 0)); + transport.abort(); + } + + private void failAll(LibsqlException error) { + if (!hello.isDone()) { + hello.completeExceptionally(error); + } + for (Map.Entry> entry : pending.entrySet()) { + CompletableFuture future = pending.remove(entry.getKey()); + if (future != null) { + future.completeExceptionally(error); + } + } + } + + /** + * A minimal RFC 6455 client connection: the opening handshake plus frame encoding and decoding. + * + *

A reader thread routes complete text/binary messages to {@link #dispatch(String)} and + * fails the pending futures when the connection ends. It is a virtual thread, so an idle + * connection never keeps the JVM alive. + */ + private final class Transport { + + private Socket socket; + private InputStream in; + private OutputStream out; + private final Object writeLock = new Object(); + + private volatile boolean open = true; + + private Transport(URI uri, Duration timeout) { + Socket connected = null; + try { + connected = connect(uri, timeout); + this.socket = connected; + this.in = connected.getInputStream(); + this.out = connected.getOutputStream(); + handshake(uri, in, out, timeout); + } catch (IOException e) { + closeQuietly(connected); + throw new LibsqlException( + "Failed to open Hrana WebSocket to " + uri + ": " + e.getMessage(), e); + } + Thread.ofVirtual().name("hrana-ws-reader").start(this::readLoop); + } + + private void send(byte[] payload, boolean binary) { + sendFrame(binary ? OP_BINARY : OP_TEXT, payload); + } + + private void close(int statusCode, String reason) { + if (open) { + try { + sendFrame(OP_CLOSE, closePayload(statusCode, reason)); + } catch (LibsqlException e) { + // The peer is already gone; closing the socket below is enough. + } + } + open = false; + failAll(new LibsqlException( + "Hrana WebSocket closed with code " + statusCode, "WS_CLOSED", 0)); + closeQuietly(socket); + } + + private void abort() { + open = false; + closeQuietly(socket); + } + + private void sendFrame(int opcode, byte[] payload) { + synchronized (writeLock) { + if (!open) { + throw new LibsqlException("Hrana WebSocket is closed", "WS_CLOSED", 0); + } + try { + out.write(0x80 | opcode); + int length = payload.length; + if (length <= 125) { + out.write(0x80 | length); + } else if (length <= 0xFFFF) { + out.write(0x80 | 126); + out.write(length >>> 8); + out.write(length); + } else { + out.write(0x80 | 127); + for (int shift = 56; shift >= 0; shift -= 8) { + out.write((int) ((long) length >>> shift) & 0xFF); + } + } + byte[] mask = new byte[4]; + ThreadLocalRandom.current().nextBytes(mask); + out.write(mask); + for (int i = 0; i < length; i++) { + out.write(payload[i] ^ mask[i & 3]); + } + out.flush(); + } catch (IOException e) { + open = false; + throw new LibsqlException( + "Failed to send Hrana message: " + e.getMessage(), e); + } + } + } + + private void readLoop() { + ByteArrayOutputStream fragments = new ByteArrayOutputStream(); + int fragmentedOpcode = -1; + try { + while (open) { + int first = in.read(); + if (first < 0) { + break; + } + int second = readByte(in); + boolean fin = (first & 0x80) != 0; + int opcode = first & 0x0F; + if ((second & 0x80) != 0) { + protocolViolation("server sent a masked frame"); + return; + } + long length = second & 0x7F; + if (length == 126) { + length = (readByte(in) << 8) | readByte(in); + } else if (length == 127) { + length = 0; + for (int i = 0; i < 8; i++) { + length = (length << 8) | readByte(in); + } + } + if (length < 0 || length > MAX_FRAME_BYTES) { + protocolViolation("frame is too large: " + length + " bytes"); + return; + } + byte[] payload = in.readNBytes((int) length); + if (payload.length != length) { + break; + } + + switch (opcode) { + case OP_TEXT, OP_BINARY -> { + if (fragmentedOpcode != -1) { + protocolViolation("new message before the fragmented one finished"); + return; + } + if (fin) { + deliver(opcode, payload); + } else { + fragmentedOpcode = opcode; + fragments.write(payload); + } + } + case OP_CONTINUATION -> { + if (fragmentedOpcode == -1) { + protocolViolation("continuation without a started message"); + return; + } + fragments.write(payload); + if (fin) { + byte[] complete = fragments.toByteArray(); + int messageOpcode = fragmentedOpcode; + fragments.reset(); + fragmentedOpcode = -1; + deliver(messageOpcode, complete); + } + } + case OP_PING -> sendFrame(OP_PONG, payload); + case OP_PONG -> { + // Unsolicited pongs are allowed and need no answer. + } + case OP_CLOSE -> { + handlePeerClose(payload); + return; + } + default -> { + protocolViolation("unknown frame opcode 0x" + Integer.toHexString(opcode)); + return; + } + } + } + if (open) { + connectionLost("connection closed by the server without a close frame"); + } + } catch (IOException e) { + if (open) { + connectionLost("Hrana WebSocket connection failed: " + e.getMessage()); + } + } finally { + open = false; + closeQuietly(socket); + } + } + + private void deliver(int opcode, byte[] payload) { + boolean expectsBinary = codec.isBinary(); + if ((opcode == OP_BINARY) != expectsBinary) { + protocolViolation("frame type does not match the negotiated encoding"); + return; + } + dispatch(payload); + } + + private void handlePeerClose(byte[] payload) { + int statusCode = payload.length >= 2 + ? ((payload[0] & 0xFF) << 8) | (payload[1] & 0xFF) + : CLOSE_NORMAL; + String reason = payload.length > 2 + ? new String(payload, 2, payload.length - 2, StandardCharsets.UTF_8) + : ""; + try { + // Echo the close frame while the connection is still writable. + sendFrame(OP_CLOSE, closePayload(statusCode, "")); + } catch (LibsqlException e) { + // The peer closed; the socket is closed by the reader's finally block. + } + open = false; + connectionLost(reason.isBlank() + ? "Hrana WebSocket closed with code " + statusCode + : "Hrana WebSocket closed with code " + statusCode + ": " + reason); + } + + private void connectionLost(String detail) { + failAll(new LibsqlException(detail, "WS_CLOSED", 0)); + } + } + + private static Socket connect(URI uri, Duration timeout) throws IOException { + boolean tls = "wss".equalsIgnoreCase(uri.getScheme()); + String host = uri.getHost(); + if (host == null) { + throw new IOException("WebSocket URL has no host: " + uri); + } + int port = uri.getPort() > 0 ? uri.getPort() : (tls ? 443 : 80); + + Socket socket = new Socket(); + try { + int connectTimeout = (int) Math.min(Integer.MAX_VALUE, Math.max(0, timeout.toMillis())); + socket.connect(new InetSocketAddress(host, port), connectTimeout); + socket.setTcpNoDelay(true); + if (!tls) { + return socket; + } + SSLSocketFactory factory = (SSLSocketFactory) SSLSocketFactory.getDefault(); + SSLSocket ssl = (SSLSocket) factory.createSocket(socket, host, port, true); + SSLParameters parameters = ssl.getSSLParameters(); + parameters.setEndpointIdentificationAlgorithm("HTTPS"); + ssl.setSSLParameters(parameters); + ssl.startHandshake(); + return ssl; + } catch (IOException e) { + closeQuietly(socket); + throw e; + } + } + + private static void handshake(URI uri, InputStream in, OutputStream out, Duration timeout) + throws IOException { + String key = createKey(); + String expectedAccept = acceptKey(key); + String path = uri.getRawPath() == null || uri.getRawPath().isEmpty() ? "/" : uri.getRawPath(); + if (uri.getRawQuery() != null && !uri.getRawQuery().isEmpty()) { + path = path + "?" + uri.getRawQuery(); + } + String host = uri.getHost(); + int port = uri.getPort(); + String hostHeader = port > 0 ? host + ":" + port : host; + + StringBuilder request = new StringBuilder(256); + request.append("GET ").append(path).append(" HTTP/1.1\r\n") + .append("Host: ").append(hostHeader).append("\r\n") + .append("Upgrade: websocket\r\n") + .append("Connection: Upgrade\r\n") + .append("Sec-WebSocket-Key: ").append(key).append("\r\n") + .append("Sec-WebSocket-Version: 13\r\n") + .append("Sec-WebSocket-Protocol: ").append(SUBPROTOCOL).append("\r\n") + .append("\r\n"); + out.write(request.toString().getBytes(StandardCharsets.US_ASCII)); + out.flush(); + + String statusLine = readLine(in); + if (statusLine == null || !statusLine.contains(" 101 ")) { + throw new IOException("WebSocket upgrade rejected: " + statusLine); + } + Map headers = new HashMap<>(); + String line; + while ((line = readLine(in)) != null && !line.isEmpty()) { + int colon = line.indexOf(':'); + if (colon > 0) { + headers.put(line.substring(0, colon).trim().toLowerCase(Locale.ROOT), + line.substring(colon + 1).trim()); + } + } + if (!"websocket".equalsIgnoreCase(headers.get("upgrade"))) { + throw new IOException("WebSocket upgrade response missing Upgrade: websocket"); + } + if (!expectedAccept.equals(headers.get("sec-websocket-accept"))) { + throw new IOException("WebSocket upgrade response has an invalid Sec-WebSocket-Accept"); + } + String negotiated = headers.get("sec-websocket-protocol"); + if (!SUBPROTOCOL.equals(negotiated)) { + throw new IOException("Server did not negotiate subprotocol " + SUBPROTOCOL + + " (got " + negotiated + ")"); + } + } + + private static String createKey() { + byte[] nonce = new byte[16]; + new SecureRandom().nextBytes(nonce); + return Base64.getEncoder().encodeToString(nonce); + } + + private static String acceptKey(String key) { + try { + MessageDigest sha1 = MessageDigest.getInstance("SHA-1"); + byte[] digest = sha1.digest((key + WEBSOCKET_GUID).getBytes(StandardCharsets.US_ASCII)); + return Base64.getEncoder().encodeToString(digest); + } catch (NoSuchAlgorithmException e) { + throw new IllegalStateException("SHA-1 is required by the WebSocket protocol", e); + } + } + + private static byte[] closePayload(int statusCode, String reason) { + byte[] reasonBytes = reason == null ? new byte[0] : reason.getBytes(StandardCharsets.UTF_8); + byte[] payload = new byte[2 + reasonBytes.length]; + payload[0] = (byte) (statusCode >>> 8); + payload[1] = (byte) statusCode; + System.arraycopy(reasonBytes, 0, payload, 2, reasonBytes.length); + return payload; + } + + private static String readLine(InputStream in) throws IOException { + StringBuilder line = new StringBuilder(); + int b; + while ((b = in.read()) >= 0) { + if (b == '\n') { + int length = line.length(); + if (length > 0 && line.charAt(length - 1) == '\r') { + line.setLength(length - 1); + } + return line.toString(); + } + line.append((char) b); + } + return line.isEmpty() ? null : line.toString(); + } + + private static int readByte(InputStream in) throws IOException { + int b = in.read(); + if (b < 0) { + throw new IOException("Unexpected end of stream"); + } + return b; + } + + private static void closeQuietly(Socket socket) { + if (socket == null) { + return; + } + try { + socket.close(); + } catch (IOException ignored) { + // Nothing useful can be done with a failure to close. + } + } +} diff --git a/libsql-hrana-codec/src/main/java/com/github/magnusp/libsql/hrana/codec/DirectHranaJsonCodec.java b/libsql-hrana-codec/src/main/java/com/github/magnusp/libsql/hrana/codec/DirectHranaJsonCodec.java index d46042b..2ab6ac2 100644 --- a/libsql-hrana-codec/src/main/java/com/github/magnusp/libsql/hrana/codec/DirectHranaJsonCodec.java +++ b/libsql-hrana-codec/src/main/java/com/github/magnusp/libsql/hrana/codec/DirectHranaJsonCodec.java @@ -15,7 +15,7 @@ * High-performance, zero-dependency streaming JSON codec purpose-built for Hrana v3 protocol payloads. * Allocates no Jackson objects, no intermediate tree nodes, and operates directly with low-overhead scanning. */ -public class DirectHranaJsonCodec implements HranaJsonCodec { +public class DirectHranaJsonCodec implements HranaJsonCodec, HranaWsCodec { private static final Base64.Encoder B64_ENCODER = Base64.getEncoder(); private static final Base64.Decoder B64_DECODER = Base64.getDecoder(); @@ -583,6 +583,186 @@ private HranaError parseError(FastJsonParser p) throws IOException { return new HranaError(message, code); } + // ------------------------------------------------------------- + // Hrana over WebSocket (JSON text frames) + // ------------------------------------------------------------- + + @Override + public boolean isBinary() { + return false; + } + + @Override + public byte[] serialize(WsClientMsg msg) throws IOException { + FastJsonWriter w = new FastJsonWriter(256); + switch (msg) { + case WsClientMsg.Hello hello -> { + w.writeStartObject(); + w.writeStringField("type", "hello"); + w.writeComma(); + if (hello.jwt() == null) { + w.writeNullField("jwt"); + } else { + w.writeStringField("jwt", hello.jwt()); + } + w.writeEndObject(); + } + case WsClientMsg.Request request -> { + w.writeStartObject(); + w.writeStringField("type", "request"); + w.writeComma(); + w.writeNumberField("request_id", request.requestId()); + w.writeComma(); + w.writeFieldName("request"); + writeWsRequest(w, request.request()); + w.writeEndObject(); + } + } + return w.toByteArray(); + } + + private void writeWsRequest(FastJsonWriter w, WsRequest request) { + w.writeStartObject(); + switch (request) { + case WsRequest.OpenStream open -> { + w.writeStringField("type", "open_stream"); + w.writeComma(); + w.writeNumberField("stream_id", open.streamId()); + } + case WsRequest.CloseStream close -> { + w.writeStringField("type", "close_stream"); + w.writeComma(); + w.writeNumberField("stream_id", close.streamId()); + } + case WsRequest.Execute execute -> { + w.writeStringField("type", "execute"); + w.writeComma(); + w.writeNumberField("stream_id", execute.streamId()); + w.writeComma(); + w.writeFieldName("stmt"); + writeStmt(w, execute.stmt()); + } + case WsRequest.Batch batch -> { + w.writeStringField("type", "batch"); + w.writeComma(); + w.writeNumberField("stream_id", batch.streamId()); + w.writeComma(); + w.writeFieldName("batch"); + writeBatch(w, batch.batch()); + } + } + w.writeEndObject(); + } + + @Override + public WsServerMsg deserialize(byte[] payload) throws IOException { + FastJsonParser p = new FastJsonParser(payload); + p.expectObjectStart(); + String type = null; + Integer requestId = null; + WsResponse response = null; + HranaError error = null; + + while (!p.checkObjectEnd()) { + String field = p.nextFieldName(); + switch (field) { + case "type" -> type = p.nextString(); + case "request_id" -> requestId = (int) p.nextLong(); + case "response" -> response = parseWsResponse(p); + case "error" -> error = parseError(p); + default -> p.skipValue(); + } + } + + if (type == null) { + throw new IOException("WebSocket server message missing type field"); + } + return switch (type) { + case "hello_ok" -> WsServerMsg.HelloOk.INSTANCE; + case "hello_error" -> new WsServerMsg.HelloError(requireError(error, type)); + case "response_ok" -> new WsServerMsg.ResponseOk( + requireRequestId(requestId, type), requireResponse(response, type)); + case "response_error" -> new WsServerMsg.ResponseError( + requireRequestId(requestId, type), requireError(error, type)); + default -> throw new IOException("Unknown WebSocket server message type: " + type); + }; + } + + private WsResponse parseWsResponse(FastJsonParser p) throws IOException { + // JSON object field order is not significant, so find the type before reading the payload. + int start = p.pos; + p.expectObjectStart(); + String type = null; + while (!p.checkObjectEnd()) { + String field = p.nextFieldName(); + if ("type".equals(field)) { + type = p.nextString(); + } else { + p.skipValue(); + } + } + if (type == null) { + throw new IOException("WebSocket response missing type field"); + } + + p.pos = start; + p.expectObjectStart(); + StmtResult stmtResult = null; + BatchResult batchResult = null; + while (!p.checkObjectEnd()) { + String field = p.nextFieldName(); + if ("result".equals(field)) { + if ("execute".equals(type)) { + stmtResult = parseStmtResult(p); + } else if ("batch".equals(type)) { + batchResult = parseBatchResult(p); + } else { + p.skipValue(); + } + } else { + p.skipValue(); + } + } + return switch (type) { + case "open_stream" -> WsResponse.OpenStream.INSTANCE; + case "close_stream" -> WsResponse.CloseStream.INSTANCE; + case "execute" -> { + if (stmtResult == null) { + throw new IOException("WebSocket execute response missing result"); + } + yield new WsResponse.Execute(stmtResult); + } + case "batch" -> { + if (batchResult == null) { + throw new IOException("WebSocket batch response missing result"); + } + yield new WsResponse.Batch(batchResult); + } + default -> throw new IOException("Unsupported WebSocket response type: " + type); + }; + } + + private static int requireRequestId(Integer requestId, String type) throws IOException { + if (requestId == null) { + throw new IOException("WebSocket " + type + " message missing request_id"); + } + return requestId; + } + + private static WsResponse requireResponse(WsResponse response, String type) throws IOException { + if (response == null) { + throw new IOException("WebSocket " + type + " message missing response"); + } + return response; + } + + private static HranaError requireError(HranaError error, String type) throws IOException { + if (error == null) { + throw new IOException("WebSocket " + type + " message missing error"); + } + return error; + } + // ------------------------------------------------------------- // Direct Fast JSON Writer // ------------------------------------------------------------- diff --git a/libsql-hrana-codec/src/main/java/com/github/magnusp/libsql/hrana/codec/HranaWsCodec.java b/libsql-hrana-codec/src/main/java/com/github/magnusp/libsql/hrana/codec/HranaWsCodec.java new file mode 100644 index 0000000..720b091 --- /dev/null +++ b/libsql-hrana-codec/src/main/java/com/github/magnusp/libsql/hrana/codec/HranaWsCodec.java @@ -0,0 +1,19 @@ +package com.github.magnusp.libsql.hrana.codec; + +import java.io.IOException; + +/** + * SPI for serializing and deserializing Hrana 3 WebSocket messages. + * + *

One instance handles one frame at a time. With the JSON encoding every message is a complete + * UTF-8 JSON text frame; with the Protobuf encoding it is a binary frame. + */ +public interface HranaWsCodec { + + /** Whether messages must be sent as binary frames ({@code true}) or text frames. */ + boolean isBinary(); + + byte[] serialize(WsClientMsg msg) throws IOException; + + WsServerMsg deserialize(byte[] payload) throws IOException; +} diff --git a/libsql-hrana-codec/src/main/java/com/github/magnusp/libsql/hrana/codec/WsClientMsg.java b/libsql-hrana-codec/src/main/java/com/github/magnusp/libsql/hrana/codec/WsClientMsg.java new file mode 100644 index 0000000..ec3b442 --- /dev/null +++ b/libsql-hrana-codec/src/main/java/com/github/magnusp/libsql/hrana/codec/WsClientMsg.java @@ -0,0 +1,17 @@ +package com.github.magnusp.libsql.hrana.codec; + +/** + * A message sent by the client over a Hrana 3 WebSocket connection. + * + *

The {@code hello} message authenticates the connection; every other message is a numbered + * request whose response carries the same {@code request_id}. See the Hrana 3 specification, + * "Hrana over WebSocket". + */ +public sealed interface WsClientMsg { + + /** Authenticates (or re-authenticates) the connection. A null JWT means no authentication. */ + record Hello(String jwt) implements WsClientMsg {} + + /** Carries a numbered request; the server answers with a matching {@code response_ok/error}. */ + record Request(int requestId, WsRequest request) implements WsClientMsg {} +} diff --git a/libsql-hrana-codec/src/main/java/com/github/magnusp/libsql/hrana/codec/WsRequest.java b/libsql-hrana-codec/src/main/java/com/github/magnusp/libsql/hrana/codec/WsRequest.java new file mode 100644 index 0000000..c64becc --- /dev/null +++ b/libsql-hrana-codec/src/main/java/com/github/magnusp/libsql/hrana/codec/WsRequest.java @@ -0,0 +1,20 @@ +package com.github.magnusp.libsql.hrana.codec; + +/** + * The payload of a Hrana 3 WebSocket {@code request} message. + * + *

Unlike the HTTP variant, a stream is opened once per WebSocket connection and then reused for + * an arbitrary number of execute and batch requests. + */ +public sealed interface WsRequest { + + record OpenStream(int streamId) implements WsRequest {} + + record CloseStream(int streamId) implements WsRequest {} + + record Execute(int streamId, Stmt stmt) implements WsRequest {} + + // The component type is qualified because this nested record is itself named Batch and would + // otherwise shadow the top-level Batch type. + record Batch(int streamId, com.github.magnusp.libsql.hrana.codec.Batch batch) implements WsRequest {} +} diff --git a/libsql-hrana-codec/src/main/java/com/github/magnusp/libsql/hrana/codec/WsResponse.java b/libsql-hrana-codec/src/main/java/com/github/magnusp/libsql/hrana/codec/WsResponse.java new file mode 100644 index 0000000..f36aad3 --- /dev/null +++ b/libsql-hrana-codec/src/main/java/com/github/magnusp/libsql/hrana/codec/WsResponse.java @@ -0,0 +1,17 @@ +package com.github.magnusp.libsql.hrana.codec; + +/** The payload of a successful Hrana 3 WebSocket response. Its type matches the request type. */ +public sealed interface WsResponse { + + record OpenStream() implements WsResponse { + public static final OpenStream INSTANCE = new OpenStream(); + } + + record CloseStream() implements WsResponse { + public static final CloseStream INSTANCE = new CloseStream(); + } + + record Execute(StmtResult result) implements WsResponse {} + + record Batch(BatchResult result) implements WsResponse {} +} diff --git a/libsql-hrana-codec/src/main/java/com/github/magnusp/libsql/hrana/codec/WsServerMsg.java b/libsql-hrana-codec/src/main/java/com/github/magnusp/libsql/hrana/codec/WsServerMsg.java new file mode 100644 index 0000000..3c2fb78 --- /dev/null +++ b/libsql-hrana-codec/src/main/java/com/github/magnusp/libsql/hrana/codec/WsServerMsg.java @@ -0,0 +1,15 @@ +package com.github.magnusp.libsql.hrana.codec; + +/** A message sent by the server over a Hrana 3 WebSocket connection. */ +public sealed interface WsServerMsg { + + record HelloOk() implements WsServerMsg { + public static final HelloOk INSTANCE = new HelloOk(); + } + + record HelloError(HranaError error) implements WsServerMsg {} + + record ResponseOk(int requestId, WsResponse response) implements WsServerMsg {} + + record ResponseError(int requestId, HranaError error) implements WsServerMsg {} +} diff --git a/libsql-hrana-codec/src/test/java/com/github/magnusp/libsql/hrana/codec/DirectHranaWsJsonCodecTest.java b/libsql-hrana-codec/src/test/java/com/github/magnusp/libsql/hrana/codec/DirectHranaWsJsonCodecTest.java new file mode 100644 index 0000000..08b44a9 --- /dev/null +++ b/libsql-hrana-codec/src/test/java/com/github/magnusp/libsql/hrana/codec/DirectHranaWsJsonCodecTest.java @@ -0,0 +1,200 @@ +package com.github.magnusp.libsql.hrana.codec; + +import org.junit.jupiter.api.Test; + +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.util.List; +import java.util.Map; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +class DirectHranaWsJsonCodecTest { + + private final DirectHranaJsonCodec codec = new DirectHranaJsonCodec(); + + private static String text(byte[] bytes) { + return new String(bytes, StandardCharsets.UTF_8); + } + + private static byte[] utf8(String json) { + return json.getBytes(StandardCharsets.UTF_8); + } + + @Test + void codecUsesTextFrames() { + assertThat(codec).isInstanceOf(HranaWsCodec.class); + assertThat(codec.isBinary()).isFalse(); + } + + @Test + void serializesHello() throws IOException { + assertThat(text(codec.serialize(new WsClientMsg.Hello("jwt-token")))) + .isEqualTo("{\"type\":\"hello\",\"jwt\":\"jwt-token\"}"); + assertThat(text(codec.serialize(new WsClientMsg.Hello(null)))) + .isEqualTo("{\"type\":\"hello\",\"jwt\":null}"); + } + + @Test + void serializesOpenStreamRequest() throws IOException { + assertThat(text(codec.serialize(new WsClientMsg.Request(7, new WsRequest.OpenStream(3))))) + .isEqualTo("{\"type\":\"request\",\"request_id\":7," + + "\"request\":{\"type\":\"open_stream\",\"stream_id\":3}}"); + } + + @Test + void serializesCloseStreamRequest() throws IOException { + assertThat(text(codec.serialize(new WsClientMsg.Request(8, new WsRequest.CloseStream(3))))) + .isEqualTo("{\"type\":\"request\",\"request_id\":8," + + "\"request\":{\"type\":\"close_stream\",\"stream_id\":3}}"); + } + + @Test + void serializesExecuteWithPositionalAndNamedArgs() throws IOException { + Stmt stmt = new Stmt( + "SELECT :a, :b", + null, + List.of(Value.of(42L), Value.of("text \"quoted\"")), + List.of(new NamedArg(":a", Value.of(1.5)), new NamedArg(":b", new Value.Blob(new byte[] {1, 2, 3}))), + true); + + String json = text(codec.serialize(new WsClientMsg.Request(9, new WsRequest.Execute(4, stmt)))); + + assertThat(json).isEqualTo("{\"type\":\"request\",\"request_id\":9," + + "\"request\":{\"type\":\"execute\",\"stream_id\":4,\"stmt\":{" + + "\"sql\":\"SELECT :a, :b\"," + + "\"args\":[" + + "{\"type\":\"integer\",\"value\":\"42\"}," + + "{\"type\":\"text\",\"value\":\"text \\\"quoted\\\"\"}]," + + "\"named_args\":[" + + "{\"name\":\":a\",\"value\":{\"type\":\"float\",\"value\":1.5}}," + + "{\"name\":\":b\",\"value\":{\"type\":\"blob\",\"base64\":\"AQID\"}}]," + + "\"want_rows\":true}}}"); + } + + @Test + void serializesBatchRequest() throws IOException { + Batch batch = new Batch(List.of( + new BatchStep(new Stmt("BEGIN")), + new BatchStep(new BatchCond.StepOk(0), new Stmt("INSERT INTO t VALUES (?)", List.of(Value.of(1L)))), + new BatchStep(new BatchCond.Not(BatchCond.IsAutocommit.INSTANCE), new Stmt("COMMIT")))); + + String json = text(codec.serialize(new WsClientMsg.Request(10, new WsRequest.Batch(5, batch)))); + + assertThat(json).isEqualTo("{\"type\":\"request\",\"request_id\":10," + + "\"request\":{\"type\":\"batch\",\"stream_id\":5,\"batch\":{\"steps\":[" + + "{\"stmt\":{\"sql\":\"BEGIN\",\"want_rows\":true}}," + + "{\"condition\":{\"type\":\"ok\",\"step\":0}," + + "\"stmt\":{\"sql\":\"INSERT INTO t VALUES (?)\"," + + "\"args\":[{\"type\":\"integer\",\"value\":\"1\"}],\"want_rows\":true}}," + + "{\"condition\":{\"type\":\"not\",\"cond\":{\"type\":\"is_autocommit\"}}," + + "\"stmt\":{\"sql\":\"COMMIT\",\"want_rows\":true}}]}}}"); + } + + @Test + void deserializesHelloOk() throws IOException { + assertThat(codec.deserialize(utf8("{\"type\":\"hello_ok\"}"))) + .isEqualTo(WsServerMsg.HelloOk.INSTANCE); + } + + @Test + void deserializesHelloError() throws IOException { + assertThat(codec.deserialize(utf8( + "{\"type\":\"hello_error\",\"error\":{\"message\":\"bad jwt\",\"code\":\"AUTH\"}}"))) + .isEqualTo(new WsServerMsg.HelloError(new HranaError("bad jwt", "AUTH"))); + } + + @Test + void deserializesOpenAndCloseStreamResponses() throws IOException { + assertThat(codec.deserialize(utf8( + "{\"type\":\"response_ok\",\"request_id\":1," + + "\"response\":{\"type\":\"open_stream\"}}"))) + .isEqualTo(new WsServerMsg.ResponseOk(1, WsResponse.OpenStream.INSTANCE)); + + assertThat(codec.deserialize(utf8( + "{\"type\":\"response_ok\",\"request_id\":2," + + "\"response\":{\"type\":\"close_stream\"}}"))) + .isEqualTo(new WsServerMsg.ResponseOk(2, WsResponse.CloseStream.INSTANCE)); + } + + @Test + void deserializesExecuteResponse() throws IOException { + String json = """ + {"type":"response_ok","request_id":11,"response":{"type":"execute","result":{ + "cols":[{"name":"id","decltype":"INTEGER"},{"name":"name","decltype":"TEXT"}], + "rows":[[{"type":"integer","value":"1"},{"type":"text","value":"Alice"}], + [{"type":"null"},{"type":"blob","base64":"AQID"}]], + "affected_row_count":2, + "last_insert_rowid":"1", + "rows_read":5, + "rows_written":2, + "query_duration_ms":0.5 + }}}"""; + + StmtResult expected = new StmtResult( + List.of(new Col("id", "INTEGER"), new Col("name", "TEXT")), + List.of( + new Row(List.of(new Value.Integer(1), new Value.Text("Alice"))), + new Row(List.of(Value.ofNull(), new Value.Blob(new byte[] {1, 2, 3})))), + 2, + 1L); + + assertThat(codec.deserialize(utf8(json))) + .isEqualTo(new WsServerMsg.ResponseOk(11, new WsResponse.Execute(expected))); + } + + @Test + void deserializesBatchResponse() throws IOException { + String json = """ + {"type":"response_ok","request_id":12,"response":{"type":"batch","result":{ + "step_results":[{"cols":[],"rows":[],"affected_row_count":1,"last_insert_rowid":null},null], + "step_errors":[null,{"message":"UNIQUE constraint failed: t.id","code":"SQLITE_CONSTRAINT"}] + }}}"""; + + BatchResult expected = new BatchResult( + Map.of(0, new StmtResult(List.of(), List.of(), 1, null)), + Map.of(1, new HranaError("UNIQUE constraint failed: t.id", "SQLITE_CONSTRAINT"))); + + assertThat(codec.deserialize(utf8(json))) + .isEqualTo(new WsServerMsg.ResponseOk(12, new WsResponse.Batch(expected))); + } + + @Test + void deserializesExecuteResponseRegardlessOfFieldOrder() throws IOException { + String json = """ + {"response":{"result":{"cols":[],"rows":[],"affected_row_count":0}, + "type":"execute"},"request_id":14,"type":"response_ok"}"""; + + StmtResult expected = new StmtResult(List.of(), List.of(), 0, null); + + assertThat(codec.deserialize(utf8(json))) + .isEqualTo(new WsServerMsg.ResponseOk(14, new WsResponse.Execute(expected))); + } + + @Test + void deserializesResponseError() throws IOException { + assertThat(codec.deserialize(utf8( + "{\"type\":\"response_error\",\"request_id\":13," + + "\"error\":{\"message\":\"no such table: t\",\"code\":\"SQLITE_ERROR\"}}"))) + .isEqualTo(new WsServerMsg.ResponseError( + 13, new HranaError("no such table: t", "SQLITE_ERROR"))); + } + + @Test + void ignoresUnknownFields() throws IOException { + assertThat(codec.deserialize(utf8( + "{\"type\":\"hello_ok\",\"future_field\":{\"nested\":[1,2,3]}}"))) + .isEqualTo(WsServerMsg.HelloOk.INSTANCE); + } + + @Test + void rejectsUnknownMessageTypes() { + assertThatThrownBy(() -> codec.deserialize(utf8("{\"type\":\"mystery\"}"))) + .isInstanceOf(IOException.class) + .hasMessageContaining("mystery"); + assertThatThrownBy(() -> codec.deserialize(utf8("{}"))) + .isInstanceOf(IOException.class) + .hasMessageContaining("type"); + } +} diff --git a/libsql-itests/src/test/java/com/github/magnusp/libsql/itest/SqldContainerBase.java b/libsql-itests/src/test/java/com/github/magnusp/libsql/itest/SqldContainerBase.java index 1294a73..6025e8e 100644 --- a/libsql-itests/src/test/java/com/github/magnusp/libsql/itest/SqldContainerBase.java +++ b/libsql-itests/src/test/java/com/github/magnusp/libsql/itest/SqldContainerBase.java @@ -45,4 +45,8 @@ public static void startContainer() { public 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); + } } diff --git a/libsql-itests/src/test/java/com/github/magnusp/libsql/itest/WebSocketTransportIT.java b/libsql-itests/src/test/java/com/github/magnusp/libsql/itest/WebSocketTransportIT.java new file mode 100644 index 0000000..bd60e25 --- /dev/null +++ b/libsql-itests/src/test/java/com/github/magnusp/libsql/itest/WebSocketTransportIT.java @@ -0,0 +1,120 @@ +package com.github.magnusp.libsql.itest; + +import com.github.magnusp.libsql.client.LibsqlClientConfig; +import com.github.magnusp.libsql.jdbc.LibsqlUrlParser; + +import org.junit.jupiter.api.Test; + +import java.sql.Connection; +import java.sql.DatabaseMetaData; +import java.sql.DriverManager; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.sql.Statement; +import java.util.Properties; + +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. + */ +public class WebSocketTransportIT extends SqldContainerBase { + + private static Connection connect() throws SQLException { + return DriverManager.getConnection("jdbc:libsql:" + getWebSocketUrl()); + } + + @Test + void selectReturnsTypedValues() throws SQLException { + try (Connection connection = connect(); + 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")) { + assertThat(resultSet.next()).isTrue(); + assertThat(resultSet.getInt("i")).isEqualTo(1); + assertThat(resultSet.getDouble("f")).isEqualTo(1.5); + assertThat(resultSet.getString("t")).isEqualTo("hello"); + assertThat(resultSet.getBytes("b")).containsExactly(1, 2, 3); + assertThat(resultSet.getObject("n")).isNull(); + assertThat(resultSet.next()).isFalse(); + } + } + + @Test + void transactionCommitsAndRollsBackOnOneStream() throws SQLException { + try (Connection connection = connect()) { + try (Statement statement = connection.createStatement()) { + statement.execute("CREATE TABLE ws_tx (id INTEGER PRIMARY KEY, name TEXT)"); + } + + connection.setAutoCommit(false); + try (Statement statement = connection.createStatement()) { + statement.execute("INSERT INTO ws_tx VALUES (1, 'committed')"); + } + connection.commit(); + + try (Statement statement = connection.createStatement()) { + statement.execute("INSERT INTO ws_tx 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")) { + assertThat(resultSet.next()).isTrue(); + assertThat(resultSet.getInt(1)).isEqualTo(1); + assertThat(resultSet.getString(2)).isEqualTo("committed"); + assertThat(resultSet.next()).isFalse(); + } + } + } + + @Test + void failingStatementMapsToSqlExceptionAndLeavesTheConnectionUsable() throws SQLException { + try (Connection connection = connect(); Statement statement = connection.createStatement()) { + assertThatThrownBy(() -> statement.executeQuery("SELECT * FROM ws_missing_table")) + .isInstanceOf(SQLException.class) + .hasMessageContaining("no such table"); + + try (ResultSet resultSet = statement.executeQuery("SELECT 40 + 2")) { + assertThat(resultSet.next()).isTrue(); + assertThat(resultSet.getInt(1)).isEqualTo(42); + } + } + } + + @Test + void metadataAndBatchWorkOverWebSocket() throws SQLException { + try (Connection connection = connect()) { + DatabaseMetaData metaData = connection.getMetaData(); + assertThat(metaData.getDatabaseProductName()).isEqualTo("SQLite (libSQL)"); + assertThat(metaData.getURL()).isEqualTo(getWebSocketUrl()); + + 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')"); + assertThat(statement.executeBatch()).containsExactly(1, 1); + } + + try (Statement statement = connection.createStatement(); + ResultSet resultSet = statement.executeQuery("SELECT COUNT(*) FROM ws_batch")) { + assertThat(resultSet.next()).isTrue(); + assertThat(resultSet.getInt(1)).isEqualTo(2); + } + } + } + + @Test + void urlParserAcceptsWebSocketSchemes() throws SQLException { + assertThat(LibsqlUrlParser.acceptsUrl("jdbc:libsql:ws://localhost:8080")).isTrue(); + assertThat(LibsqlUrlParser.acceptsUrl("jdbc:libsql:wss://example.com/dev")).isTrue(); + assertThat(LibsqlUrlParser.acceptsUrl("jdbc:sqlite:ws://localhost:8080")).isTrue(); + + LibsqlClientConfig config = LibsqlUrlParser.parse( + "jdbc:libsql:ws://localhost:8080/dev/tenant?tls=true", new Properties()); + assertThat(config.url()).isEqualTo("wss://localhost:8080/dev/tenant"); + } +} diff --git a/libsql-jdbc/src/main/java/com/github/magnusp/libsql/jdbc/LibsqlConnection.java b/libsql-jdbc/src/main/java/com/github/magnusp/libsql/jdbc/LibsqlConnection.java index 4f1b211..4d12831 100644 --- a/libsql-jdbc/src/main/java/com/github/magnusp/libsql/jdbc/LibsqlConnection.java +++ b/libsql-jdbc/src/main/java/com/github/magnusp/libsql/jdbc/LibsqlConnection.java @@ -1,21 +1,16 @@ package com.github.magnusp.libsql.jdbc; +import com.github.magnusp.libsql.client.LibsqlClient; import com.github.magnusp.libsql.client.LibsqlClientConfig; import com.github.magnusp.libsql.client.LibsqlException; -import com.github.magnusp.libsql.client.LibsqlHttpClient; import com.github.magnusp.libsql.hrana.codec.Batch; import com.github.magnusp.libsql.hrana.codec.BatchCond; +import com.github.magnusp.libsql.hrana.codec.BatchResult; import com.github.magnusp.libsql.hrana.codec.BatchStep; import com.github.magnusp.libsql.hrana.codec.HranaError; -import com.github.magnusp.libsql.hrana.codec.PipelineReqBody; -import com.github.magnusp.libsql.hrana.codec.PipelineRespBody; import com.github.magnusp.libsql.hrana.codec.Stmt; import com.github.magnusp.libsql.hrana.codec.StmtResult; -import com.github.magnusp.libsql.hrana.codec.StreamRequest; -import com.github.magnusp.libsql.hrana.codec.StreamResponse; -import com.github.magnusp.libsql.hrana.codec.StreamResult; -import java.net.URI; import java.sql.Array; import java.sql.BatchUpdateException; import java.sql.Blob; @@ -44,9 +39,9 @@ * A JDBC connection whose state is serialized on one connection-level lock, the way sqlite-jdbc * serializes every call on its {@code DB} object. * - *

Hrana over HTTP requires serialized requests per stream and consumes a baton once, so each - * pipeline round trip and each transaction transition ({@link #setAutoCommit(boolean)}, {@link - * #commit()}, {@link #rollback()}, {@link #close()}) runs while holding {@link #stateLock}. + *

Both transports serialize requests on one interactive stream, and Hrana over HTTP consumes a + * baton once, so each round trip and each transaction transition ({@link #setAutoCommit(boolean)}, + * {@link #commit()}, {@link #rollback()}, {@link #close()}) runs while holding {@link #stateLock}. * Without it, two statements on one non-auto-commit connection can read the same baton, a commit * can land between another thread's begin and its send, and a close can leave a stream open. * @@ -60,9 +55,9 @@ public class LibsqlConnection implements Connection { private final LibsqlClientConfig config; - private final LibsqlHttpClient client; + private final LibsqlClient client; - /** Guards every field below and every pipeline round trip. See the class javadoc. */ + /** Guards every field below and every transport round trip. See the class javadoc. */ private final Object stateLock = new Object(); private boolean autoCommit = true; @@ -103,11 +98,9 @@ public static TransactionMode from(String mode) { } } - // Interactive stream state - private String baton = null; - private URI baseUrl = null; + // Interactive stream state lives in the client; see LibsqlClient.hasOpenStream(). - public LibsqlConnection(LibsqlClientConfig config, LibsqlHttpClient client) { + public LibsqlConnection(LibsqlClientConfig config, LibsqlClient client) { this.config = config; this.client = client; } @@ -133,23 +126,7 @@ public StmtResult executeStatement(Stmt stmt) throws SQLException { return client.executeOneShot(stmt); } else { ensureTransactionStarted(); - PipelineReqBody req = new PipelineReqBody(baton, List.of(new StreamRequest.Execute(stmt))); - PipelineRespBody resp = client.sendPipeline(req, baseUrl); - updateStreamState(resp); - - if (resp.results().isEmpty()) { - throw new SQLException("Empty result received from server"); - } - StreamResult sr = resp.results().get(0); - return switch (sr) { - case StreamResult.Ok ok -> { - if (ok.response() instanceof StreamResponse.Execute exec) { - yield exec.result(); - } - throw new SQLException("Unexpected response: " + ok.response()); - } - case StreamResult.Error err -> throw LibsqlErrorMapper.toSQLException(err.error()); - }; + return client.executeOnStream(stmt); } } catch (LibsqlException e) { throw LibsqlErrorMapper.toSQLException(e); @@ -186,44 +163,25 @@ List executeBatchResults(List statements) throws SQLException } Batch batch = new Batch(steps); + BatchResult result; if (autoCommit) { - // One-shot batch - PipelineReqBody req = new PipelineReqBody(null, List.of( - new StreamRequest.BatchReq(batch), - StreamRequest.Close.INSTANCE - )); - PipelineRespBody resp = client.sendPipeline(req, baseUrl); - return extractBatchResults(resp, statements.size()); + result = client.executeBatchOneShot(batch); } else { ensureTransactionStarted(); - PipelineReqBody req = new PipelineReqBody(baton, List.of(new StreamRequest.BatchReq(batch))); - PipelineRespBody resp = client.sendPipeline(req, baseUrl); - updateStreamState(resp); - return extractBatchResults(resp, statements.size()); + result = client.executeBatchOnStream(batch); } + return extractBatchResults(result, statements.size()); } catch (LibsqlException e) { throw LibsqlErrorMapper.toSQLException(e); } } } - private List extractBatchResults(PipelineRespBody resp, int expectedSize) + private List extractBatchResults(BatchResult batchResult, int expectedSize) throws SQLException { - if (resp.results().isEmpty()) { - throw new SQLException("Empty response for batch execution"); - } - StreamResult sr = resp.results().get(0); - if (sr instanceof StreamResult.Error err) { - throw LibsqlErrorMapper.toSQLException(err.error()); - } - StreamResponse response = ((StreamResult.Ok) sr).response(); - if (!(response instanceof StreamResponse.BatchResp batchResp)) { - throw new SQLException("Unexpected batch response: " + response); - } - List results = new ArrayList<>(expectedSize); for (int i = 0; i < expectedSize; i++) { - HranaError error = batchResp.result().stepErrors().get(i); + HranaError error = batchResult.stepErrors().get(i); if (error != null) { // JDBC stops the batch at the first failure and reports only the steps that ran. // The mapped exception is the cause, so SQLState and vendor code survive too. @@ -235,7 +193,7 @@ private List extractBatchResults(PipelineRespBody resp, int expected steppedUpdateCounts(results), mapped); } - StmtResult result = batchResp.result().stepResults().get(i); + StmtResult result = batchResult.stepResults().get(i); if (result == null) { // A skipped step is legitimate only after a failed step, which the loop has // already thrown on. Here the server executed the step but reported nothing. @@ -262,59 +220,38 @@ private void ensureTransactionStarted() throws SQLException { } /** - * The begin-pending state is derived rather than stored: auto-commit is off and no stream is - * open, so the next statement must begin the transaction. A separate boolean would be a second - * source of truth for a fact the baton already carries. + * The begin-pending state is derived rather than stored: auto-commit is off and the client has + * no open stream, so the next statement must begin the transaction. A separate boolean would be + * a second source of truth for a fact the client's stream state already carries. * *

Callers hold {@link #stateLock}. */ private boolean beginPending() { - return !autoCommit && baton == null; + return !autoCommit && !client.hasOpenStream(); } /** Callers hold {@link #stateLock}. */ private void beginTransaction() throws SQLException { - PipelineRespBody resp; try { - PipelineReqBody req = new PipelineReqBody(null, - List.of(new StreamRequest.Execute(new Stmt(transactionMode.beginSql())))); - resp = client.sendPipeline(req, baseUrl); + client.executeOnStream(new Stmt(transactionMode.beginSql())); } catch (LibsqlException e) { + // The begin failed, so the fresh stream holds no transaction state. + closeStreamQuietly(); throw LibsqlErrorMapper.toSQLException(e); } - updateStreamState(resp); - if (resp.results().isEmpty()) { - closeStreamSilently(); - throw new SQLException("Empty result received from server"); - } - StreamResult result = resp.results().get(0); - if (result instanceof StreamResult.Error error) { - closeStreamSilently(); - throw LibsqlErrorMapper.toSQLException(error.error()); - } - if (!(result instanceof StreamResult.Ok ok) - || !(ok.response() instanceof StreamResponse.Execute)) { - closeStreamSilently(); - throw new SQLException("Unexpected response: " + result); - } } /** Callers hold {@link #stateLock}. */ - private void updateStreamState(PipelineRespBody resp) { - this.baton = resp.baton(); - if (resp.baseUrl() != null && !resp.baseUrl().isBlank()) { - this.baseUrl = URI.create(resp.baseUrl()); + private void closeStreamQuietly() { + if (client == null) { + // Only a connection built directly with no transport (as some tests do) reaches this. + return; } - } - - /** Callers hold {@link #stateLock}. */ - private void closeStreamSilently() { - if (baton != null) { - try { - client.sendPipeline(new PipelineReqBody(baton, List.of(StreamRequest.Close.INSTANCE)), baseUrl); - } catch (Exception ignored) { - } - this.baton = null; + try { + client.closeStream(); + } catch (LibsqlException ignored) { + // Closing a stream is best effort: the transaction is already over or the transport is + // gone, and the server reaps the stream either way. } } @@ -417,33 +354,17 @@ private void endTransaction(String sql) throws SQLException { return; } - PipelineRespBody resp; try { - PipelineReqBody req = new PipelineReqBody(baton, - List.of(new StreamRequest.Execute(new Stmt(sql)))); - resp = client.sendPipeline(req, baseUrl); + client.executeOnStream(new Stmt(sql)); } catch (LibsqlException e) { throw LibsqlErrorMapper.toSQLException(e); } - updateStreamState(resp); - - if (resp.results().isEmpty()) { - throw new SQLException("Empty result received from server"); - } - StreamResult result = resp.results().get(0); - if (result instanceof StreamResult.Error error) { - throw LibsqlErrorMapper.toSQLException(error.error()); - } - if (!(result instanceof StreamResult.Ok ok) - || !(ok.response() instanceof StreamResponse.Execute)) { - throw new SQLException("Unexpected response: " + result); - } // The server committed or rolled back, so the stream has no state left worth holding. // Closing it lets an idle connection drop its server-side stream. On an error above the - // baton stays, so the caller can retry or roll back on the same stream. + // stream stays, so the caller can retry or roll back on the same stream. clearSavepoints(); - closeStreamSilently(); + closeStreamQuietly(); } @Override @@ -452,7 +373,10 @@ public void close() throws SQLException { if (!closed) { closed = true; clearSavepoints(); - closeStreamSilently(); + closeStreamQuietly(); + if (client != null) { + client.close(); + } } } } @@ -631,7 +555,7 @@ private Savepoint setSavepointInternal(String name) throws SQLException { executeTransactionStatement(sql); } catch (SQLException e) { // No transaction was started, so the fresh stream holds no state. - closeStreamSilently(); + closeStreamQuietly(); throw e; } autoCommit = false; @@ -686,7 +610,7 @@ public void releaseSavepoint(Savepoint savepoint) throws SQLException { // new one, as after a commit. transactionStartedBySavepoint = false; savepointNames.clear(); - closeStreamSilently(); + closeStreamQuietly(); } } } @@ -694,25 +618,11 @@ public void releaseSavepoint(Savepoint savepoint) throws SQLException { /** Callers hold {@link #stateLock}. */ private void executeTransactionStatement(String sql) throws SQLException { - PipelineRespBody resp; try { - PipelineReqBody req = new PipelineReqBody(baton, List.of(new StreamRequest.Execute(new Stmt(sql)))); - resp = client.sendPipeline(req, baseUrl); + client.executeOnStream(new Stmt(sql)); } catch (LibsqlException e) { throw LibsqlErrorMapper.toSQLException(e); } - updateStreamState(resp); - if (resp.results().isEmpty()) { - throw new SQLException("Empty result received from server"); - } - StreamResult result = resp.results().get(0); - if (result instanceof StreamResult.Error error) { - throw LibsqlErrorMapper.toSQLException(error.error()); - } - if (!(result instanceof StreamResult.Ok ok) - || !(ok.response() instanceof StreamResponse.Execute)) { - throw new SQLException("Unexpected response: " + result); - } } private static String savepointSqlName(Savepoint savepoint) throws SQLException { diff --git a/libsql-jdbc/src/main/java/com/github/magnusp/libsql/jdbc/LibsqlDriver.java b/libsql-jdbc/src/main/java/com/github/magnusp/libsql/jdbc/LibsqlDriver.java index 93777bd..c13dc11 100644 --- a/libsql-jdbc/src/main/java/com/github/magnusp/libsql/jdbc/LibsqlDriver.java +++ b/libsql-jdbc/src/main/java/com/github/magnusp/libsql/jdbc/LibsqlDriver.java @@ -1,7 +1,9 @@ package com.github.magnusp.libsql.jdbc; +import com.github.magnusp.libsql.client.LibsqlClient; import com.github.magnusp.libsql.client.LibsqlClientConfig; import com.github.magnusp.libsql.client.LibsqlHttpClient; +import com.github.magnusp.libsql.client.LibsqlWebSocketClient; import java.sql.Connection; import java.sql.Driver; @@ -31,10 +33,16 @@ public Connection connect(String url, Properties info) throws SQLException { return null; } LibsqlClientConfig config = LibsqlUrlParser.parse(url, info); - LibsqlHttpClient client = new LibsqlHttpClient(config); + LibsqlClient client = isWebSocketUrl(config.url()) + ? new LibsqlWebSocketClient(config) + : new LibsqlHttpClient(config); return new LibsqlConnection(config, client); } + private static boolean isWebSocketUrl(String url) { + return url.startsWith("ws://") || url.startsWith("wss://"); + } + @Override public boolean acceptsURL(String url) throws SQLException { return LibsqlUrlParser.acceptsUrl(url); diff --git a/libsql-jdbc/src/main/java/com/github/magnusp/libsql/jdbc/LibsqlUrlParser.java b/libsql-jdbc/src/main/java/com/github/magnusp/libsql/jdbc/LibsqlUrlParser.java index 9db2e22..2dbe019 100644 --- a/libsql-jdbc/src/main/java/com/github/magnusp/libsql/jdbc/LibsqlUrlParser.java +++ b/libsql-jdbc/src/main/java/com/github/magnusp/libsql/jdbc/LibsqlUrlParser.java @@ -17,7 +17,10 @@ public class LibsqlUrlParser { public static boolean acceptsUrl(String url) { if (url == null) return false; if (url.startsWith(PREFIX)) return true; - return url.startsWith("jdbc:sqlite:http://") || url.startsWith("jdbc:sqlite:https://"); + return url.startsWith("jdbc:sqlite:http://") + || url.startsWith("jdbc:sqlite:https://") + || url.startsWith("jdbc:sqlite:ws://") + || url.startsWith("jdbc:sqlite:wss://"); } public static LibsqlClientConfig parse(String url, Properties info) throws SQLException { @@ -32,8 +35,11 @@ public static LibsqlClientConfig parse(String url, Properties info) throws SQLEx rawUriStr = url.substring(SQLITE_PREFIX.length()); } - boolean hasHttpScheme = rawUriStr.startsWith("http://") || rawUriStr.startsWith("https://"); - if (!hasHttpScheme) { + boolean hasExplicitScheme = rawUriStr.startsWith("http://") + || rawUriStr.startsWith("https://") + || rawUriStr.startsWith("ws://") + || rawUriStr.startsWith("wss://"); + if (!hasExplicitScheme) { if (rawUriStr.startsWith("//")) { rawUriStr = "http:" + rawUriStr; } else { @@ -60,10 +66,14 @@ public static LibsqlClientConfig parse(String url, Properties info) throws SQLEx authToken = info.getProperty("authToken", info.getProperty("password", null)); } - // Check TLS override + // Check TLS override. A ws URL upgrades to wss just like http upgrades to https. String scheme = uri.getScheme(); if ("true".equalsIgnoreCase(queryParams.get("tls")) || "true".equalsIgnoreCase(queryParams.get("ssl"))) { - scheme = "https"; + scheme = switch (scheme) { + case "http" -> "https"; + case "ws" -> "wss"; + default -> scheme; + }; } // Timeouts @@ -85,14 +95,14 @@ public static LibsqlClientConfig parse(String url, Properties info) throws SQLEx String host = uri.getHost(); String path = uri.getPath() != null ? uri.getPath() : ""; - StringBuilder httpUrl = new StringBuilder(); - httpUrl.append(scheme).append("://").append(host); + StringBuilder endpointUrl = new StringBuilder(); + endpointUrl.append(scheme).append("://").append(host); if (port > 0) { - httpUrl.append(":").append(port); + endpointUrl.append(":").append(port); } - httpUrl.append(path); + endpointUrl.append(path); - return new LibsqlClientConfig(httpUrl.toString(), authToken, connectTimeout, busyTimeout); + return new LibsqlClientConfig(endpointUrl.toString(), authToken, connectTimeout, busyTimeout); } private static Map parseQuery(String query) { diff --git a/libsql-jdbc/src/test/java/com/github/magnusp/libsql/jdbc/LibsqlConnectionTest.java b/libsql-jdbc/src/test/java/com/github/magnusp/libsql/jdbc/LibsqlConnectionTest.java index d2862dd..645c2c7 100644 --- a/libsql-jdbc/src/test/java/com/github/magnusp/libsql/jdbc/LibsqlConnectionTest.java +++ b/libsql-jdbc/src/test/java/com/github/magnusp/libsql/jdbc/LibsqlConnectionTest.java @@ -1,8 +1,9 @@ package com.github.magnusp.libsql.jdbc; +import com.github.magnusp.libsql.client.LibsqlClient; import com.github.magnusp.libsql.client.LibsqlClientConfig; import com.github.magnusp.libsql.client.LibsqlException; -import com.github.magnusp.libsql.client.LibsqlHttpClient; +import com.github.magnusp.libsql.hrana.codec.Batch; import com.github.magnusp.libsql.hrana.codec.BatchCond; import com.github.magnusp.libsql.hrana.codec.BatchResult; import com.github.magnusp.libsql.hrana.codec.BatchStep; @@ -20,7 +21,6 @@ import org.junit.jupiter.params.provider.EnumSource; import org.junit.jupiter.params.provider.ValueSource; -import java.net.URI; import java.sql.BatchUpdateException; import java.sql.SQLException; import java.sql.SQLIntegrityConstraintViolationException; @@ -669,7 +669,7 @@ private static String executedSql(PipelineReqBody request) { return execute.stmt().sql(); } - private static final class ScriptedClient extends LibsqlHttpClient { + private static final class ScriptedClient implements LibsqlClient { @FunctionalInterface private interface Handler { @@ -682,14 +682,12 @@ private interface Handler { private int batonCount = 0; private ScriptedClient(Handler handler) { - super(CONFIG); this.handler = handler; } /** - * Mirrors the real client's one-shot close path, which does not go through {@link - * #sendPipeline(PipelineReqBody, URI)}: it wraps a server error in a {@link LibsqlException} - * so the connection's mapper sees the protocol error. + * Models the HTTP transport's observable behaviour: a one-shot request opens and closes its + * own stream, so it never touches the interactive baton. */ @Override public StmtResult executeOneShot(Stmt stmt) { @@ -697,8 +695,7 @@ public StmtResult executeOneShot(Stmt stmt) { new StreamRequest.Execute(stmt), StreamRequest.Close.INSTANCE)); requests.add(request); - PipelineRespBody response = handler.handle(request, requests.size() - 1); - StreamResult first = response.results().get(0); + StreamResult first = handler.handle(request, requests.size() - 1).results().get(0); return switch (first) { case StreamResult.Ok ok -> { if (ok.response() instanceof StreamResponse.Execute exec) { @@ -711,14 +708,78 @@ public StmtResult executeOneShot(Stmt stmt) { } @Override - public PipelineRespBody sendPipeline(PipelineReqBody request, URI targetUri) { + public BatchResult executeBatchOneShot(Batch batch) { + PipelineReqBody request = new PipelineReqBody(null, List.of( + new StreamRequest.BatchReq(batch), + StreamRequest.Close.INSTANCE)); + requests.add(request); + return batchResult(handler.handle(request, requests.size() - 1).results().get(0)); + } + + @Override + public StmtResult executeOnStream(Stmt stmt) { + StreamResult result = roundTrip(new PipelineReqBody( + issuedBaton, List.of(new StreamRequest.Execute(stmt)))); + return switch (result) { + case StreamResult.Ok ok -> { + if (ok.response() instanceof StreamResponse.Execute exec) { + yield exec.result(); + } + throw new LibsqlException("Unexpected stream response: " + ok.response(), null, 200); + } + case StreamResult.Error err -> throw new LibsqlException(err.error()); + }; + } + + @Override + public BatchResult executeBatchOnStream(Batch batch) { + StreamResult result = roundTrip(new PipelineReqBody( + issuedBaton, List.of(new StreamRequest.BatchReq(batch)))); + return batchResult(result); + } + + @Override + public boolean hasOpenStream() { + return issuedBaton != null; + } + + @Override + public void closeStream() { + if (issuedBaton == null) { + return; + } + PipelineReqBody request = new PipelineReqBody( + issuedBaton, List.of(StreamRequest.Close.INSTANCE)); + requests.add(request); + handler.handle(request, requests.size() - 1); + issuedBaton = null; + } + + @Override + public void close() { + // No connection-level state to release. + } + + private StreamResult roundTrip(PipelineReqBody request) { requests.add(request); if (request.baton() != null && !request.baton().equals(issuedBaton)) { throw new LibsqlException("stale baton " + request.baton(), "STALE_BATON", 400); } PipelineRespBody response = handler.handle(request, requests.size() - 1); issuedBaton = "baton-" + (++batonCount); - return new PipelineRespBody(issuedBaton, response.baseUrl(), response.results()); + return response.results().get(0); + } + + private static BatchResult batchResult(StreamResult result) { + return switch (result) { + case StreamResult.Ok ok -> { + if (ok.response() instanceof StreamResponse.BatchResp batchResp) { + yield batchResp.result(); + } + throw new LibsqlException("Unexpected stream response: " + ok.response(), null, 200); + } + case StreamResult.Error err -> throw new LibsqlException(err.error()); + }; } private List executedStatements() { diff --git a/libsql-jdbc/src/test/java/com/github/magnusp/libsql/jdbc/LibsqlSavepointTest.java b/libsql-jdbc/src/test/java/com/github/magnusp/libsql/jdbc/LibsqlSavepointTest.java index 708bed5..883152b 100644 --- a/libsql-jdbc/src/test/java/com/github/magnusp/libsql/jdbc/LibsqlSavepointTest.java +++ b/libsql-jdbc/src/test/java/com/github/magnusp/libsql/jdbc/LibsqlSavepointTest.java @@ -1,8 +1,10 @@ package com.github.magnusp.libsql.jdbc; +import com.github.magnusp.libsql.client.LibsqlClient; import com.github.magnusp.libsql.client.LibsqlClientConfig; import com.github.magnusp.libsql.client.LibsqlException; -import com.github.magnusp.libsql.client.LibsqlHttpClient; +import com.github.magnusp.libsql.hrana.codec.Batch; +import com.github.magnusp.libsql.hrana.codec.BatchResult; import com.github.magnusp.libsql.hrana.codec.HranaError; import com.github.magnusp.libsql.hrana.codec.PipelineReqBody; import com.github.magnusp.libsql.hrana.codec.PipelineRespBody; @@ -13,7 +15,6 @@ import com.github.magnusp.libsql.hrana.codec.StreamResult; import org.junit.jupiter.api.Test; -import java.net.URI; import java.sql.SQLException; import java.sql.SQLSyntaxErrorException; import java.sql.Savepoint; @@ -331,7 +332,7 @@ private static String executedSql(PipelineReqBody request) { return execute.stmt().sql(); } - private static final class ScriptedClient extends LibsqlHttpClient { + private static final class ScriptedClient implements LibsqlClient { @FunctionalInterface private interface Handler { @@ -344,7 +345,6 @@ private interface Handler { private int batonCount = 0; private ScriptedClient(Handler handler) { - super(CONFIG); this.handler = handler; } @@ -354,8 +354,7 @@ public StmtResult executeOneShot(Stmt stmt) { new StreamRequest.Execute(stmt), StreamRequest.Close.INSTANCE)); requests.add(request); - PipelineRespBody response = handler.handle(request, requests.size() - 1); - StreamResult first = response.results().get(0); + StreamResult first = handler.handle(request, requests.size() - 1).results().get(0); return switch (first) { case StreamResult.Ok ok -> { if (ok.response() instanceof StreamResponse.Execute exec) { @@ -368,14 +367,78 @@ public StmtResult executeOneShot(Stmt stmt) { } @Override - public PipelineRespBody sendPipeline(PipelineReqBody request, URI targetUri) { + public BatchResult executeBatchOneShot(Batch batch) { + PipelineReqBody request = new PipelineReqBody(null, List.of( + new StreamRequest.BatchReq(batch), + StreamRequest.Close.INSTANCE)); + requests.add(request); + return batchResult(handler.handle(request, requests.size() - 1).results().get(0)); + } + + @Override + public StmtResult executeOnStream(Stmt stmt) { + StreamResult result = roundTrip(new PipelineReqBody( + issuedBaton, List.of(new StreamRequest.Execute(stmt)))); + return switch (result) { + case StreamResult.Ok ok -> { + if (ok.response() instanceof StreamResponse.Execute exec) { + yield exec.result(); + } + throw new LibsqlException("Unexpected stream response: " + ok.response(), null, 200); + } + case StreamResult.Error err -> throw new LibsqlException(err.error()); + }; + } + + @Override + public BatchResult executeBatchOnStream(Batch batch) { + StreamResult result = roundTrip(new PipelineReqBody( + issuedBaton, List.of(new StreamRequest.BatchReq(batch)))); + return batchResult(result); + } + + @Override + public boolean hasOpenStream() { + return issuedBaton != null; + } + + @Override + public void closeStream() { + if (issuedBaton == null) { + return; + } + PipelineReqBody request = new PipelineReqBody( + issuedBaton, List.of(StreamRequest.Close.INSTANCE)); + requests.add(request); + handler.handle(request, requests.size() - 1); + issuedBaton = null; + } + + @Override + public void close() { + // No connection-level state to release. + } + + private StreamResult roundTrip(PipelineReqBody request) { requests.add(request); if (request.baton() != null && !request.baton().equals(issuedBaton)) { throw new LibsqlException("stale baton " + request.baton(), "STALE_BATON", 400); } PipelineRespBody response = handler.handle(request, requests.size() - 1); issuedBaton = "baton-" + (++batonCount); - return new PipelineRespBody(issuedBaton, response.baseUrl(), response.results()); + return response.results().get(0); + } + + private static BatchResult batchResult(StreamResult result) { + return switch (result) { + case StreamResult.Ok ok -> { + if (ok.response() instanceof StreamResponse.BatchResp batchResp) { + yield batchResp.result(); + } + throw new LibsqlException("Unexpected stream response: " + ok.response(), null, 200); + } + case StreamResult.Error err -> throw new LibsqlException(err.error()); + }; } private List executedStatements() { diff --git a/libsql-jdbc/src/test/java/com/github/magnusp/libsql/jdbc/LibsqlStatementConcurrencyTest.java b/libsql-jdbc/src/test/java/com/github/magnusp/libsql/jdbc/LibsqlStatementConcurrencyTest.java index 08dc22b..e375fb5 100644 --- a/libsql-jdbc/src/test/java/com/github/magnusp/libsql/jdbc/LibsqlStatementConcurrencyTest.java +++ b/libsql-jdbc/src/test/java/com/github/magnusp/libsql/jdbc/LibsqlStatementConcurrencyTest.java @@ -1,7 +1,10 @@ package com.github.magnusp.libsql.jdbc; +import com.github.magnusp.libsql.client.LibsqlClient; import com.github.magnusp.libsql.client.LibsqlClientConfig; -import com.github.magnusp.libsql.client.LibsqlHttpClient; +import com.github.magnusp.libsql.client.LibsqlException; +import com.github.magnusp.libsql.hrana.codec.Batch; +import com.github.magnusp.libsql.hrana.codec.BatchResult; import com.github.magnusp.libsql.hrana.codec.Col; import com.github.magnusp.libsql.hrana.codec.PipelineReqBody; import com.github.magnusp.libsql.hrana.codec.PipelineRespBody; @@ -14,7 +17,6 @@ import com.github.magnusp.libsql.hrana.codec.Value; import org.junit.jupiter.api.Test; -import java.net.URI; import java.sql.ResultSet; import java.sql.SQLException; import java.util.ArrayList; @@ -303,7 +305,7 @@ protected void beforeResultSetAssignment() throws SQLException { * request on a stream at a time, and a baton that is consumed by its request. Every rule break * is recorded instead of thrown, so a failing case reports why the interleaving was invalid. */ - private static final class ProtocolClient extends LibsqlHttpClient { + private static final class ProtocolClient implements LibsqlClient { private final Object lock = new Object(); private final List violations = new ArrayList<>(); @@ -317,18 +319,107 @@ private static final class ProtocolClient extends LibsqlHttpClient { private boolean streamOpen; private ProtocolClient(CyclicBarrier beginRace) { - super(CONFIG); this.beginRace = beginRace; } @Override - public PipelineRespBody sendPipeline(PipelineReqBody request, URI targetUri) { + public StmtResult executeOneShot(Stmt stmt) { + PipelineReqBody request = new PipelineReqBody(null, List.of( + new StreamRequest.Execute(stmt), + StreamRequest.Close.INSTANCE)); + PipelineRespBody response = roundTrip(request, false, true); + StreamResult first = response.results().get(0); + return switch (first) { + case StreamResult.Ok ok -> { + if (ok.response() instanceof StreamResponse.Execute exec) { + yield exec.result(); + } + throw new LibsqlException("Unexpected stream response: " + ok.response(), null, 200); + } + case StreamResult.Error err -> throw new LibsqlException(err.error()); + }; + } + + @Override + public StmtResult executeOnStream(Stmt stmt) { + PipelineReqBody request = new PipelineReqBody( + currentBaton(), List.of(new StreamRequest.Execute(stmt))); + StreamResult first = roundTrip(request, true, false).results().get(0); + return switch (first) { + case StreamResult.Ok ok -> { + if (ok.response() instanceof StreamResponse.Execute exec) { + yield exec.result(); + } + throw new LibsqlException("Unexpected stream response: " + ok.response(), null, 200); + } + case StreamResult.Error err -> throw new LibsqlException(err.error()); + }; + } + + @Override + public BatchResult executeBatchOneShot(Batch batch) { + PipelineReqBody request = new PipelineReqBody(null, List.of( + new StreamRequest.BatchReq(batch), + StreamRequest.Close.INSTANCE)); + return batchResult(roundTrip(request, false, true).results().get(0)); + } + + @Override + public BatchResult executeBatchOnStream(Batch batch) { + PipelineReqBody request = new PipelineReqBody( + currentBaton(), List.of(new StreamRequest.BatchReq(batch))); + return batchResult(roundTrip(request, true, false).results().get(0)); + } + + @Override + public boolean hasOpenStream() { + synchronized (lock) { + return streamOpen; + } + } + + @Override + public void closeStream() { + PipelineReqBody request; + synchronized (lock) { + if (!streamOpen) { + return; + } + request = new PipelineReqBody(liveBaton, List.of(StreamRequest.Close.INSTANCE)); + } + roundTrip(request, false, true); + } + + @Override + public void close() { + // No connection-level state to release. + } + + private String currentBaton() { + synchronized (lock) { + return streamOpen ? liveBaton : null; + } + } + + private static BatchResult batchResult(StreamResult result) { + return switch (result) { + case StreamResult.Ok ok -> { + if (ok.response() instanceof StreamResponse.BatchResp batchResp) { + yield batchResp.result(); + } + throw new LibsqlException("Unexpected stream response: " + ok.response(), null, 200); + } + case StreamResult.Error err -> throw new LibsqlException(err.error()); + }; + } + + private PipelineRespBody roundTrip( + PipelineReqBody request, boolean opensStream, boolean closesStream) { String baton = request.baton(); StreamRequest first = request.requests().get(0); String sql = first instanceof StreamRequest.Execute execute ? execute.stmt().sql() : null; - boolean close = first instanceof StreamRequest.Close; synchronized (lock) { if (inFlightRequests > 0) { @@ -359,15 +450,15 @@ public PipelineRespBody sendPipeline(PipelineReqBody request, URI targetUri) { } } synchronized (lock) { - if (close) { + if (closesStream) { liveBaton = null; streamOpen = false; - } else { + } else if (opensStream || streamOpen) { liveBaton = "baton-" + (++batonCount); streamOpen = true; } return new PipelineRespBody( - close ? null : liveBaton, null, responsesFor(request)); + closesStream ? null : liveBaton, null, responsesFor(request)); } } finally { synchronized (lock) { @@ -428,12 +519,6 @@ private List violations() { return List.copyOf(violations); } } - - private boolean hasOpenStream() { - synchronized (lock) { - return streamOpen; - } - } } private static final class RecordingConnection extends LibsqlConnection { diff --git a/libsql-jdbc/src/test/java/com/github/magnusp/libsql/jdbc/LibsqlUrlParserTest.java b/libsql-jdbc/src/test/java/com/github/magnusp/libsql/jdbc/LibsqlUrlParserTest.java index 2d1c5e9..d295da9 100644 --- a/libsql-jdbc/src/test/java/com/github/magnusp/libsql/jdbc/LibsqlUrlParserTest.java +++ b/libsql-jdbc/src/test/java/com/github/magnusp/libsql/jdbc/LibsqlUrlParserTest.java @@ -35,4 +35,46 @@ void testParseUrlWithNamespacePathAndQueryParams() throws SQLException { assertThat(config.authToken()).isEqualTo("secretToken"); assertThat(config.connectTimeout().toMillis()).isEqualTo(5000); } + + @Test + void testParseWebSocketUrlKeepsTheSchemeAndPath() throws SQLException { + LibsqlClientConfig config = LibsqlUrlParser.parse( + "jdbc:libsql:ws://localhost:8080/dev/tenant?authToken=secretToken", new Properties()); + + assertThat(config.url()).isEqualTo("ws://localhost:8080/dev/tenant"); + assertThat(config.authToken()).isEqualTo("secretToken"); + } + + @Test + void testParseWebSocketUrlFromSqlitePrefix() throws SQLException { + LibsqlClientConfig config = LibsqlUrlParser.parse( + "jdbc:sqlite:wss://example.com/dev/tenant", new Properties()); + + assertThat(config.url()).isEqualTo("wss://example.com/dev/tenant"); + } + + @Test + void testTlsUpgradesWebSocketToSecureWebSocket() throws SQLException { + LibsqlClientConfig config = LibsqlUrlParser.parse( + "jdbc:libsql:ws://localhost:8080?tls=true", new Properties()); + + assertThat(config.url()).isEqualTo("wss://localhost:8080"); + } + + @Test + void testTlsKeepsAlreadySecureSchemes() throws SQLException { + assertThat(LibsqlUrlParser.parse("jdbc:libsql:https://example.com?tls=true", new Properties()).url()) + .isEqualTo("https://example.com"); + assertThat(LibsqlUrlParser.parse("jdbc:libsql:wss://example.com?tls=true", new Properties()).url()) + .isEqualTo("wss://example.com"); + } + + @Test + void testAcceptsWebSocketUrlForms() { + assertThat(LibsqlUrlParser.acceptsUrl("jdbc:libsql:ws://localhost:8080")).isTrue(); + assertThat(LibsqlUrlParser.acceptsUrl("jdbc:libsql:wss://localhost:8080")).isTrue(); + assertThat(LibsqlUrlParser.acceptsUrl("jdbc:sqlite:ws://localhost:8080")).isTrue(); + assertThat(LibsqlUrlParser.acceptsUrl("jdbc:sqlite:wss://localhost:8080")).isTrue(); + assertThat(LibsqlUrlParser.acceptsUrl("jdbc:other:ws://localhost:8080")).isFalse(); + } }