Reexport dbus objects when self number changes

[signal-cli] / src / main / java / org / asamk / signal / commands / DaemonCommand.java

diff --git a/src/main/java/org/asamk/signal/commands/DaemonCommand.java b/src/main/java/org/asamk/signal/commands/DaemonCommand.java
index 5e7156474d74f27c8703b412e700891b2e14c6fc..79e0cd82965ab20f240afe14adf5385b5aae11a7 100644 (file)
--- a/src/main/java/org/asamk/signal/commands/DaemonCommand.java
+++ b/src/main/java/org/asamk/signal/commands/DaemonCommand.java
@@ -5,20 +5,21 @@ import net.sourceforge.argparse4j.inf.Namespace;
 import net.sourceforge.argparse4j.inf.Subparser;
 
 import org.asamk.signal.DbusConfig;
 import net.sourceforge.argparse4j.inf.Subparser;
 
 import org.asamk.signal.DbusConfig;

-import org.asamk.signal.JsonReceiveMessageHandler;
-import org.asamk.signal.JsonWriter;
-import org.asamk.signal.JsonWriterImpl;

 import org.asamk.signal.OutputType;
 import org.asamk.signal.OutputType;

-import org.asamk.signal.OutputWriter;
-import org.asamk.signal.PlainTextWriter;

 import org.asamk.signal.ReceiveMessageHandler;
 import org.asamk.signal.commands.exceptions.CommandException;
 import org.asamk.signal.commands.exceptions.IOErrorException;
 import org.asamk.signal.commands.exceptions.UnexpectedErrorException;
 import org.asamk.signal.dbus.DbusSignalControlImpl;
 import org.asamk.signal.dbus.DbusSignalImpl;
 import org.asamk.signal.ReceiveMessageHandler;
 import org.asamk.signal.commands.exceptions.CommandException;
 import org.asamk.signal.commands.exceptions.IOErrorException;
 import org.asamk.signal.commands.exceptions.UnexpectedErrorException;
 import org.asamk.signal.dbus.DbusSignalControlImpl;
 import org.asamk.signal.dbus.DbusSignalImpl;

+import org.asamk.signal.json.JsonReceiveMessageHandler;

 import org.asamk.signal.jsonrpc.SignalJsonRpcDispatcherHandler;
 import org.asamk.signal.manager.Manager;
 import org.asamk.signal.jsonrpc.SignalJsonRpcDispatcherHandler;
 import org.asamk.signal.manager.Manager;

+import org.asamk.signal.manager.MultiAccountManager;
+import org.asamk.signal.output.JsonWriter;
+import org.asamk.signal.output.JsonWriterImpl;
+import org.asamk.signal.output.OutputWriter;
+import org.asamk.signal.output.PlainTextWriter;

 import org.asamk.signal.util.IOUtils;
 import org.freedesktop.dbus.connections.impl.DBusConnection;
 import org.freedesktop.dbus.exceptions.DBusException;
 import org.asamk.signal.util.IOUtils;
 import org.freedesktop.dbus.connections.impl.DBusConnection;
 import org.freedesktop.dbus.exceptions.DBusException;


@@ -35,8 +36,8 @@ import java.nio.channels.SocketChannel;
 import java.nio.charset.StandardCharsets;
 import java.util.List;
 import java.util.Objects;
 import java.nio.charset.StandardCharsets;
 import java.util.List;
 import java.util.Objects;

+import java.util.concurrent.atomic.AtomicInteger;

 import java.util.function.Consumer;
 import java.util.function.Consumer;

-import java.util.stream.Collectors;

 
 public class DaemonCommand implements MultiLocalCommand, LocalCommand {
 
 
 public class DaemonCommand implements MultiLocalCommand, LocalCommand {
 


@@ -131,6 +132,12 @@ public class DaemonCommand implements MultiLocalCommand, LocalCommand {
             runDbusSingleAccount(m, false, receiveMode != ReceiveMode.ON_START);
         }
 
             runDbusSingleAccount(m, false, receiveMode != ReceiveMode.ON_START);
         }
 

+        m.addClosedListener(() -> {
+            synchronized (this) {
+                notifyAll();
+            }
+        });
+

         synchronized (this) {
             try {
                 wait();
         synchronized (this) {
             try {
                 wait();


@@ -141,14 +148,14 @@ public class DaemonCommand implements MultiLocalCommand, LocalCommand {
 
     @Override
     public void handleCommand(
 
     @Override
     public void handleCommand(

-            final Namespace ns, final SignalCreator c, final OutputWriter outputWriter
+            final Namespace ns, final MultiAccountManager c, final OutputWriter outputWriter

     ) throws CommandException {
         logger.info("Starting daemon in multi-account mode");
         final var noReceiveStdOut = Boolean.TRUE.equals(ns.getBoolean("no-receive-stdout"));
         final var receiveMode = ns.<ReceiveMode>get("receive-mode");
         final var ignoreAttachments = Boolean.TRUE.equals(ns.getBoolean("ignore-attachments"));
 
     ) throws CommandException {
         logger.info("Starting daemon in multi-account mode");
         final var noReceiveStdOut = Boolean.TRUE.equals(ns.getBoolean("no-receive-stdout"));
         final var receiveMode = ns.<ReceiveMode>get("receive-mode");
         final var ignoreAttachments = Boolean.TRUE.equals(ns.getBoolean("ignore-attachments"));
 

-        c.getAccountNumbers().stream().map(c::getManager).filter(Objects::nonNull).forEach(m -> {
+        c.getManagers().forEach(m -> {

             m.setIgnoreAttachments(ignoreAttachments);
             addDefaultReceiveHandler(m, noReceiveStdOut ? null : outputWriter, receiveMode != ReceiveMode.ON_START);
         });
             m.setIgnoreAttachments(ignoreAttachments);
             addDefaultReceiveHandler(m, noReceiveStdOut ? null : outputWriter, receiveMode != ReceiveMode.ON_START);
         });


@@ -220,7 +227,7 @@ public class DaemonCommand implements MultiLocalCommand, LocalCommand {
     }
 
     private void runSocketMultiAccount(
     }
 
     private void runSocketMultiAccount(

-            final SignalCreator c, final ServerSocketChannel serverChannel, final boolean noReceiveOnStart
+            final MultiAccountManager c, final ServerSocketChannel serverChannel, final boolean noReceiveOnStart

     ) {
         runSocket(serverChannel, channel -> {
             final var handler = getSignalJsonRpcDispatcherHandler(channel, noReceiveOnStart);
     ) {
         runSocket(serverChannel, channel -> {
             final var handler = getSignalJsonRpcDispatcherHandler(channel, noReceiveOnStart);


@@ -228,9 +235,10 @@ public class DaemonCommand implements MultiLocalCommand, LocalCommand {
         });
     }
 
         });
     }
 

+    private static final AtomicInteger threadNumber = new AtomicInteger(0);
+

     private void runSocket(final ServerSocketChannel serverChannel, Consumer<SocketChannel> socketHandler) {
     private void runSocket(final ServerSocketChannel serverChannel, Consumer<SocketChannel> socketHandler) {

-        final var mainThread = Thread.currentThread();
-        new Thread(() -> {
+        final var thread = new Thread(() -> {

             while (true) {
                 final SocketChannel channel;
                 final String clientString;
             while (true) {
                 final SocketChannel channel;
                 final String clientString;


@@ -240,19 +248,25 @@ public class DaemonCommand implements MultiLocalCommand, LocalCommand {
                     logger.info("Accepted new client: " + clientString);
                 } catch (IOException e) {
                     logger.error("Failed to accept new socket connection", e);
                     logger.info("Accepted new client: " + clientString);
                 } catch (IOException e) {
                     logger.error("Failed to accept new socket connection", e);

-                    mainThread.notifyAll();
+                    synchronized (this) {
+                        notifyAll();
+                    }

                     break;
                 }
                     break;
                 }

-                new Thread(() -> {
+                final var connectionThread = new Thread(() -> {

                     try (final var c = channel) {
                         socketHandler.accept(c);
                         logger.info("Connection closed: " + clientString);
                     } catch (IOException e) {
                         logger.warn("Failed to close channel", e);
                     }
                     try (final var c = channel) {
                         socketHandler.accept(c);
                         logger.info("Connection closed: " + clientString);
                     } catch (IOException e) {
                         logger.warn("Failed to close channel", e);
                     }

-                }).start();
+                });
+                connectionThread.setName("daemon-connection-" + threadNumber.getAndIncrement());
+                connectionThread.start();

             }
             }

-        }).start();
+        });
+        thread.setName("daemon-listener");
+        thread.start();

     }
 
     private SignalJsonRpcDispatcherHandler getSignalJsonRpcDispatcherHandler(
     }
 
     private SignalJsonRpcDispatcherHandler getSignalJsonRpcDispatcherHandler(


@@ -276,7 +290,7 @@ public class DaemonCommand implements MultiLocalCommand, LocalCommand {
     }
 
     private void runDbusMultiAccount(
     }
 
     private void runDbusMultiAccount(

-            final SignalCreator c, final boolean noReceiveOnStart, final boolean isDbusSystem
+            final MultiAccountManager c, final boolean noReceiveOnStart, final boolean isDbusSystem

     ) throws UnexpectedErrorException {
         runDbus(isDbusSystem, (connection, objectPath) -> {
             final var signalControl = new DbusSignalControlImpl(c, objectPath);
     ) throws UnexpectedErrorException {
         runDbus(isDbusSystem, (connection, objectPath) -> {
             final var signalControl = new DbusSignalControlImpl(c, objectPath);


@@ -291,14 +305,22 @@ public class DaemonCommand implements MultiLocalCommand, LocalCommand {
                     }
                 }
             });
                     }
                 }
             });

+            c.addOnManagerRemovedHandler(m -> {
+                final var path = DbusConfig.getObjectPath(m.getSelfNumber());
+                try {
+                    final var object = connection.getExportedObject(null, path);
+                    if (object instanceof DbusSignalImpl dbusSignal) {
+                        dbusSignal.close();
+                    }
+                } catch (DBusException ignored) {
+                }
+            });

 
 

-            final var initThreads = c.getAccountNumbers()
+            final var initThreads = c.getManagers()

                     .stream()
                     .stream()

-                    .map(c::getManager)
-                    .filter(Objects::nonNull)

                     .map(m -> exportMultiAccountManager(connection, m, noReceiveOnStart))
                     .filter(Objects::nonNull)
                     .map(m -> exportMultiAccountManager(connection, m, noReceiveOnStart))
                     .filter(Objects::nonNull)

-                    .collect(Collectors.toList());
+                    .toList();

 
             for (var t : initThreads) {
                 try {
 
             for (var t : initThreads) {
                 try {


@@ -347,8 +369,8 @@ public class DaemonCommand implements MultiLocalCommand, LocalCommand {
             final DBusConnection conn, final String objectPath, final Manager m, final boolean noReceiveOnStart
     ) throws DBusException {
         final var signal = new DbusSignalImpl(m, conn, objectPath, noReceiveOnStart);
             final DBusConnection conn, final String objectPath, final Manager m, final boolean noReceiveOnStart
     ) throws DBusException {
         final var signal = new DbusSignalImpl(m, conn, objectPath, noReceiveOnStart);

-        conn.exportObject(signal);

         final var initThread = new Thread(signal::initObjects);
         final var initThread = new Thread(signal::initObjects);

+        initThread.setName("dbus-init");

         initThread.start();
 
         logger.debug("Exported dbus object: " + objectPath);
         initThread.start();
 
         logger.debug("Exported dbus object: " + objectPath);