Fix listener leaked per json-rpc connection

Fixes #2136
Fixes #1653

lib/src/main/java/org/asamk/signal/manager/MultiAccountManager.java

@@ -12,8 +12,12 @@ public interface MultiAccountManager extends AutoCloseable {
 
     void addOnManagerAddedHandler(Consumer<Manager> handler);
 
+    void removeOnManagerAddedHandler(Consumer<Manager> handler);
+
     void addOnManagerRemovedHandler(Consumer<Manager> handler);
 
+    void removeOnManagerRemovedHandler(Consumer<Manager> handler);
+
     Manager getManager(String phoneNumber);
 
     URI getNewProvisioningDeviceLinkUri() throws TimeoutException, IOException;

lib/src/main/java/org/asamk/signal/manager/internal/MultiAccountManagerImpl.java

@@ -68,6 +68,13 @@ public class MultiAccountManagerImpl implements MultiAccountManager {
         }
     }
 
+    @Override
+    public void removeOnManagerAddedHandler(final Consumer<Manager> handler) {
+        synchronized (onManagerAddedHandlers) {
+            onManagerAddedHandlers.remove(handler);
+        }
+    }
+
     void removeManager(final Manager m) {
         synchronized (managers) {
             if (!managers.remove(m)) {
@@ -88,6 +95,13 @@ public class MultiAccountManagerImpl implements MultiAccountManager {
         }
     }
 
+    @Override
+    public void removeOnManagerRemovedHandler(final Consumer<Manager> handler) {
+        synchronized (onManagerRemovedHandlers) {
+            onManagerRemovedHandlers.remove(handler);
+        }
+    }
+
     @Override
     public Manager getManager(final String identifier) {
         synchronized (managers) {

src/main/java/org/asamk/signal/dbus/DbusMultiAccountManagerImpl.java

@@ -58,6 +58,13 @@ public class DbusMultiAccountManagerImpl implements MultiAccountManager {
         }
     }
 
+    @Override
+    public void removeOnManagerAddedHandler(final Consumer<Manager> handler) {
+        synchronized (onManagerAddedHandlers) {
+            onManagerAddedHandlers.remove(handler);
+        }
+    }
+
     @Override
     public void addOnManagerRemovedHandler(final Consumer<Manager> handler) {
         synchronized (onManagerRemovedHandlers) {
@@ -65,6 +72,13 @@ public class DbusMultiAccountManagerImpl implements MultiAccountManager {
         }
     }
 
+    @Override
+    public void removeOnManagerRemovedHandler(final Consumer<Manager> handler) {
+        synchronized (onManagerRemovedHandlers) {
+            onManagerRemovedHandlers.remove(handler);
+        }
+    }
+
     @Override
     public Manager getManager(final String identifier) {
         return new DbusManagerImpl(getRemoteObject(signalControl.getAccount(identifier), Signal.class),

src/main/java/org/asamk/signal/jsonrpc/SignalJsonRpcDispatcherHandler.java

@@ -31,6 +31,7 @@ import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Consumer;
 import java.util.function.Supplier;
 import java.util.stream.Collectors;
 
@@ -61,17 +62,31 @@ public class SignalJsonRpcDispatcherHandler {
     public void handleConnection(final MultiAccountManager c) {
         this.commandHandler = new SignalJsonRpcCommandHandler(c, this::getCommand);
 
+        final Consumer<Manager> onManagerAddedReceiveHandler = m -> subscribeReceive(m, true);
+        final Consumer<Manager> onManagerRemovedReceiveHandler = this::unsubscribeReceive;
+        final Consumer<Manager> onManagerAddedReceiveSubscriptionsHandler = m -> receiveHandlers
+            .forEach((subscriptionId, handlers) -> handlers.add(createReceiveHandler(m, subscriptionId, false)));
+        final Consumer<Manager> onManagerAddedCallEventSubscriptionsHandler = m -> callEventHandlers
+            .forEach((subscriptionId, handlers) -> handlers.add(createCallEventHandler(m, subscriptionId)));
+
         if (!noReceiveOnStart) {
             this.subscribeReceive(c.getManagers(), true);
-            c.addOnManagerAddedHandler(m -> subscribeReceive(m, true));
-            c.addOnManagerRemovedHandler(this::unsubscribeReceive);
+            c.addOnManagerAddedHandler(onManagerAddedReceiveHandler);
+            c.addOnManagerRemovedHandler(onManagerRemovedReceiveHandler);
         }
-        c.addOnManagerAddedHandler(m -> receiveHandlers.forEach((subscriptionId, handlers) -> handlers.add(
-                createReceiveHandler(m, subscriptionId, false))));
-        c.addOnManagerAddedHandler(m -> callEventHandlers.forEach((subscriptionId, handlers) -> handlers.add(
-                createCallEventHandler(m, subscriptionId))));
+        c.addOnManagerAddedHandler(onManagerAddedReceiveSubscriptionsHandler);
+        c.addOnManagerAddedHandler(onManagerAddedCallEventSubscriptionsHandler);
 
-        handleConnection();
+        try {
+            handleConnection();
+        } finally {
+            if (!noReceiveOnStart) {
+                c.removeOnManagerAddedHandler(onManagerAddedReceiveHandler);
+                c.removeOnManagerRemovedHandler(onManagerRemovedReceiveHandler);
+            }
+            c.removeOnManagerAddedHandler(onManagerAddedReceiveSubscriptionsHandler);
+            c.removeOnManagerAddedHandler(onManagerAddedCallEventSubscriptionsHandler);
+        }
     }
 
     public void handleConnection(final Manager m) {

src/test/java/org/asamk/signal/jsonrpc/SubscribeCallEventsTest.java

@@ -67,6 +67,8 @@ class SubscribeCallEventsTest {
 
         final List<Manager> managers;
         final List<Consumer<Manager>> addedHandlers = new ArrayList<>();
+        final List<Consumer<Manager>> removedHandlers = new ArrayList<>();
+        int addedHandlerRegistrationCount;
 
         StubMultiAccountManager(List<Manager> managers) {
             this.managers = new ArrayList<>(managers);
@@ -79,11 +81,23 @@ class SubscribeCallEventsTest {
 
         @Override
         public void addOnManagerAddedHandler(Consumer<Manager> handler) {
+            addedHandlerRegistrationCount++;
             addedHandlers.add(handler);
         }
 
+        @Override
+        public void removeOnManagerAddedHandler(Consumer<Manager> handler) {
+            addedHandlers.remove(handler);
+        }
+
         @Override
         public void addOnManagerRemovedHandler(Consumer<Manager> handler) {
+            removedHandlers.add(handler);
+        }
+
+        @Override
+        public void removeOnManagerRemovedHandler(Consumer<Manager> handler) {
+            removedHandlers.remove(handler);
         }
 
         @Override
@@ -223,8 +237,9 @@ class SubscribeCallEventsTest {
 
         assertEquals(1, manager1.state().addCallEventListenerCount.get(), "manager1 should have one listener");
         assertEquals(1, manager2.state().addCallEventListenerCount.get(), "manager2 should have one listener");
-        // Also registers an onManagerAdded handler for receive and one for call events
-        assertEquals(2, multi.addedHandlers.size(), "should register onManagerAdded handlers");
+        assertEquals(2, multi.addedHandlerRegistrationCount, "should register onManagerAdded handlers");
+        assertEquals(0, multi.addedHandlers.size(), "should remove onManagerAdded handlers after connection closes");
+        assertEquals(0, multi.removedHandlers.size(), "should not register onManagerRemoved handlers in manual mode");
     }
 
     @Test
@@ -263,4 +278,17 @@ class SubscribeCallEventsTest {
                 manager1.state().addCallEventListenerCount.get(),
                 "call events should not be auto-subscribed in multi mode");
     }
+
+    @Test
+    void multiAccountConnectionRemovesAllManagerHandlersWhenReceiveIsEnabled() {
+        var manager = createManager("+15551111111");
+        var multi = new StubMultiAccountManager(List.of(manager.manager()));
+        var handler = new SignalJsonRpcDispatcherHandler(new CapturingJsonWriter(), () -> null, false);
+
+        handler.handleConnection(multi);
+
+        assertEquals(3, multi.addedHandlerRegistrationCount);
+        assertEquals(0, multi.addedHandlers.size(), "should remove all onManagerAdded handlers after close");
+        assertEquals(0, multi.removedHandlers.size(), "should remove onManagerRemoved handler after close");
+    }
 }