webmaster/signal-cli
1
0
Fork 0
mirror of https://github.com/AsamK/signal-cli synced 2025-08-29 10:30:38 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Use new threads API

Browse source
This commit is contained in:
AsamK 2023-10-17 21:56:10 +02:00
parent 8d55dfb66b
commit ed8ac5b84c
7 changed files with 40 additions and 75 deletions
Download patch file Download diff file Expand all files Collapse all files

70
src/main/java/org/asamk/signal/commands/DaemonCommand.java
View file

@ -39,6 +39,7 @@ import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;
@ -257,19 +258,19 @@ public class DaemonCommand implements MultiLocalCommand, LocalCommand {
m.addReceiveHandler(handler, isWeakListener);
}
private void runSocketSingleAccount(
private Thread runSocketSingleAccount(
final Manager m, final ServerSocketChannel serverChannel, final boolean noReceiveOnStart
) {
runSocket(serverChannel, channel -> {
return runSocket(serverChannel, channel -> {
final var handler = getSignalJsonRpcDispatcherHandler(channel, noReceiveOnStart);
handler.handleConnection(m);
});
}
private void runSocketMultiAccount(
private Thread runSocketMultiAccount(
final MultiAccountManager c, final ServerSocketChannel serverChannel, final boolean noReceiveOnStart
) {
runSocket(serverChannel, channel -> {
return runSocket(serverChannel, channel -> {
final var handler = getSignalJsonRpcDispatcherHandler(channel, noReceiveOnStart);
handler.handleConnection(c);
});
@ -277,39 +278,37 @@ public class DaemonCommand implements MultiLocalCommand, LocalCommand {
private static final AtomicInteger threadNumber = new AtomicInteger(0);
private void runSocket(final ServerSocketChannel serverChannel, Consumer<SocketChannel> socketHandler) {
final var thread = new Thread(() -> {
while (true) {
final var connectionId = threadNumber.getAndIncrement();
final SocketChannel channel;
final String clientString;
try {
channel = serverChannel.accept();
clientString = channel.getRemoteAddress() + " " + IOUtils.getUnixDomainPrincipal(channel);
logger.info("Accepted new client connection {}: {}", connectionId, clientString);
} catch (IOException e) {
logger.error("Failed to accept new socket connection", e);
synchronized (this) {
notifyAll();
}
break;
}
final var connectionThread = new Thread(() -> {
try (final var c = channel) {
socketHandler.accept(c);
private Thread runSocket(final ServerSocketChannel serverChannel, Consumer<SocketChannel> socketHandler) {
return Thread.ofPlatform().name("daemon-listener").start(() -> {
try (final var executor = Executors.newCachedThreadPool()) {
while (true) {
final var connectionId = threadNumber.getAndIncrement();
final SocketChannel channel;
final String clientString;
try {
channel = serverChannel.accept();
clientString = channel.getRemoteAddress() + " " + IOUtils.getUnixDomainPrincipal(channel);
logger.info("Accepted new client connection {}: {}", connectionId, clientString);
} catch (IOException e) {
logger.warn("Failed to close channel", e);
} catch (Throwable e) {
logger.warn("Connection handler failed, closing connection", e);
logger.error("Failed to accept new socket connection", e);
break;
}
logger.info("Connection {} closed: {}", connectionId, clientString);
});
connectionThread.setName("daemon-connection-" + connectionId);
connectionThread.start();
executor.submit(() -> {
try (final var c = channel) {
socketHandler.accept(c);
} catch (IOException e) {
logger.warn("Failed to close channel", e);
} catch (Throwable e) {
logger.warn("Connection handler failed, closing connection", e);
}
logger.info("Connection {} closed: {}", connectionId, clientString);
});
}
}
synchronized (this) {
notifyAll();
}
});
thread.setName("daemon-listener");
thread.start();
}
private SignalJsonRpcDispatcherHandler getSignalJsonRpcDispatcherHandler(
@ -411,11 +410,8 @@ public class DaemonCommand implements MultiLocalCommand, LocalCommand {
final DBusConnection conn, final String objectPath, final Manager m, final boolean noReceiveOnStart
) {
final var signal = new DbusSignalImpl(m, conn, objectPath, noReceiveOnStart);
final var initThread = new Thread(signal::initObjects);
initThread.setName("dbus-init");
initThread.start();
return initThread;
return Thread.ofPlatform().name("dbus-init-" + m.getSelfNumber()).start(signal::initObjects);
}
interface DbusRunner {

4
src/main/java/org/asamk/signal/dbus/DbusSignalControlImpl.java
View file

@ -103,7 +103,7 @@ public class DbusSignalControlImpl implements org.asamk.SignalControl {
public String link(final String newDeviceName) throws Error.Failure {
try {
final URI deviceLinkUri = c.getNewProvisioningDeviceLinkUri();
final var thread = new Thread(() -> {
Thread.ofPlatform().name("dbus-link").start(() -> {
final ProvisioningManager provisioningManager = c.getProvisioningManagerFor(deviceLinkUri);
try {
provisioningManager.finishDeviceLink(newDeviceName);
@ -111,8 +111,6 @@ public class DbusSignalControlImpl implements org.asamk.SignalControl {
e.printStackTrace();
}
});
thread.setName("dbus-link");
thread.start();
return deviceLinkUri.toString();
} catch (TimeoutException | IOException e) {
throw new SignalControl.Error.Failure(e.getClass().getSimpleName() + " " + e.getMessage());

2
src/main/java/org/asamk/signal/http/HttpServerHandler.java
View file

@ -54,7 +54,7 @@ public class HttpServerHandler {
logger.info("Starting server on " + address.toString());
final var server = HttpServer.create(address, 0);
server.setExecutor(Executors.newFixedThreadPool(10));
server.setExecutor(Executors.newCachedThreadPool());
server.createContext("/api/v1/rpc", this::handleRpcEndpoint);
server.createContext("/api/v1/events", this::handleEventsEndpoint);

10
src/main/java/org/asamk/signal/jsonrpc/JsonRpcReader.java
View file

@ -55,8 +55,7 @@ public class JsonRpcReader {
return;
}
final var executor = Executors.newFixedThreadPool(10);
try {
try (final var executor = Executors.newCachedThreadPool()) {
while (!Thread.interrupted()) {
final var input = lineSupplier.get();
if (input == null) {
@ -72,8 +71,6 @@ public class JsonRpcReader {
executor.submit(() -> handleMessage(message, requestHandler, responseHandler));
}
} finally {
Util.closeExecutorService(executor);
}
}
@ -94,8 +91,7 @@ public class JsonRpcReader {
case JsonRpcBatchMessage jsonRpcBatchMessage -> {
final var messages = jsonRpcBatchMessage.getMessages();
final var responseList = new ArrayList<JsonRpcResponse>(messages.size());
final var executor = Executors.newFixedThreadPool(10);
try {
try (final var executor = Executors.newCachedThreadPool()) {
final var lock = new ReentrantLock();
messages.forEach(jsonNode -> {
final JsonRpcRequest request;
@ -124,8 +120,6 @@ public class JsonRpcReader {
}
});
});
} finally {
Util.closeExecutorService(executor);
}
if (!responseList.isEmpty()) {

17
src/main/java/org/asamk/signal/util/Util.java
View file

@ -16,8 +16,6 @@ import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
public class Util {
@ -86,19 +84,4 @@ public class Util {
}
return map;
}
public static void closeExecutorService(ExecutorService executor) {
executor.shutdown();
try {
if (!executor.awaitTermination(5, TimeUnit.MINUTES)) {
executor.shutdownNow();
if (!executor.awaitTermination(1, TimeUnit.MINUTES)) {
logger.warn("Failed to shutdown executor service");
}
}
} catch (InterruptedException e) {
executor.shutdownNow();
Thread.currentThread().interrupt();
}
}
}