Merge pull request #448 from tcheeric/codex/add-non-blocking-subscription-api

chore: tidy spring websocket client files

Author: tcheeric

.github/pull_request_template.md

@@ -1,5 +1,6 @@
-## Why now?
+## Summary
 <!-- Explain the problem, context, and why this change is needed. Link to the issue. -->
+
 Related issue: #____
 
 ## What changed?

.github/workflows/codex.yml

@@ -224,3 +224,4 @@ data
 
 # Original versions of merged files
 *.orig
+/.qodana/

.github/workflows/release-please.yml

@@ -19,6 +19,37 @@ See [`docs/CODEBASE_OVERVIEW.md`](docs/CODEBASE_OVERVIEW.md) for details about r
 ## Examples
 Examples are located in the [`nostr-java-examples`](./nostr-java-examples) module.
 
+- [`SpringSubscriptionExample`](nostr-java-examples/src/main/java/nostr/examples/SpringSubscriptionExample.java)
+  shows how to open a non-blocking `NostrSpringWebSocketClient` subscription and close it after a
+  fixed duration.
+
+## Streaming subscriptions
+
+The client and API layers expose a non-blocking streaming API for long-lived subscriptions. Use
+`NostrSpringWebSocketClient.subscribe` to open a REQ subscription and receive relay messages via a
+callback:
+
+```java
+Filters filters = new Filters(new KindFilter<>(Kind.TEXT_NOTE));
+AutoCloseable subscription =
+    client.subscribe(
+        filters,
+        "example-subscription",
+        message -> {
+          // handle EVENT/NOTICE payloads on your own executor to avoid blocking the socket thread
+        },
+        error -> log.warn("Subscription error", error));
+
+// ... keep the subscription open while processing events ...
+
+subscription.close(); // sends CLOSE to the relay and releases the underlying WebSocket
+```
+
+Subscriptions must be closed by the caller to ensure a CLOSE frame is sent to the relay and to free
+the dedicated WebSocket connection created for the REQ. Callbacks run on the WebSocket thread; for
+high-throughput feeds, hand off work to a queue or executor to provide backpressure and keep the
+socket responsive.
+
 ## Supported NIPs
 The API currently implements the following [NIPs](https://github.com/nostr-protocol/nips):
 - [NIP-1](https://github.com/nostr-protocol/nips/blob/master/01.md) - Basic protocol flow description

.gitignore

@@ -65,6 +65,14 @@ Abstraction over a WebSocket connection to a relay.
 ```java
 <T extends BaseMessage> List<String> send(T eventMessage) throws IOException
 List<String> send(String json) throws IOException
+AutoCloseable subscribe(String requestJson,
+                        Consumer<String> messageListener,
+                        Consumer<Throwable> errorListener,
+                        Runnable closeListener) throws IOException
+<T extends BaseMessage> AutoCloseable subscribe(T eventMessage,
+                                                Consumer<String> messageListener,
+                                                Consumer<Throwable> errorListener,
+                                                Runnable closeListener) throws IOException
 void close() throws IOException
 ```
 
@@ -75,6 +83,10 @@ Spring `TextWebSocketHandler` based implementation of `WebSocketClientIF`.
 public StandardWebSocketClient(String relayUri)
 public <T extends BaseMessage> List<String> send(T eventMessage) throws IOException
 public List<String> send(String json) throws IOException
+public AutoCloseable subscribe(String requestJson,
+                               Consumer<String> messageListener,
+                               Consumer<Throwable> errorListener,
+                               Runnable closeListener) throws IOException
 public void close() throws IOException
 ```
 
@@ -84,6 +96,14 @@ Wrapper that adds retry logic around a `WebSocketClientIF`.
 ```java
 public List<String> send(BaseMessage eventMessage) throws IOException
 public List<String> send(String json) throws IOException
+public AutoCloseable subscribe(BaseMessage requestMessage,
+                               Consumer<String> messageListener,
+                               Consumer<Throwable> errorListener,
+                               Runnable closeListener) throws IOException
+public AutoCloseable subscribe(String json,
+                               Consumer<String> messageListener,
+                               Consumer<Throwable> errorListener,
+                               Runnable closeListener) throws IOException
 public List<String> recover(IOException ex, String json) throws IOException
 public void close() throws IOException
 ```
@@ -95,12 +115,24 @@ High level client coordinating multiple relay connections and signing.
 public NostrIF setRelays(Map<String,String> relays)
 public List<String> sendEvent(IEvent event)
 public List<String> sendRequest(List<Filters> filters, String subscriptionId)
+public AutoCloseable subscribe(Filters filters, String subscriptionId, Consumer<String> listener)
+public AutoCloseable subscribe(Filters filters,
+                               String subscriptionId,
+                               Consumer<String> listener,
+                               Consumer<Throwable> errorListener)
 public NostrIF sign(Identity identity, ISignable signable)
 public boolean verify(GenericEvent event)
 public Map<String,String> getRelays()
 public void close()
 ```
 
+`subscribe` opens a dedicated WebSocket per relay, returns immediately, and streams raw relay
+messages to the provided listener. The returned `AutoCloseable` sends a `CLOSE` command and releases
+resources when invoked. Because callbacks execute on the WebSocket thread, delegate heavy
+processing to another executor to avoid stalling inbound traffic. The
+[`SpringSubscriptionExample`](../../nostr-java-examples/src/main/java/nostr/examples/SpringSubscriptionExample.java)
+demonstrates how to open a subscription and close it after a fixed duration.
+
 ### Configuration
 - `RetryConfig` – enables Spring Retry support.
 - `RelaysProperties` – maps relay names to URLs via configuration properties.

PR_DRAFT.md

@@ -3,6 +3,7 @@
 import java.io.IOException;
 import java.util.List;
 import java.util.Map;
+import java.util.function.Consumer;
 import lombok.NonNull;
 import nostr.base.IEvent;
 import nostr.base.ISignable;
@@ -90,6 +91,34 @@ List<String> sendRequest(
       @NonNull String subscriptionId,
       Map<String, String> relays);
 
+  /**
+   * Subscribe to a stream of events for the given filter on configured relays.
+   *
+   * @param filters the filter describing events to stream
+   * @param subscriptionId identifier for the subscription
+   * @param listener consumer invoked for each raw relay message
+   * @return a handle that cancels the subscription when closed
+   */
+  AutoCloseable subscribe(
+      @NonNull Filters filters,
+      @NonNull String subscriptionId,
+      @NonNull Consumer<String> listener);
+
+  /**
+   * Subscribe to a stream of events with custom error handling.
+   *
+   * @param filters the filter describing events to stream
+   * @param subscriptionId identifier for the subscription
+   * @param listener consumer invoked for each raw relay message
+   * @param errorListener optional consumer invoked when a transport error occurs
+   * @return a handle that cancels the subscription when closed
+   */
+  AutoCloseable subscribe(
+      @NonNull Filters filters,
+      @NonNull String subscriptionId,
+      @NonNull Consumer<String> listener,
+      Consumer<Throwable> errorListener);
+
   /**
    * Sign a signable object with the provided identity.
    *