feat(client): add Hrana 3 JSON over WebSocket transport

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.
+ *
+ * <p>Both transports expose the same two execution styles. A <em>one-shot</em> 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 <em>on-stream</em> call reuses a lazily opened
+ * interactive stream, which is what a transaction needs: the server keeps the transaction open
+ * between calls.
+ *
+ * <p>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();
+}

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<String, HttpClient> 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);
     }