mirror of
https://github.com/AsamK/signal-cli
synced 2025-08-29 10:30:38 +00:00
Refactor DaemonCommand
This commit is contained in:
AsamK
parent
85b0647a3e
commit
1e35ac380e
1 changed files with 225 additions and 217 deletions
|
|
@ -18,7 +18,6 @@ import org.asamk.signal.json.JsonReceiveMessageHandler;
|
|||
import org.asamk.signal.jsonrpc.SignalJsonRpcDispatcherHandler;
|
||||
import org.asamk.signal.manager.Manager;
|
||||
import org.asamk.signal.manager.MultiAccountManager;
|
||||
import org.asamk.signal.manager.api.ReceiveConfig;
|
||||
import org.asamk.signal.output.JsonWriter;
|
||||
import org.asamk.signal.output.JsonWriterImpl;
|
||||
import org.asamk.signal.output.OutputWriter;
|
||||
|
|
@ -32,6 +31,7 @@ import org.slf4j.LoggerFactory;
|
|||
|
||||
import java.io.File;
|
||||
import java.io.IOException;
|
||||
import java.net.InetSocketAddress;
|
||||
import java.net.UnixDomainSocketAddress;
|
||||
import java.nio.channels.Channel;
|
||||
import java.nio.channels.Channels;
|
||||
|
|
@ -43,6 +43,8 @@ import java.util.concurrent.Executors;
|
|||
import java.util.concurrent.atomic.AtomicInteger;
|
||||
import java.util.function.Consumer;
|
||||
|
||||
import static org.asamk.signal.util.CommandUtil.getReceiveConfig;
|
||||
|
||||
public class DaemonCommand implements MultiLocalCommand, LocalCommand {
|
||||
|
||||
private final static Logger logger = LoggerFactory.getLogger(DaemonCommand.class);
|
||||
|
|
@ -105,59 +107,13 @@ public class DaemonCommand implements MultiLocalCommand, LocalCommand {
|
|||
logger.info("Starting daemon in single-account mode for " + m.getSelfNumber());
|
||||
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"));
|
||||
final var ignoreStories = Boolean.TRUE.equals(ns.getBoolean("ignore-stories"));
|
||||
final var sendReadReceipts = Boolean.TRUE.equals(ns.getBoolean("send-read-receipts"));
|
||||
final var receiveConfig = getReceiveConfig(ns);
|
||||
|
||||
m.setReceiveConfig(new ReceiveConfig(ignoreAttachments, ignoreStories, sendReadReceipts));
|
||||
m.setReceiveConfig(receiveConfig);
|
||||
addDefaultReceiveHandler(m, noReceiveStdOut ? null : outputWriter, receiveMode != ReceiveMode.ON_START);
|
||||
|
||||
final Channel inheritedChannel;
|
||||
try {
|
||||
inheritedChannel = System.inheritedChannel();
|
||||
if (inheritedChannel instanceof ServerSocketChannel serverChannel) {
|
||||
logger.info("Using inherited socket: " + serverChannel.getLocalAddress());
|
||||
runSocketSingleAccount(m, serverChannel, receiveMode == ReceiveMode.MANUAL);
|
||||
}
|
||||
} catch (IOException e) {
|
||||
throw new IOErrorException("Failed to use inherited socket", e);
|
||||
}
|
||||
final var socketFile = ns.<File>get("socket");
|
||||
if (socketFile != null) {
|
||||
final var address = UnixDomainSocketAddress.of(socketFile.toPath());
|
||||
final var serverChannel = IOUtils.bindSocket(address);
|
||||
runSocketSingleAccount(m, serverChannel, receiveMode == ReceiveMode.MANUAL);
|
||||
}
|
||||
final var tcpAddress = ns.getString("tcp");
|
||||
if (tcpAddress != null) {
|
||||
final var address = IOUtils.parseInetSocketAddress(tcpAddress);
|
||||
final var serverChannel = IOUtils.bindSocket(address);
|
||||
runSocketSingleAccount(m, serverChannel, receiveMode == ReceiveMode.MANUAL);
|
||||
}
|
||||
final var httpAddress = ns.getString("http");
|
||||
if (httpAddress != null) {
|
||||
final var address = IOUtils.parseInetSocketAddress(httpAddress);
|
||||
final var handler = new HttpServerHandler(address, m);
|
||||
try {
|
||||
handler.init();
|
||||
} catch (IOException ex) {
|
||||
throw new IOErrorException("Failed to initialize HTTP Server", ex);
|
||||
}
|
||||
}
|
||||
final var isDbusSystem = Boolean.TRUE.equals(ns.getBoolean("dbus-system"));
|
||||
if (isDbusSystem) {
|
||||
runDbusSingleAccount(m, true, receiveMode != ReceiveMode.ON_START);
|
||||
}
|
||||
final var isDbusSession = Boolean.TRUE.equals(ns.getBoolean("dbus"));
|
||||
if (isDbusSession || (
|
||||
!isDbusSystem
|
||||
&& socketFile == null
|
||||
&& tcpAddress == null
|
||||
&& httpAddress == null
|
||||
&& !(inheritedChannel instanceof ServerSocketChannel)
|
||||
)) {
|
||||
runDbusSingleAccount(m, false, receiveMode != ReceiveMode.ON_START);
|
||||
}
|
||||
try (final var daemonHandler = new SingleAccountDaemonHandler(m, receiveMode)) {
|
||||
setup(ns, daemonHandler);
|
||||
|
||||
m.addClosedListener(() -> {
|
||||
synchronized (this) {
|
||||
|
|
@ -172,6 +128,7 @@ public class DaemonCommand implements MultiLocalCommand, LocalCommand {
|
|||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void handleCommand(
|
||||
|
|
@ -180,11 +137,7 @@ public class DaemonCommand implements MultiLocalCommand, LocalCommand {
|
|||
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"));
|
||||
final var ignoreStories = Boolean.TRUE.equals(ns.getBoolean("ignore-stories"));
|
||||
final var sendReadReceipts = Boolean.TRUE.equals(ns.getBoolean("send-read-receipts"));
|
||||
|
||||
final var receiveConfig = new ReceiveConfig(ignoreAttachments, ignoreStories, sendReadReceipts);
|
||||
final var receiveConfig = getReceiveConfig(ns);
|
||||
c.getManagers().forEach(m -> {
|
||||
m.setReceiveConfig(receiveConfig);
|
||||
addDefaultReceiveHandler(m, noReceiveStdOut ? null : outputWriter, receiveMode != ReceiveMode.ON_START);
|
||||
|
|
@ -194,42 +147,55 @@ public class DaemonCommand implements MultiLocalCommand, LocalCommand {
|
|||
addDefaultReceiveHandler(m, noReceiveStdOut ? null : outputWriter, receiveMode != ReceiveMode.ON_START);
|
||||
});
|
||||
|
||||
try (final var daemonHandler = new MultiAccountDaemonHandler(c, receiveMode)) {
|
||||
setup(ns, daemonHandler);
|
||||
|
||||
synchronized (this) {
|
||||
try {
|
||||
wait();
|
||||
} catch (InterruptedException ignored) {
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private static void setup(final Namespace ns, final DaemonHandler daemonHandler) throws CommandException {
|
||||
final Channel inheritedChannel;
|
||||
try {
|
||||
inheritedChannel = System.inheritedChannel();
|
||||
if (inheritedChannel instanceof ServerSocketChannel serverChannel) {
|
||||
logger.info("Using inherited socket: " + serverChannel.getLocalAddress());
|
||||
runSocketMultiAccount(c, serverChannel, receiveMode == ReceiveMode.MANUAL);
|
||||
daemonHandler.runSocket(serverChannel);
|
||||
}
|
||||
} catch (IOException e) {
|
||||
throw new IOErrorException("Failed to use inherited socket", e);
|
||||
}
|
||||
|
||||
final var socketFile = ns.<File>get("socket");
|
||||
if (socketFile != null) {
|
||||
final var address = UnixDomainSocketAddress.of(socketFile.toPath());
|
||||
final var serverChannel = IOUtils.bindSocket(address);
|
||||
runSocketMultiAccount(c, serverChannel, receiveMode == ReceiveMode.MANUAL);
|
||||
daemonHandler.runSocket(serverChannel);
|
||||
}
|
||||
|
||||
final var tcpAddress = ns.getString("tcp");
|
||||
if (tcpAddress != null) {
|
||||
final var address = IOUtils.parseInetSocketAddress(tcpAddress);
|
||||
final var serverChannel = IOUtils.bindSocket(address);
|
||||
runSocketMultiAccount(c, serverChannel, receiveMode == ReceiveMode.MANUAL);
|
||||
daemonHandler.runSocket(serverChannel);
|
||||
}
|
||||
|
||||
final var httpAddress = ns.getString("http");
|
||||
if (httpAddress != null) {
|
||||
final var address = IOUtils.parseInetSocketAddress(httpAddress);
|
||||
final var handler = new HttpServerHandler(address, c);
|
||||
try {
|
||||
handler.init();
|
||||
} catch (IOException ex) {
|
||||
throw new IOErrorException("Failed to initialize HTTP Server", ex);
|
||||
}
|
||||
daemonHandler.runHttp(address);
|
||||
}
|
||||
|
||||
final var isDbusSystem = Boolean.TRUE.equals(ns.getBoolean("dbus-system"));
|
||||
if (isDbusSystem) {
|
||||
runDbusMultiAccount(c, receiveMode != ReceiveMode.ON_START, true);
|
||||
daemonHandler.runDbus(true);
|
||||
}
|
||||
|
||||
final var isDbusSession = Boolean.TRUE.equals(ns.getBoolean("dbus"));
|
||||
if (isDbusSession || (
|
||||
!isDbusSystem
|
||||
|
|
@ -238,14 +204,7 @@ public class DaemonCommand implements MultiLocalCommand, LocalCommand {
|
|||
&& httpAddress == null
|
||||
&& !(inheritedChannel instanceof ServerSocketChannel)
|
||||
)) {
|
||||
runDbusMultiAccount(c, receiveMode != ReceiveMode.ON_START, false);
|
||||
}
|
||||
|
||||
synchronized (this) {
|
||||
try {
|
||||
wait();
|
||||
} catch (InterruptedException ignored) {
|
||||
}
|
||||
daemonHandler.runDbus(false);
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -258,28 +217,23 @@ public class DaemonCommand implements MultiLocalCommand, LocalCommand {
|
|||
m.addReceiveHandler(handler, isWeakListener);
|
||||
}
|
||||
|
||||
private Thread runSocketSingleAccount(
|
||||
final Manager m, final ServerSocketChannel serverChannel, final boolean noReceiveOnStart
|
||||
) {
|
||||
return runSocket(serverChannel, channel -> {
|
||||
final var handler = getSignalJsonRpcDispatcherHandler(channel, noReceiveOnStart);
|
||||
handler.handleConnection(m);
|
||||
});
|
||||
}
|
||||
|
||||
private Thread runSocketMultiAccount(
|
||||
final MultiAccountManager c, final ServerSocketChannel serverChannel, final boolean noReceiveOnStart
|
||||
) {
|
||||
return runSocket(serverChannel, channel -> {
|
||||
final var handler = getSignalJsonRpcDispatcherHandler(channel, noReceiveOnStart);
|
||||
handler.handleConnection(c);
|
||||
});
|
||||
}
|
||||
private static abstract class DaemonHandler implements AutoCloseable {
|
||||
|
||||
protected final ReceiveMode receiveMode;
|
||||
private static final AtomicInteger threadNumber = new AtomicInteger(0);
|
||||
|
||||
private Thread runSocket(final ServerSocketChannel serverChannel, Consumer<SocketChannel> socketHandler) {
|
||||
return Thread.ofPlatform().name("daemon-listener").start(() -> {
|
||||
public DaemonHandler(final ReceiveMode receiveMode) {
|
||||
this.receiveMode = receiveMode;
|
||||
}
|
||||
|
||||
public abstract void runSocket(ServerSocketChannel serverChannel) throws CommandException;
|
||||
|
||||
public abstract void runDbus(boolean isDbusSystem) throws CommandException;
|
||||
|
||||
public abstract void runHttp(InetSocketAddress address) throws CommandException;
|
||||
|
||||
protected void runSocket(final ServerSocketChannel serverChannel, Consumer<SocketChannel> socketHandler) {
|
||||
Thread.ofPlatform().name("daemon-listener").start(() -> {
|
||||
try (final var executor = Executors.newCachedThreadPool()) {
|
||||
while (true) {
|
||||
final var connectionId = threadNumber.getAndIncrement();
|
||||
|
|
@ -305,73 +259,26 @@ public class DaemonCommand implements MultiLocalCommand, LocalCommand {
|
|||
});
|
||||
}
|
||||
}
|
||||
synchronized (this) {
|
||||
notifyAll();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
private SignalJsonRpcDispatcherHandler getSignalJsonRpcDispatcherHandler(
|
||||
final SocketChannel c, final boolean noReceiveOnStart
|
||||
) {
|
||||
protected SignalJsonRpcDispatcherHandler getSignalJsonRpcDispatcherHandler(final SocketChannel c) {
|
||||
final var lineSupplier = IOUtils.getLineSupplier(Channels.newReader(c, StandardCharsets.UTF_8));
|
||||
final var jsonOutputWriter = new JsonWriterImpl(Channels.newWriter(c, StandardCharsets.UTF_8));
|
||||
|
||||
return new SignalJsonRpcDispatcherHandler(jsonOutputWriter, lineSupplier, noReceiveOnStart);
|
||||
return new SignalJsonRpcDispatcherHandler(jsonOutputWriter,
|
||||
lineSupplier,
|
||||
receiveMode == ReceiveMode.MANUAL);
|
||||
}
|
||||
|
||||
private void runDbusSingleAccount(
|
||||
final Manager m, final boolean isDbusSystem, final boolean noReceiveOnStart
|
||||
) throws CommandException {
|
||||
runDbus(isDbusSystem, (conn, objectPath) -> {
|
||||
try {
|
||||
exportDbusObject(conn, objectPath, m, noReceiveOnStart).join();
|
||||
} catch (InterruptedException ignored) {
|
||||
}
|
||||
});
|
||||
protected Thread exportDbusObject(final DBusConnection conn, final String objectPath, final Manager m) {
|
||||
final var signal = new DbusSignalImpl(m, conn, objectPath, receiveMode != ReceiveMode.ON_START);
|
||||
|
||||
return Thread.ofPlatform().name("dbus-init-" + m.getSelfNumber()).start(signal::initObjects);
|
||||
}
|
||||
|
||||
private void runDbusMultiAccount(
|
||||
final MultiAccountManager c, final boolean noReceiveOnStart, final boolean isDbusSystem
|
||||
) throws CommandException {
|
||||
runDbus(isDbusSystem, (connection, objectPath) -> {
|
||||
final var signalControl = new DbusSignalControlImpl(c, objectPath);
|
||||
connection.exportObject(signalControl);
|
||||
|
||||
c.addOnManagerAddedHandler(m -> {
|
||||
final var thread = exportMultiAccountManager(connection, m, noReceiveOnStart);
|
||||
try {
|
||||
thread.join();
|
||||
} catch (InterruptedException ignored) {
|
||||
}
|
||||
});
|
||||
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.getManagers()
|
||||
.stream()
|
||||
.map(m -> exportMultiAccountManager(connection, m, noReceiveOnStart))
|
||||
.toList();
|
||||
|
||||
for (var t : initThreads) {
|
||||
try {
|
||||
t.join();
|
||||
} catch (InterruptedException ignored) {
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
private void runDbus(
|
||||
final boolean isDbusSystem, DbusRunner dbusRunner
|
||||
protected void runDbus(
|
||||
final boolean isDbusSystem, MultiAccountDaemonHandler.DbusRunner dbusRunner
|
||||
) throws CommandException {
|
||||
DBusConnection.DBusBusType busType;
|
||||
if (isDbusSystem) {
|
||||
|
|
@ -392,30 +299,131 @@ public class DaemonCommand implements MultiLocalCommand, LocalCommand {
|
|||
try {
|
||||
conn.requestBusName(DbusConfig.getBusname());
|
||||
} catch (DBusException e) {
|
||||
throw new UnexpectedErrorException("Dbus command failed, maybe signal-cli dbus daemon is already running: "
|
||||
+ e.getMessage(), e);
|
||||
throw new UnexpectedErrorException(
|
||||
"Dbus command failed, maybe signal-cli dbus daemon is already running: " + e.getMessage(),
|
||||
e);
|
||||
}
|
||||
|
||||
logger.info("DBus daemon running on {} bus: {}", busType, DbusConfig.getBusname());
|
||||
}
|
||||
|
||||
private Thread exportMultiAccountManager(
|
||||
final DBusConnection conn, final Manager m, final boolean noReceiveOnStart
|
||||
) {
|
||||
final var objectPath = DbusConfig.getObjectPath(m.getSelfNumber());
|
||||
return exportDbusObject(conn, objectPath, m, noReceiveOnStart);
|
||||
@Override
|
||||
public void close() {
|
||||
// TODO
|
||||
}
|
||||
}
|
||||
|
||||
private Thread exportDbusObject(
|
||||
final DBusConnection conn, final String objectPath, final Manager m, final boolean noReceiveOnStart
|
||||
) {
|
||||
final var signal = new DbusSignalImpl(m, conn, objectPath, noReceiveOnStart);
|
||||
private static final class SingleAccountDaemonHandler extends DaemonHandler {
|
||||
|
||||
return Thread.ofPlatform().name("dbus-init-" + m.getSelfNumber()).start(signal::initObjects);
|
||||
private final Manager m;
|
||||
|
||||
private SingleAccountDaemonHandler(final Manager m, final ReceiveMode receiveMode) {
|
||||
super(receiveMode);
|
||||
this.m = m;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void runSocket(final ServerSocketChannel serverChannel) {
|
||||
runSocket(serverChannel, channel -> {
|
||||
final var handler = getSignalJsonRpcDispatcherHandler(channel);
|
||||
handler.handleConnection(m);
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public void runDbus(final boolean isDbusSystem) throws CommandException {
|
||||
runDbus(isDbusSystem, (conn, objectPath) -> {
|
||||
try {
|
||||
exportDbusObject(conn, objectPath, m).join();
|
||||
} catch (InterruptedException ignored) {
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public void runHttp(InetSocketAddress address) throws CommandException {
|
||||
final var handler = new HttpServerHandler(address, m);
|
||||
try {
|
||||
handler.init();
|
||||
} catch (IOException ex) {
|
||||
throw new IOErrorException("Failed to initialize HTTP Server", ex);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private static final class MultiAccountDaemonHandler extends DaemonHandler {
|
||||
|
||||
private final MultiAccountManager c;
|
||||
|
||||
private MultiAccountDaemonHandler(final MultiAccountManager c, final ReceiveMode receiveMode) {
|
||||
super(receiveMode);
|
||||
this.c = c;
|
||||
}
|
||||
|
||||
public void runSocket(final ServerSocketChannel serverChannel) {
|
||||
runSocket(serverChannel, channel -> {
|
||||
final var handler = getSignalJsonRpcDispatcherHandler(channel);
|
||||
handler.handleConnection(c);
|
||||
});
|
||||
}
|
||||
|
||||
public void runDbus(final boolean isDbusSystem) throws CommandException {
|
||||
runDbus(isDbusSystem, (connection, objectPath) -> {
|
||||
final var signalControl = new DbusSignalControlImpl(c, objectPath);
|
||||
connection.exportObject(signalControl);
|
||||
|
||||
c.addOnManagerAddedHandler(m -> {
|
||||
final var thread = exportMultiAccountManager(connection, m);
|
||||
try {
|
||||
thread.join();
|
||||
} catch (InterruptedException ignored) {
|
||||
}
|
||||
});
|
||||
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.getManagers()
|
||||
.stream()
|
||||
.map(m -> exportMultiAccountManager(connection, m))
|
||||
.toList();
|
||||
|
||||
for (var t : initThreads) {
|
||||
try {
|
||||
t.join();
|
||||
} catch (InterruptedException ignored) {
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public void runHttp(final InetSocketAddress address) throws CommandException {
|
||||
final var handler = new HttpServerHandler(address, c);
|
||||
try {
|
||||
handler.init();
|
||||
} catch (IOException ex) {
|
||||
throw new IOErrorException("Failed to initialize HTTP Server", ex);
|
||||
}
|
||||
}
|
||||
|
||||
private Thread exportMultiAccountManager(
|
||||
final DBusConnection conn, final Manager m
|
||||
) {
|
||||
final var objectPath = DbusConfig.getObjectPath(m.getSelfNumber());
|
||||
return exportDbusObject(conn, objectPath, m);
|
||||
}
|
||||
|
||||
interface DbusRunner {
|
||||
|
||||
void run(DBusConnection connection, String objectPath) throws DBusException;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue