feat: implement PubSubClient + WorkflowClient (SSE subscribe, full HTTP)

Author: Heinrich Neb

src/main/java/dev/cachly/CachlyClient.java

@@ -579,7 +579,7 @@ public Iterator<String> streamGet(String key) {
      */
     public PubSubClient pubsub() {
         if (pubsubUrl == null) throw new CachlyException("pubsub() requires pubsubUrl.");
-        return new PubSubClient(pubsubUrl);
+        return new PubSubClient(pubsubUrl, HttpClient.newHttpClient(), MAPPER);
     }
 
     /**
@@ -588,7 +588,7 @@ public PubSubClient pubsub() {
      */
     public WorkflowClient workflow() {
         if (workflowUrl == null) throw new CachlyException("workflow() requires workflowUrl.");
-        return new WorkflowClient(workflowUrl);
+        return new WorkflowClient(workflowUrl, HttpClient.newHttpClient(), MAPPER);
     }
 
     /**

src/main/java/dev/cachly/PubSubClient.java

@@ -1,21 +1,201 @@
 package dev.cachly;
 
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+import java.io.BufferedReader;
+import java.io.InputStreamReader;
+import java.net.URI;
+import java.net.http.HttpClient;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Consumer;
+
 /**
- * Pub/Sub client for real-time messaging via the cachly API (feature U).
+ * Pub/Sub client for real-time messaging via the cachly API.
  * Obtain via {@link CachlyClient#pubsub()}.
- *
- * <p><em>Note:</em> Full streaming SSE subscription requires a separate implementation
- * or can be handled via server-sent events with an appropriate HTTP client.
  */
 public class PubSubClient {
 
-    private final String url;
+    // ── Inner types ───────────────────────────────────────────────────────────
 
-    public PubSubClient(String url) {
-        this.url = url;
+    /** Result of a publish operation. */
+    public static final class PublishResult {
+        public final String channel;
+        public final int    receivers;
+        public final String at;
+
+        public PublishResult(String channel, int receivers, String at) {
+            this.channel   = channel;
+            this.receivers = receivers;
+            this.at        = at;
+        }
     }
 
-    // TODO: Implement publish, subscribe, channels, stats methods
-    // For now, this is a placeholder to satisfy the compilation.
-}
+    /** Information about a single active channel. */
+    public static final class ChannelInfo {
+        public final String name;
+        public final int    subscribers;
+
+        public ChannelInfo(String name, int subscribers) {
+            this.name        = name;
+            this.subscribers = subscribers;
+        }
+    }
+
+    /** Aggregate pub/sub statistics. */
+    public static final class StatsResult {
+        public final int activeChannels;
+        public final int totalSubscribers;
+        public final int errors;
+
+        public StatsResult(int activeChannels, int totalSubscribers, int errors) {
+            this.activeChannels    = activeChannels;
+            this.totalSubscribers  = totalSubscribers;
+            this.errors            = errors;
+        }
+    }
+
+    /** A single message received via SSE subscription. */
+    public static final class Message {
+        public final String channel;
+        public final String message;
+        public final String at;
+
+        public Message(String channel, String message, String at) {
+            this.channel = channel;
+            this.message = message;
+            this.at      = at;
+        }
+    }
+
+    // ── Implementation ────────────────────────────────────────────────────────
+
+    private final String     pubsubUrl;
+    private final HttpClient http;
+    private final ObjectMapper mapper;
 
+    PubSubClient(String pubsubUrl, HttpClient http, ObjectMapper mapper) {
+        this.pubsubUrl = pubsubUrl;
+        this.http      = http;
+        this.mapper    = mapper;
+    }
+
+    /**
+     * Convenience constructor used when no shared HttpClient / ObjectMapper exists.
+     * @deprecated Prefer the constructor with shared HttpClient via {@link CachlyClient#pubsub()}.
+     */
+    PubSubClient(String pubsubUrl) {
+        this(pubsubUrl,
+             HttpClient.newBuilder().connectTimeout(Duration.ofSeconds(10)).build(),
+             new ObjectMapper());
+    }
+
+    private JsonNode httpSync(String method, String url, Object body) throws CachlyException {
+        try {
+            HttpRequest.Builder b = HttpRequest.newBuilder()
+                .uri(URI.create(url))
+                .timeout(Duration.ofSeconds(10))
+                .header("Accept", "application/json");
+            if (body != null) {
+                String json = mapper.writeValueAsString(body);
+                b.header("Content-Type", "application/json")
+                 .method(method, HttpRequest.BodyPublishers.ofString(json));
+            } else {
+                b.method(method, HttpRequest.BodyPublishers.noBody());
+            }
+            HttpResponse<String> resp = http.send(b.build(), HttpResponse.BodyHandlers.ofString());
+            if (resp.statusCode() >= 400)
+                throw new CachlyException("cachly pubsub " + method + " " + url + " → " + resp.statusCode());
+            String rb = resp.body();
+            return (rb == null || rb.isEmpty()) ? mapper.createObjectNode() : mapper.readTree(rb);
+        } catch (CachlyException e) {
+            throw e;
+        } catch (Exception e) {
+            throw new CachlyException("cachly pubsub request failed: " + method + " " + url, e);
+        }
+    }
+
+    /**
+     * Publish a message to a channel.
+     * {@code POST {pubsubUrl}/publish}
+     */
+    public PublishResult publish(String channel, String message) throws CachlyException {
+        JsonNode n = httpSync("POST", pubsubUrl + "/publish",
+                Map.of("channel", channel, "message", message));
+        return new PublishResult(
+                n.path("channel").asText(channel),
+                n.path("receivers").asInt(0),
+                n.path("at").asText(""));
+    }
+
+    /**
+     * Subscribe to one or more channels via SSE (blocking, calls {@code consumer} for each message).
+     * Returns when the stream closes or an error occurs.
+     * {@code POST {pubsubUrl}/subscribe}
+     */
+    public void subscribe(List<String> channels, Consumer<Message> consumer) throws CachlyException {
+        try {
+            String bodyJson = mapper.writeValueAsString(Map.of("channels", channels));
+            HttpRequest req = HttpRequest.newBuilder()
+                .uri(URI.create(pubsubUrl + "/subscribe"))
+                .header("Content-Type", "application/json")
+                .header("Accept", "text/event-stream")
+                .POST(HttpRequest.BodyPublishers.ofString(bodyJson))
+                .timeout(Duration.ofSeconds(300))
+                .build();
+            HttpResponse<java.io.InputStream> resp =
+                http.send(req, HttpResponse.BodyHandlers.ofInputStream());
+            if (resp.statusCode() >= 400)
+                throw new CachlyException("cachly pubsub subscribe → " + resp.statusCode());
+            try (BufferedReader reader = new BufferedReader(new InputStreamReader(resp.body()))) {
+                String line;
+                while ((line = reader.readLine()) != null) {
+                    if (!line.startsWith("data:")) continue;
+                    String data = line.substring(5).trim();
+                    if (data.isEmpty() || data.equals("{}")) continue;
+                    try {
+                        JsonNode node = mapper.readTree(data);
+                        consumer.accept(new Message(
+                                node.path("channel").asText(""),
+                                node.path("message").asText(""),
+                                node.path("at").asText("")));
+                    } catch (Exception ignored) { /* skip malformed events */ }
+                }
+            }
+        } catch (CachlyException e) {
+            throw e;
+        } catch (Exception e) {
+            throw new CachlyException("cachly pubsub subscribe failed", e);
+        }
+    }
+
+    /**
+     * List active channels.
+     * {@code GET {pubsubUrl}/channels}
+     */
+    public List<ChannelInfo> channels() throws CachlyException {
+        JsonNode n = httpSync("GET", pubsubUrl + "/channels", null);
+        List<ChannelInfo> result = new ArrayList<>();
+        JsonNode arr = n.isArray() ? n : n.path("channels");
+        for (JsonNode ch : arr)
+            result.add(new ChannelInfo(ch.path("name").asText(""), ch.path("subscribers").asInt(0)));
+        return result;
+    }
+
+    /**
+     * Pub/Sub aggregate statistics.
+     * {@code GET {pubsubUrl}/stats}
+     */
+    public StatsResult stats() throws CachlyException {
+        JsonNode n = httpSync("GET", pubsubUrl + "/stats", null);
+        return new StatsResult(
+                n.path("active_channels").asInt(0),
+                n.path("total_subscribers").asInt(0),
+                n.path("errors").asInt(0));
+    }
+}