chore: tidy spring websocket client files

README.md

@@ -19,6 +19,33 @@ 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.
 
+## 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

docs/reference/nostr-java-api.md

@@ -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,22 @@ 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.
+
 ### Configuration
 - `RetryConfig` – enables Spring Retry support.
 - `RelaysProperties` – maps relay names to URLs via configuration properties.

nostr-java-api/src/main/java/nostr/api/NostrIF.java

@@ -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.
    *

nostr-java-api/src/main/java/nostr/api/NostrSpringWebSocketClient.java

@@ -1,15 +1,18 @@
 package nostr.api;
 
 import java.io.IOException;
+import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutionException;
+import java.util.function.Consumer;
 import java.util.stream.Collectors;
 import lombok.Getter;
 import lombok.NoArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
 import lombok.NonNull;
 import nostr.api.service.NoteService;
 import nostr.api.service.impl.DefaultNoteService;
@@ -27,6 +30,7 @@
  * Default Nostr client using Spring WebSocket clients to send events and requests to relays.
  */
 @NoArgsConstructor
+@Slf4j
 public class NostrSpringWebSocketClient implements NostrIF {
   private final Map<String, WebSocketClientHandler> clientMap = new ConcurrentHashMap<>();
   @Getter private Identity sender;
@@ -106,7 +110,7 @@
   }
 
   @Override
   /**
    * Configure one or more relays by name and URI; creates client handlers lazily.
    */
   public NostrIF setRelays(@NonNull Map<String, String> relays) {
@@ -117,7 +121,7 @@
               try {
                 clientMap.putIfAbsent(
                     relayEntry.getKey(),
-                    new WebSocketClientHandler(relayEntry.getKey(), relayEntry.getValue()));
+                    newWebSocketClientHandler(relayEntry.getKey(), relayEntry.getValue()));
               } catch (ExecutionException | InterruptedException e) {
                 throw new RuntimeException("Failed to initialize WebSocket client handler", e);
               }
@@ -126,7 +130,7 @@
   }
 
   @Override
   /**
    * Send an event to all configured relays using the {@link NoteService}.
    */
   public List<String> sendEvent(@NonNull IEvent event) {
@@ -140,7 +144,7 @@
   }
 
   @Override
   /**
    * Send an event to the provided relays.
    */
   public List<String> sendEvent(@NonNull IEvent event, Map<String, String> relays) {
@@ -149,7 +153,7 @@
   }
 
   @Override
   /**
    * Send a REQ with a single filter to specific relays.
    */
   public List<String> sendRequest(
@@ -158,7 +162,7 @@
   }
 
   @Override
   /**
    * Send REQ with multiple filters to specific relays.
    */
   public List<String> sendRequest(
@@ -170,7 +174,7 @@
   }
 
   @Override
   /**
    * Send REQ with multiple filters to configured relays; flattens distinct responses.
    */
   public List<String> sendRequest(
@@ -200,7 +204,7 @@
   }
 
   @Override
   /**
    * Send a REQ with a single filter to configured relays using a per-subscription client.
    */
   public List<String> sendRequest(@NonNull Filters filters, @NonNull String subscriptionId) {
@@ -216,8 +220,78 @@
         .toList();
   }
 
+  @Override
+  public AutoCloseable subscribe(
+      @NonNull Filters filters,
+      @NonNull String subscriptionId,
+      @NonNull Consumer<String> listener) {
+    return subscribe(filters, subscriptionId, listener, null);
+  }
+
+  @Override
+  public AutoCloseable subscribe(
+      @NonNull Filters filters,
+      @NonNull String subscriptionId,
+      @NonNull Consumer<String> listener,
+      Consumer<Throwable> errorListener) {
+    Consumer<Throwable> safeError =
+        errorListener != null
+            ? errorListener
+            : throwable ->
+                log.warn(
+                    "Subscription error for {} on relays {}", subscriptionId, clientMap.keySet(),
+                    throwable);
+
+    List<AutoCloseable> handles = new ArrayList<>();
+    try {
+      clientMap.entrySet().stream()
+          .filter(entry -> !entry.getKey().contains(":"))
+          .map(Map.Entry::getValue)
+          .forEach(
+              handler -> {
+                AutoCloseable handle = handler.subscribe(filters, subscriptionId, listener, safeError);
+                handles.add(handle);
+              });
+    } catch (RuntimeException e) {
+      handles.forEach(
+          handle -> {
+            try {
+              handle.close();
+            } catch (Exception closeEx) {
+              safeError.accept(closeEx);
+            }
+          });
+      throw e;
+    }
+
+    return () -> {
+      IOException ioFailure = null;
+      Exception nonIoFailure = null;
+      for (AutoCloseable handle : handles) {
+        try {
+          handle.close();
+        } catch (IOException e) {
+          safeError.accept(e);
+          if (ioFailure == null) {
+            ioFailure = e;
+          }
+        } catch (Exception e) {
+          safeError.accept(e);
+          nonIoFailure = e;
+        }
+      }
+
+      if (ioFailure != null) {
+        throw ioFailure;
+      }
+      if (nonIoFailure != null) {
+        throw new IOException("Failed to close subscription", nonIoFailure);
+      }
+    };
+  }
+
   @Override
   /**
    * Sign a signable object with the provided identity.
    */
   public NostrIF sign(@NonNull Identity identity, @NonNull ISignable signable) {
@@ -226,7 +300,7 @@
   }
 
   @Override
   /**
    * Verify the Schnorr signature of a GenericEvent.
    */
   public boolean verify(@NonNull GenericEvent event) {
@@ -245,7 +319,7 @@
   }
 
   @Override
   /**
    * Return a copy of the current relay mapping (name -> URI).
    */
   public Map<String, String> getRelays() {