ensure cormorant responses and notifications are always serialized per client connection

### src/main/java/com/sparrowwallet/sparrow/net/cormorant/electrum/ElectrumNotificationTransport.java
@@ -2,24 +2,16 @@
 
 import com.github.arteam.simplejsonrpc.client.Transport;
 
-import java.io.IOException;
-import java.io.OutputStreamWriter;
-import java.io.PrintWriter;
-import java.net.Socket;
-import java.nio.charset.StandardCharsets;
-
 public class ElectrumNotificationTransport implements Transport {
-    private final Socket clientSocket;
+    private final RequestHandler requestHandler;
 
-    public ElectrumNotificationTransport(Socket clientSocket) {
-        this.clientSocket = clientSocket;
+    public ElectrumNotificationTransport(RequestHandler requestHandler) {
+        this.requestHandler = requestHandler;
     }
 
     @Override
-    public String pass(String request) throws IOException {
-        PrintWriter out = new PrintWriter(new OutputStreamWriter(clientSocket.getOutputStream(), StandardCharsets.UTF_8));
-        out.println(request);
-        out.flush();
+    public String pass(String request) {
+        requestHandler.send(request);
 
         return "{\"result\":{},\"error\":null,\"id\":1}";
     }

### src/main/java/com/sparrowwallet/sparrow/net/cormorant/electrum/RequestHandler.java
@@ -11,48 +11,53 @@
 import java.io.*;
 import java.net.Socket;
 import java.nio.charset.StandardCharsets;
-import java.util.HashSet;
 import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
 
 public class RequestHandler implements Runnable {
     private static final Logger log = LoggerFactory.getLogger(RequestHandler.class);
     private final Socket clientSocket;
     private final ElectrumServerService electrumServerService;
     private final JsonRpcServer rpcServer = new JsonRpcServer();
+    private volatile PrintWriter out;
 
-    private boolean headersSubscribed;
-    private final Set<String> scriptHashesSubscribed = new HashSet<>();
+    private volatile boolean headersSubscribed;
+    private final Set<String> scriptHashesSubscribed = ConcurrentHashMap.newKeySet();
 
     public RequestHandler(Socket clientSocket, BitcoindClient bitcoindClient, int electrumPort) {
         this.clientSocket = clientSocket;
         this.electrumServerService = new ElectrumServerService(bitcoindClient, this, electrumPort);
     }
 
     public void run() {
-        Cormorant.getEventBus().register(this);
-
         try {
             InputStream input  = clientSocket.getInputStream();
             BufferedReader reader = new BufferedReader(new InputStreamReader(input, StandardCharsets.UTF_8));
 
             OutputStream output = clientSocket.getOutputStream();
-            PrintWriter out = new PrintWriter(new BufferedWriter(new OutputStreamWriter(output, StandardCharsets.UTF_8)));
+            out = new PrintWriter(new BufferedWriter(new OutputStreamWriter(output, StandardCharsets.UTF_8)));
 
-            while(true) {
-                String request = reader.readLine();
-                if(request == null) {
-                    break;
-                }
+            Cormorant.getEventBus().register(this);
+            try {
+                while(true) {
+                    String request = reader.readLine();
+                    if(request == null) {
+                        break;
+                    }
 
-                String response = rpcServer.handle(request, electrumServerService);
-                out.println(response);
-                out.flush();
+                    send(rpcServer.handle(request, electrumServerService));
+                }
+            } finally {
+                Cormorant.getEventBus().unregister(this);
             }
         } catch(IOException e) {
             log.error("Could not communicate with client socket", e);
         }
+    }
 
-        Cormorant.getEventBus().unregister(this);
+    synchronized void send(String message) {
+        out.println(message);
+        out.flush();
     }
 
     public void setHeadersSubscribed(boolean headersSubscribed) {
@@ -70,7 +75,7 @@ public boolean isScriptHashSubscribed(String scriptHash) {
     @Subscribe
     public void newBlock(ElectrumBlockHeader electrumBlockHeader) {
         if(headersSubscribed) {
-            ElectrumNotificationTransport electrumNotificationTransport = new ElectrumNotificationTransport(clientSocket);
+            ElectrumNotificationTransport electrumNotificationTransport = new ElectrumNotificationTransport(this);
             JsonRpcClient jsonRpcClient = new JsonRpcClient(electrumNotificationTransport);
             jsonRpcClient.onDemand(ElectrumNotificationService.class).notifyHeaders(electrumBlockHeader);
         }
@@ -79,7 +84,7 @@ public void newBlock(ElectrumBlockHeader electrumBlockHeader) {
     @Subscribe
     public void scriptHashStatus(ScriptHashStatus scriptHashStatus) {
         if(isScriptHashSubscribed(scriptHashStatus.scriptHash())) {
-            ElectrumNotificationTransport electrumNotificationTransport = new ElectrumNotificationTransport(clientSocket);
+            ElectrumNotificationTransport electrumNotificationTransport = new ElectrumNotificationTransport(this);
             JsonRpcClient jsonRpcClient = new JsonRpcClient(electrumNotificationTransport);
             jsonRpcClient.onDemand(ElectrumNotificationService.class).notifyScriptHash(scriptHashStatus.scriptHash(), scriptHashStatus.status());
         }

### src/test/java/com/sparrowwallet/sparrow/net/cormorant/electrum/RequestHandlerTest.java
@@ -0,0 +1,119 @@
+package com.sparrowwallet.sparrow.net.cormorant.electrum;
+
+import org.junit.jupiter.api.Test;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.net.Socket;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.concurrent.*;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+public class RequestHandlerTest {
+    private static final String SCRIPT_HASH = "a".repeat(64);
+
+    @Test
+    public void testNotificationWaitsForResponseInProgress() throws Exception {
+        CountDownLatch readerStarted = new CountDownLatch(1);
+        CountDownLatch endOfInput = new CountDownLatch(1);
+        CountDownLatch firstChunkWritten = new CountDownLatch(1);
+        CountDownLatch releaseFirstChunk = new CountDownLatch(1);
+        ByteArrayOutputStream written = new ByteArrayOutputStream();
+
+        InputStream input = new InputStream() {
+            @Override
+            public int read() throws IOException {
+                return read(new byte[1], 0, 1);
+            }
+
+            @Override
+            public int read(byte[] b, int off, int len) throws IOException {
+                readerStarted.countDown();
+                try {
+                    endOfInput.await(10, TimeUnit.SECONDS);
+                } catch(InterruptedException e) {
+                    Thread.currentThread().interrupt();
+                }
+                return -1;
+            }
+        };
+
+        //Holds the first write, which is the first chunk of a response too long for the writer's buffer, until released
+        OutputStream output = new OutputStream() {
+            private boolean paused;
+
+            @Override
+            public void write(int b) throws IOException {
+                write(new byte[] {(byte)b}, 0, 1);
+            }
+
+            @Override
+            public void write(byte[] b, int off, int len) throws IOException {
+                boolean pause;
+                synchronized(written) {
+                    written.write(b, off, len);
+                    pause = !paused;
+                    paused = true;
+                }
+                if(pause) {
+                    firstChunkWritten.countDown();
+                    try {
+                        releaseFirstChunk.await(10, TimeUnit.SECONDS);
+                    } catch(InterruptedException e) {
+                        Thread.currentThread().interrupt();
+                    }
+                }
+            }
+        };
+
+        Socket socket = new Socket() {
+            @Override
+            public InputStream getInputStream() {
+                return input;
+            }
+
+            @Override
+            public OutputStream getOutputStream() {
+                return output;
+            }
+        };
+
+        RequestHandler requestHandler = new RequestHandler(socket, null, 0);
+        requestHandler.subscribeScriptHash(SCRIPT_HASH);
+        String response = "{\"jsonrpc\":\"2.0\",\"id\":1,\"result\":\"" + "0".repeat(20000) + "\"}";
+
+        ExecutorService executor = Executors.newFixedThreadPool(3);
+        try {
+            Future<?> handler = executor.submit(requestHandler);
+            assertTrue(readerStarted.await(5, TimeUnit.SECONDS), "The handler must reach its read loop");
+
+            Future<?> responseSent = executor.submit(() -> requestHandler.send(response));
+            assertTrue(firstChunkWritten.await(5, TimeUnit.SECONDS), "The response must begin writing");
+
+            //The polling thread posts status notifications while a response is being written
+            Future<?> notificationSent = executor.submit(() -> requestHandler.scriptHashStatus(new ScriptHashStatus(SCRIPT_HASH, "status")));
+            assertThrows(TimeoutException.class, () -> notificationSent.get(200, TimeUnit.MILLISECONDS), "A notification must wait for the response being written");
+
+            releaseFirstChunk.countDown();
+            responseSent.get(5, TimeUnit.SECONDS);
+            notificationSent.get(5, TimeUnit.SECONDS);
+
+            endOfInput.countDown();
+            handler.get(5, TimeUnit.SECONDS);
+
+            List<String> lines = written.toString(StandardCharsets.UTF_8).lines().toList();
+            assertEquals(2, lines.size(), "The response and the notification must each arrive as one whole line");
+            assertEquals(response, lines.get(0));
+            assertTrue(lines.get(1).contains("\"blockchain.scripthash.subscribe\""));
+            assertTrue(lines.get(1).contains(SCRIPT_HASH));
+        } finally {
+            releaseFirstChunk.countDown();
+            endOfInput.countDown();
+            executor.shutdownNow();
+        }
+    }
+}