add python class RpcUnixSocket (does not completely work yet)

deltachat-rpc-client/build/lib/deltachat_rpc_client/rpc.py

@@ -0,0 +1,258 @@
+"""JSON-RPC client module."""
+
+from __future__ import annotations
+
+import itertools
+import json
+import logging
+import os
+import subprocess
+import sys
+import socket
+from queue import Empty, Queue
+from threading import Thread
+from typing import TYPE_CHECKING, Any, Iterator, Optional
+
+if TYPE_CHECKING:
+    import io
+
+
+class JsonRpcError(Exception):
+    """JSON-RPC error."""
+
+
+class RpcShutdownError(JsonRpcError):
+    """Raised in RPC methods if the connection to server is closing."""
+
+
+class RpcMethod:
+    """RPC method."""
+
+    def __init__(self, rpc: "BaseRpc", name: str):
+        self.rpc = rpc
+        self.name = name
+
+    def __call__(self, *args) -> Any:
+        """Call JSON-RPC method synchronously."""
+        future = self.future(*args)
+        return future()
+
+    def future(self, *args) -> Any:
+        """Call JSON-RPC method asynchronously."""
+        request_id = next(self.rpc.id_iterator)
+        request = {
+            "jsonrpc": "2.0",
+            "method": self.name,
+            "params": args,
+            "id": request_id,
+        }
+        self.rpc.request_results[request_id] = queue = Queue()
+        self.rpc.request_queue.put(request)
+
+        def rpc_future():
+            """Wait for the request to receive a result."""
+            response = queue.get()
+            if response is None:
+                raise RpcShutdownError(f"no response for {request_id}/{self.name} while rpc is shutting down")
+            if "error" in response:
+                raise JsonRpcError(response["error"])
+            return response.get("result", None)
+
+        return rpc_future
+
+
+class BaseRpc:
+    """Base Rpc class which requires 'connect_to_server' and 'disconnect_from_server' methods
+    from subclasses to work concretely."""
+
+    def __init__(self):
+        self.id_iterator: Iterator[int]
+        self.event_queues: dict[int, Queue]
+        # Map from request ID to a Queue which provides a single result
+        self.request_results: dict[int, Queue]
+        self.request_queue: Queue[Any]
+        self.server_stdin: io.Writer[bytes]
+        self.server_stdout: io.Reader[bytes]
+        self.closing: bool
+        self.reader_thread: Thread
+        self.writer_thread: Thread
+        self.events_thread: Thread
+
+    def start(self) -> None:
+        """Start RPC server subprocess."""
+        self.server_stdout, self.server_stdin = self.connect_to_server()
+        self.id_iterator = itertools.count(start=1)
+        self.event_queues = {}
+        self.request_results = {}
+        self.request_queue = Queue()
+        self.closing = False
+        self.reader_thread = Thread(target=self.reader_loop)
+        self.reader_thread.start()
+        self.writer_thread = Thread(target=self.writer_loop)
+        self.writer_thread.start()
+        self.events_thread = Thread(target=self.events_loop)
+        self.events_thread.start()
+
+    def close(self) -> None:
+        """Terminate RPC server process and wait until the reader loop finishes."""
+        self.closing = True
+        self.disconnect_from_server()
+        self.reader_thread.join()
+        self.events_thread.join()
+        self.request_queue.put(None)
+        self.writer_thread.join()
+
+    def __enter__(self):
+        self.start()
+        return self
+
+    def __exit__(self, _exc_type, _exc, _tb):
+        self.close()
+
+    def reader_loop(self) -> None:
+        """Process JSON-RPC responses from the RPC server process output."""
+        try:
+            while line := self.server_stdout.readline():
+                response = json.loads(line)
+                if "id" in response:
+                    response_id = response["id"]
+                    self.request_results.pop(response_id).put(response)
+                else:
+                    logging.warning("Got a response without ID: %s", response)
+        except Exception:
+            # Log an exception if the reader loop dies.
+            logging.exception("Exception in the reader loop")
+
+        # terminate pending rpc requests because no responses can arrive anymore
+        for queue in self.request_results.values():
+            queue.put(None)
+
+    def writer_loop(self) -> None:
+        """Writer loop ensuring only a single thread writes requests."""
+        try:
+            while request := self.request_queue.get():
+                data = (json.dumps(request) + "\n").encode()
+                self.server_stdin.write(data)
+                self.server_stdin.flush()
+
+        except Exception:
+            # Log an exception if the writer loop dies.
+            logging.exception("Exception in the writer loop")
+
+    def get_queue(self, account_id: int) -> Queue:
+        """Get event queue corresponding to the given account ID."""
+        if account_id not in self.event_queues:
+            self.event_queues[account_id] = Queue()
+        return self.event_queues[account_id]
+
+    def events_loop(self) -> None:
+        """Request new events and distributes them between queues."""
+        try:
+            while True:
+                if self.closing:
+                    return
+                try:
+                    event = self.get_next_event()
+                except RpcShutdownError:
+                    return
+                account_id = event["contextId"]
+                queue = self.get_queue(account_id)
+                event = event["event"]
+                logging.debug("account_id=%d got an event %s", account_id, event)
+                queue.put(event)
+        except Exception:
+            # Log an exception if the event loop dies.
+            logging.exception("Exception in the event loop")
+
+    def wait_for_event(self, account_id: int) -> Optional[dict]:
+        """Wait for the next event from the given account and returns it."""
+        queue = self.get_queue(account_id)
+        return queue.get()
+
+    def clear_all_events(self, account_id: int):
+        """Remove all queued-up events for a given account. Useful for tests."""
+        queue = self.get_queue(account_id)
+        try:
+            while True:
+                queue.get_nowait()
+        except Empty:
+            pass
+
+    def __getattr__(self, attr: str):
+        return RpcMethod(self, attr)
+
+
+class RpcSubprocess(BaseRpc):
+    """RPC client that runs and connects to a deltachat-rpc-server in a subprocess."""
+
+    def __init__(self, accounts_dir: Optional[str] = None, rpc_server_path: Optional[str] = "deltachat-rpc-server"):
+        """Initialize RPC client.
+
+        The given arguments will be passed to subprocess.Popen().
+        """
+        super(RpcSubprocess, self).__init__()
+        self._accounts_dir = accounts_dir
+        self.rpc_server_path: str = rpc_server_path
+
+    def connect_to_server(self):
+        popen_kwargs = {"stdin": subprocess.PIPE, "stdout": subprocess.PIPE}
+        if sys.version_info >= (3, 11):
+            # Prevent subprocess from capturing SIGINT.
+            popen_kwargs["process_group"] = 0
+        else:
+            # `process_group` is not supported before Python 3.11.
+            popen_kwargs["preexec_fn"] = os.setpgrp  # noqa: PLW1509
+
+        if self._accounts_dir:
+            popen_kwargs["env"] = os.environ.copy()
+            popen_kwargs["env"]["DC_ACCOUNTS_PATH"] = str(self._accounts_dir)
+
+        process = subprocess.Popen(self.rpc_server_path, **popen_kwargs)
+        return process.stdout, process.stdin
+
+    def disconnect_from_server(self):
+        self.stop_io_for_all_accounts()
+        self.server_stdin.close()
+
+
+# backward compatibility
+Rpc = RpcSubprocess
+
+
+class RpcFIFO(BaseRpc):
+    """RPC client that runs and connects to a deltachat-rpc-server through FIFO files."""
+
+    def __init__(self, fn_request_fifo: str, fn_response_fifo: str):
+        super(RpcFIFO, self).__init__()
+        self.fn_request_fifo = fn_request_fifo
+        self.fn_response_fifo = fn_response_fifo
+
+    def connect_to_server(self):
+        server_stdin = open(self.fn_request_fifo, "wb")  # noqa
+        server_stdout = open(self.fn_response_fifo, "rb")  # noqa
+        return server_stdout, server_stdin
+
+    def disconnect_from_server(self):
+        self.server_stdin.close()
+        self.server_stdout.close()
+
+class RpcUnixSocket(BaseRpc):
+    """RPC client that connects to a deltachat-rpc-server through FIFO files."""
+
+    def __init__(self, socket_path: str):
+        super(RpcUnixSocket, self).__init__()
+        self.socket_path = socket_path
+        self.client = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+
+    def connect_to_server(self):
+        print(str(self.socket_path))
+        self.client.connect(self.socket_path)
+        print("c1")
+        writer = self.client.makefile("wb")
+        reader = self.client.makefile("rb")
+        print("c2")
+        assert False
+        return reader, writer
+
+    def disconnect_from_server(self):
+        self.client.close()

deltachat-rpc-client/src/deltachat_rpc_client/__init__.py

@@ -8,7 +8,7 @@ from .const import EventType, SpecialContactId
 from .contact import Contact
 from .deltachat import DeltaChat
 from .message import Message
-from .rpc import Rpc
+from .rpc import Rpc, RpcFIFO, RpcUnixSocket
 
 __all__ = [
     "Account",
@@ -22,6 +22,8 @@ __all__ = [
     "Message",
     "SpecialContactId",
     "Rpc",
+    "RpcFIFO",
+    "RpcUnixSocket",
     "run_bot_cli",
     "run_client_cli",
 ]

deltachat-rpc-client/src/deltachat_rpc_client/_utils.py

@@ -47,7 +47,6 @@ class AttrDict(dict):
 def run_client_cli(
     hooks: Optional[Iterable[Tuple[Callable, Union[type, "EventFilter"]]]] = None,
     argv: Optional[list] = None,
-    **kwargs,
 ) -> None:
     """Run a simple command line app, using the given hooks.
 
@@ -55,13 +54,12 @@ def run_client_cli(
     """
     from .client import Client
 
-    _run_cli(Client, hooks, argv, **kwargs)
+    _run_cli(Client, hooks, argv)
 
 
 def run_bot_cli(
     hooks: Optional[Iterable[Tuple[Callable, Union[type, "EventFilter"]]]] = None,
     argv: Optional[list] = None,
-    **kwargs,
 ) -> None:
     """Run a simple bot command line using the given hooks.
 
@@ -69,14 +67,13 @@ def run_bot_cli(
     """
     from .client import Bot
 
-    _run_cli(Bot, hooks, argv, **kwargs)
+    _run_cli(Bot, hooks, argv)
 
 
 def _run_cli(
     client_type: Type["Client"],
     hooks: Optional[Iterable[Tuple[Callable, Union[type, "EventFilter"]]]] = None,
     argv: Optional[list] = None,
-    **kwargs,
 ) -> None:
     from .deltachat import DeltaChat
     from .rpc import Rpc
@@ -94,7 +91,7 @@ def _run_cli(
     parser.add_argument("--password", action="store", help="password", default=os.getenv("DELTACHAT_PASSWORD"))
     args = parser.parse_args(argv[1:])
 
-    with Rpc(accounts_dir=args.accounts_dir, **kwargs) as rpc:
+    with Rpc(accounts_dir=args.accounts_dir) as rpc:
         deltachat = DeltaChat(rpc)
         core_version = (deltachat.get_system_info()).deltachat_core_version
         accounts = deltachat.get_all_accounts()

deltachat-rpc-client/src/deltachat_rpc_client/rpc.py

@@ -8,19 +8,27 @@ import logging
 import os
 import subprocess
 import sys
+import socket
 from queue import Empty, Queue
 from threading import Thread
-from typing import Any, Iterator, Optional
+from typing import TYPE_CHECKING, Any, Iterator, Optional
+
+if TYPE_CHECKING:
+    import io
 
 
 class JsonRpcError(Exception):
     """JSON-RPC error."""
 
 
+class RpcShutdownError(JsonRpcError):
+    """Raised in RPC methods if the connection to server is closing."""
+
+
 class RpcMethod:
     """RPC method."""
 
-    def __init__(self, rpc: "Rpc", name: str):
+    def __init__(self, rpc: "BaseRpc", name: str):
         self.rpc = rpc
         self.name = name
 
@@ -44,6 +52,8 @@ class RpcMethod:
         def rpc_future():
             """Wait for the request to receive a result."""
             response = queue.get()
+            if response is None:
+                raise RpcShutdownError(f"no response for {request_id}/{self.name} while rpc is shutting down")
             if "error" in response:
                 raise JsonRpcError(response["error"])
             return response.get("result", None)
@@ -51,28 +61,18 @@ class RpcMethod:
         return rpc_future
 
 
-class Rpc:
-    """RPC client."""
+class BaseRpc:
+    """Base Rpc class which requires 'connect_to_server' and 'disconnect_from_server' methods
+    from subclasses to work concretely."""
 
-    def __init__(self, accounts_dir: Optional[str] = None, rpc_server_path="deltachat-rpc-server", **kwargs):
-        """Initialize RPC client.
-
-        The given arguments will be passed to subprocess.Popen().
-        """
-        if accounts_dir:
-            kwargs["env"] = {
-                **kwargs.get("env", os.environ),
-                "DC_ACCOUNTS_PATH": str(accounts_dir),
-            }
-
-        self._kwargs = kwargs
-        self.rpc_server_path = rpc_server_path
-        self.process: subprocess.Popen
+    def __init__(self):
         self.id_iterator: Iterator[int]
         self.event_queues: dict[int, Queue]
         # Map from request ID to a Queue which provides a single result
         self.request_results: dict[int, Queue]
         self.request_queue: Queue[Any]
+        self.server_stdin: io.Writer[bytes]
+        self.server_stdout: io.Reader[bytes]
         self.closing: bool
         self.reader_thread: Thread
         self.writer_thread: Thread
@@ -80,16 +80,7 @@ class Rpc:
 
     def start(self) -> None:
         """Start RPC server subprocess."""
-        popen_kwargs = {"stdin": subprocess.PIPE, "stdout": subprocess.PIPE}
-        if sys.version_info >= (3, 11):
-            # Prevent subprocess from capturing SIGINT.
-            popen_kwargs["process_group"] = 0
-        else:
-            # `process_group` is not supported before Python 3.11.
-            popen_kwargs["preexec_fn"] = os.setpgrp  # noqa: PLW1509
-
-        popen_kwargs.update(self._kwargs)
-        self.process = subprocess.Popen(self.rpc_server_path, **popen_kwargs)
+        self.server_stdout, self.server_stdin = self.connect_to_server()
         self.id_iterator = itertools.count(start=1)
         self.event_queues = {}
         self.request_results = {}
@@ -105,10 +96,9 @@ class Rpc:
     def close(self) -> None:
         """Terminate RPC server process and wait until the reader loop finishes."""
         self.closing = True
-        self.stop_io_for_all_accounts()
-        self.events_thread.join()
-        self.process.stdin.close()
+        self.disconnect_from_server()
         self.reader_thread.join()
+        self.events_thread.join()
         self.request_queue.put(None)
         self.writer_thread.join()
 
@@ -122,7 +112,7 @@ class Rpc:
     def reader_loop(self) -> None:
         """Process JSON-RPC responses from the RPC server process output."""
         try:
-            while line := self.process.stdout.readline():
+            while line := self.server_stdout.readline():
                 response = json.loads(line)
                 if "id" in response:
                     response_id = response["id"]
@@ -133,13 +123,17 @@ class Rpc:
             # Log an exception if the reader loop dies.
             logging.exception("Exception in the reader loop")
 
+        # terminate pending rpc requests because no responses can arrive anymore
+        for queue in self.request_results.values():
+            queue.put(None)
+
     def writer_loop(self) -> None:
         """Writer loop ensuring only a single thread writes requests."""
         try:
             while request := self.request_queue.get():
                 data = (json.dumps(request) + "\n").encode()
-                self.process.stdin.write(data)
-                self.process.stdin.flush()
+                self.server_stdin.write(data)
+                self.server_stdin.flush()
 
         except Exception:
             # Log an exception if the writer loop dies.
@@ -157,7 +151,10 @@ class Rpc:
             while True:
                 if self.closing:
                     return
-                event = self.get_next_event()
+                try:
+                    event = self.get_next_event()
+                except RpcShutdownError:
+                    return
                 account_id = event["contextId"]
                 queue = self.get_queue(account_id)
                 event = event["event"]
@@ -183,3 +180,76 @@ class Rpc:
 
     def __getattr__(self, attr: str):
         return RpcMethod(self, attr)
+
+
+class RpcSubprocess(BaseRpc):
+    """RPC client that runs and connects to a deltachat-rpc-server in a subprocess."""
+
+    def __init__(self, accounts_dir: Optional[str] = None, rpc_server_path: Optional[str] = "deltachat-rpc-server"):
+        """Initialize RPC client.
+
+        The given arguments will be passed to subprocess.Popen().
+        """
+        super(RpcSubprocess, self).__init__()
+        self._accounts_dir = accounts_dir
+        self.rpc_server_path: str = rpc_server_path
+
+    def connect_to_server(self):
+        popen_kwargs = {"stdin": subprocess.PIPE, "stdout": subprocess.PIPE}
+        if sys.version_info >= (3, 11):
+            # Prevent subprocess from capturing SIGINT.
+            popen_kwargs["process_group"] = 0
+        else:
+            # `process_group` is not supported before Python 3.11.
+            popen_kwargs["preexec_fn"] = os.setpgrp  # noqa: PLW1509
+
+        if self._accounts_dir:
+            popen_kwargs["env"] = os.environ.copy()
+            popen_kwargs["env"]["DC_ACCOUNTS_PATH"] = str(self._accounts_dir)
+
+        process = subprocess.Popen(self.rpc_server_path, **popen_kwargs)
+        return process.stdout, process.stdin
+
+    def disconnect_from_server(self):
+        self.stop_io_for_all_accounts()
+        self.server_stdin.close()
+
+
+# backward compatibility
+Rpc = RpcSubprocess
+
+
+class RpcFIFO(BaseRpc):
+    """RPC client that runs and connects to a deltachat-rpc-server through FIFO files."""
+
+    def __init__(self, fn_request_fifo: str, fn_response_fifo: str):
+        super(RpcFIFO, self).__init__()
+        self.fn_request_fifo = fn_request_fifo
+        self.fn_response_fifo = fn_response_fifo
+
+    def connect_to_server(self):
+        server_stdin = open(self.fn_request_fifo, "wb")  # noqa
+        server_stdout = open(self.fn_response_fifo, "rb")  # noqa
+        return server_stdout, server_stdin
+
+    def disconnect_from_server(self):
+        self.server_stdin.close()
+        self.server_stdout.close()
+
+class RpcUnixSocket(BaseRpc):
+    """RPC client that connects to a deltachat-rpc-server through FIFO files."""
+
+    def __init__(self, socket_path: str):
+        super(RpcUnixSocket, self).__init__()
+        self.socket_path = socket_path
+        self.client = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+
+    def connect_to_server(self):
+        print(str(self.socket_path))
+        self.client.connect(self.socket_path)
+        writer = self.client.makefile("wb")
+        reader = self.client.makefile("rb")
+        return reader, writer
+
+    def disconnect_from_server(self):
+        self.client.close()

deltachat-rpc-client/tests/test_rpc_fifo.py

@@ -0,0 +1,22 @@
+import os
+import platform  # noqa
+import subprocess
+
+import pytest
+
+from deltachat_rpc_client import DeltaChat, RpcFIFO
+
+
+@pytest.mark.skipif("platform.system() == 'Windows'")
+def test_rpc_fifo(tmp_path):
+    fn_request_fifo = tmp_path.joinpath("request_fifo")
+    fn_response_fifo = tmp_path.joinpath("response_fifo")
+    os.mkfifo(fn_request_fifo)
+    os.mkfifo(fn_response_fifo)
+    popen = subprocess.Popen(f"deltachat-rpc-server <{fn_request_fifo} >{fn_response_fifo}", shell=True)
+
+    rpc = RpcFIFO(fn_response_fifo=fn_response_fifo, fn_request_fifo=fn_request_fifo)
+    with rpc:
+        dc = DeltaChat(rpc)
+        assert dc.rpc.get_system_info()["deltachat_core_version"] is not None
+    popen.wait()

deltachat-rpc-client/tests/test_rpc_unix.py

@@ -0,0 +1,37 @@
+from os import environ
+import platform  # noqa
+import signal
+import subprocess
+from sys import stderr
+from time import sleep
+
+import pytest
+
+from deltachat_rpc_client import DeltaChat, RpcUnixSocket
+
+
+@pytest.mark.skipif("platform.system() == 'Windows'")
+def test_rpc_unix(tmp_path):
+    socket_file = "/tmp/chatmail.sock" # path needs to be relative or short
+
+    path = environ.get("PATH")
+    assert path is not None
+
+    popen = subprocess.Popen(
+        f"deltachat-rpc-server --unix {socket_file}",
+        shell=True,
+        env=dict(
+            DC_ACCOUNTS_PATH=f"{tmp_path}/accounts/test",
+            rust_log="info",
+            PATH=path
+        )
+    )
+
+    sleep(1) # wait until socket exists # TODO this should not be needed
+
+    rpc = RpcUnixSocket(socket_path=socket_file)
+    with rpc:
+        dc = DeltaChat(rpc)
+        assert dc.rpc.get_system_info()["deltachat_core_version"] is not None
+    popen.send_signal(signal.SIGINT)
+    popen.wait()

deltachat-rpc-server/README.md

@@ -1,7 +1,7 @@
 # Delta Chat RPC server
 
 This program provides a [JSON-RPC 2.0](https://www.jsonrpc.org/specification) interface to DeltaChat
-over standard I/O.
+over standard I/O or UNIX sockets.
 
 ## Install
 
@@ -35,3 +35,25 @@ languages other than Rust, for example:
 
 Run `deltachat-rpc-server --version` to check the version of the server.
 Run `deltachat-rpc-server --openrpc` to get [OpenRPC](https://open-rpc.org/) specification of the provided JSON-RPC API.
+
+### Usage over unix sockets
+
+> At this time this does not work on windows because rust does not support unix sockets on windows, yet (<https://github.com/rust-lang/rust/issues/150487>).
+
+Standard I/O is the default option, but you can also tell `deltachat-rpc-server` to use a unix socket instead.
+
+```
+RUST_LOG=info deltachat-rpc-server --unix ./chatmail-core.sock
+```
+
+While this technically allows multiple processes to communicate with the same rpc server instance,
+please note that there is still only event queue, so only one of these processed should read the events at a time.
+
+You can test it with socat:
+```sh
+socat - UNIX-CONNECT:./chatmail-core.sock
+```
+Then paste the following jsonrpc command and press enter:
+```json
+{"method": "get_system_info","id": 1,"params": []}
+```

deltachat-rpc-server/src/main.rs

@@ -14,6 +14,10 @@ use tokio::io::{self, AsyncBufReadExt, BufReader};
 use tracing_subscriber::EnvFilter;
 use yerpc::RpcServer as _;
 
+#[cfg(target_family = "unix")]
+use std::ffi::OsString;
+#[cfg(target_family = "unix")]
+use tokio::net::UnixListener;
 #[cfg(target_family = "unix")]
 use tokio::signal::unix as signal_unix;
 
@@ -24,6 +28,14 @@ use yerpc::{RpcClient, RpcSession};
 
 #[tokio::main(flavor = "multi_thread")]
 async fn main() {
+    // Logs from `log` crate and traces from `tracing` crate
+    // are configurable with `RUST_LOG` environment variable
+    // and go to stderr to avoid interfering with JSON-RPC using stdout.
+    tracing_subscriber::fmt()
+        .with_env_filter(EnvFilter::from_default_env())
+        .with_writer(std::io::stderr)
+        .init();
+
     let r = main_impl().await;
     // From tokio documentation:
     // "For technical reasons, stdin is implemented by using an ordinary blocking read on a separate
@@ -38,6 +50,7 @@ async fn main() {
 async fn main_impl() -> Result<()> {
     let mut args = env::args_os();
     let _program_name = args.next().context("no command line arguments found")?;
+    let mut unix_socket = None;
     if let Some(first_arg) = args.next() {
         if first_arg.to_str() == Some("--version") {
             if let Some(arg) = args.next() {
@@ -51,6 +64,11 @@ async fn main_impl() -> Result<()> {
             }
             println!("{}", CommandApi::openrpc_specification()?);
             return Ok(());
+        } else if first_arg.to_str() == Some("--unix") {
+            let Some(unix_socket_path) = args.next() else {
+                return Err(anyhow!("Unix Socket Path is missing"));
+            };
+            unix_socket = Some(unix_socket_path)
         } else {
             return Err(anyhow!("Unrecognized option {first_arg:?}"));
         }
@@ -64,14 +82,6 @@ async fn main_impl() -> Result<()> {
     #[cfg(target_family = "unix")]
     let mut sigterm = signal_unix::signal(signal_unix::SignalKind::terminate())?;
 
-    // Logs from `log` crate and traces from `tracing` crate
-    // are configurable with `RUST_LOG` environment variable
-    // and go to stderr to avoid interfering with JSON-RPC using stdout.
-    tracing_subscriber::fmt()
-        .with_env_filter(EnvFilter::from_default_env())
-        .with_writer(std::io::stderr)
-        .init();
-
     let path = std::env::var("DC_ACCOUNTS_PATH").unwrap_or_else(|_| "accounts".to_string());
     log::info!("Starting with accounts directory `{path}`.");
     let writable = true;
@@ -81,10 +91,58 @@ async fn main_impl() -> Result<()> {
     let accounts = Arc::new(RwLock::new(accounts));
     let state = CommandApi::from_arc(accounts.clone()).await;
 
-    let (client, mut out_receiver) = RpcClient::new();
-    let session = RpcSession::new(client.clone(), state.clone());
     let main_cancel = CancellationToken::new();
 
+    let cancel = main_cancel.clone();
+    let sigterm_task: JoinHandle<anyhow::Result<()>> = tokio::spawn(async move {
+        #[cfg(target_family = "unix")]
+        {
+            let _cancel_guard = cancel.clone().drop_guard();
+            tokio::select! {
+                _ = cancel.cancelled() => (),
+                _ = sigterm.recv() => {
+                    log::info!("got SIGTERM");
+                }
+            }
+        }
+        let _ = cancel;
+        Ok(())
+    });
+
+    let (send_task, recv_task) = if let Some(unix_socket_path) = unix_socket {
+        #[cfg(not(target_family = "unix"))]
+        {
+            return Err(anyhow!(
+                "unix sockets are only supported on unix based operating systems"
+            ));
+        }
+        #[cfg(target_family = "unix")]
+        unix_socket_impl(unix_socket_path, state, main_cancel.clone()).await?
+    } else {
+        stdio_impl(state, main_cancel.clone()).await?
+    };
+
+    main_cancel.cancelled().await;
+    accounts.read().await.stop_io().await;
+    drop(accounts);
+    send_task.await??;
+    sigterm_task.await??;
+    recv_task.await??;
+
+    Ok(())
+}
+
+async fn stdio_impl(
+    state: CommandApi,
+    main_cancel: CancellationToken,
+) -> Result<(
+    JoinHandle<anyhow::Result<()>>,
+    JoinHandle<anyhow::Result<()>>,
+)> {
+    let (client, mut out_receiver) = RpcClient::new();
+
+    let session = RpcSession::new(client.clone(), state.clone());
+
     // Send task prints JSON responses to stdout.
     let cancel = main_cancel.clone();
     let send_task: JoinHandle<anyhow::Result<()>> = tokio::spawn(async move {
@@ -103,22 +161,6 @@ async fn main_impl() -> Result<()> {
         Ok(())
     });
 
-    let cancel = main_cancel.clone();
-    let sigterm_task: JoinHandle<anyhow::Result<()>> = tokio::spawn(async move {
-        #[cfg(target_family = "unix")]
-        {
-            let _cancel_guard = cancel.clone().drop_guard();
-            tokio::select! {
-                _ = cancel.cancelled() => (),
-                _ = sigterm.recv() => {
-                    log::info!("got SIGTERM");
-                }
-            }
-        }
-        let _ = cancel;
-        Ok(())
-    });
-
     // Receiver task reads JSON requests from stdin.
     let cancel = main_cancel.clone();
     let recv_task: JoinHandle<anyhow::Result<()>> = tokio::spawn(async move {
@@ -150,13 +192,104 @@ async fn main_impl() -> Result<()> {
         Ok(())
     });
 
-    main_cancel.cancelled().await;
-    accounts.read().await.stop_io().await;
-    drop(accounts);
-    drop(state);
-    send_task.await??;
-    sigterm_task.await??;
-    recv_task.await??;
-
-    Ok(())
+    Ok((send_task, recv_task))
+}
+
+#[cfg(target_family = "unix")]
+async fn unix_socket_impl(
+    unix_socket_path: OsString,
+    state: CommandApi,
+    main_cancel: CancellationToken,
+) -> Result<(
+    JoinHandle<anyhow::Result<()>>,
+    JoinHandle<anyhow::Result<()>>,
+)> {
+    let cancel = main_cancel.clone();
+
+    let listener = UnixListener::bind(&unix_socket_path)?;
+
+    let recv_task: JoinHandle<anyhow::Result<()>> = tokio::spawn(async move {
+        let _cancel_guard = cancel.clone().drop_guard();
+        let cancel = main_cancel.clone();
+
+        loop {
+            let connection_result = tokio::select! {
+                _ = cancel.cancelled() => break,
+                _ = tokio::signal::ctrl_c() => {
+                    log::info!("got ctrl-c event");
+                    break;
+                }
+                connection_result = listener.accept() => connection_result
+            };
+            match connection_result {
+                Ok((stream, addr)) => {
+                    log::info!("new client {addr:?}");
+
+                    let (client, mut out_receiver) = RpcClient::new();
+                    let session = RpcSession::new(client.clone(), state.clone());
+
+                    let (read, mut write) = stream.into_split();
+                    let mut read_lines = BufReader::new(read).lines();
+
+                    let cancel = main_cancel.clone();
+                    let _receive_task: JoinHandle<anyhow::Result<()>> = tokio::spawn(async move {
+                        let _cancel_guard = cancel.clone().drop_guard();
+                        loop {
+                            let message = tokio::select! {
+                                _ = cancel.cancelled() => break,
+                                _ = tokio::signal::ctrl_c() => {
+                                    log::info!("got ctrl-c event");
+                                    break;
+                                }
+                                message =
+                                    read_lines.next_line()
+                                     => match message? {
+                                    None => {
+                                        log::info!("unix socket closed {addr:?}");
+                                        break;
+                                    }
+                                    Some(message) => message,
+                                }
+                            };
+                            log::trace!("RPC recv {message}");
+                            let session = session.clone();
+                            tokio::spawn(async move {
+                                session.handle_incoming(&message).await;
+                            });
+                        }
+                        Ok(())
+                    });
+
+                    let cancel = main_cancel.clone();
+                    let _send_task: JoinHandle<anyhow::Result<()>> = tokio::spawn(async move {
+                        let _cancel_guard = cancel.clone().drop_guard();
+                        loop {
+                            use tokio::io::AsyncWriteExt;
+
+                            let message = tokio::select! {
+                                _ = cancel.cancelled() => break,
+                                message = out_receiver.next() => match message {
+                                    None => break,
+                                    Some(message) => serde_json::to_string(&message)?,
+                                }
+                            };
+                            log::trace!("RPC send {message}");
+                            write.write_all(format!("{message}\n").as_bytes()).await?;
+                        }
+                        Ok(())
+                    });
+                }
+                Err(e) => {
+                    log::info!("connection failed {e:#}");
+                }
+            }
+        }
+
+        drop(listener);
+        std::fs::remove_file(unix_socket_path)?;
+
+        Ok(())
+    });
+
+    Ok((tokio::spawn(async move { Ok(()) }), recv_task))
 }