- protected void runSocket(final ServerSocketChannel serverChannel, Consumer<SocketChannel> socketHandler) {
- final List<AutoCloseable> channels = new ArrayList<>();
- final var thread = 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 (ClosedChannelException ignored) {
- logger.trace("Listening socket has been closed");
- break;
- } catch (IOException e) {
- logger.error("Failed to accept new socket connection", e);
- break;
- }
- channels.add(channel);
- 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);
- channels.remove(channel);
- });
- }
- }
- });
- closeables.add(() -> {
- serverChannel.close();
- for (final var c : new ArrayList<>(channels)) {
- c.close();
- }
- thread.join();
- });
- }
-
- 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,
- receiveMode == ReceiveMode.MANUAL);
- }
-
- protected Thread exportDbusObject(final DBusConnection conn, final String objectPath, final Manager m) {
- final var signal = new DbusSignalImpl(m, conn, objectPath, receiveMode != ReceiveMode.ON_START);
- closeables.add(signal);
-
- return Thread.ofPlatform().name("dbus-init-" + m.getSelfNumber()).start(signal::initObjects);