Add listen_fds support, for socket activation

src/trick17/__about__.py

@@ -1,4 +1,4 @@
 # SPDX-FileCopyrightText: 2023-present Stefano Miccoli <[email protected]>
 #
 # SPDX-License-Identifier: MIT
-__version__ = "0.0.3.dev0"
+__version__ = "0.0.3.dev1"

src/trick17/__init__.py

@@ -5,7 +5,11 @@
 # systemd notable paths
 SD_BOOTED_PATH = "/run/systemd/system"
 SD_JOURNAL_SOCKET_PATH = "/run/systemd/journal/socket"
+SD_LISTEN_FDS_START = 3
 
 # environmet variables possibly set by systemd
 SD_JOURNAL_STREAM_ENV = "JOURNAL_STREAM"
 SD_NOTIFY_SOCKET_ENV = "NOTIFY_SOCKET"
+SD_LISTEN_FDS_PID_ENV = "LISTEN_PID"
+SD_LISTEN_FDS_ENV = "LISTEN_FDS"
+SD_LISTEN_FDS_NAMES_ENV = "LISTEN_FDNAMES"

src/trick17/daemon.py

@@ -4,6 +4,7 @@
 
 import os
 from pathlib import Path
+from typing import Iterator, List, Tuple
 
 import trick17
 from trick17 import util
@@ -16,6 +17,52 @@ def booted() -> bool:
  return Path(trick17.SD_BOOTED_PATH).is_dir()
 
 
+def listen_fds() -> Iterator[Tuple[int, str]]:
+ """listen_fds() returns an iterator over (fd, name) tuples, where
+ - fd is an open file descriptor intialized by systemd socket-activation
+ - name is an optional name, '' if undefined.
+ """
+
+ # check pid
+ if trick17.SD_LISTEN_FDS_PID_ENV not in os.environ:
+ return iter(())
+ try:
+ pid = int(os.environ[trick17.SD_LISTEN_FDS_PID_ENV])
+ except ValueError as err:
+ msg = f"Unable to get pid from environment: {err}"
+ raise RuntimeError(msg) from err
+ if os.getpid() != pid:
+ return iter(())
+
+ # check FDS
+ nfds: int
+ try:
+ nfds = int(os.environ[trick17.SD_LISTEN_FDS_ENV])
+ except KeyError as err:
+ msg = f"Unable to get number of fd's from environment: {err}"
+ raise RuntimeError(msg) from err
+ except ValueError as err:
+ msg = f"Nvalid number of fd's in environment: {err}"
+ raise RuntimeError(msg) from err
+ fds = range(trick17.SD_LISTEN_FDS_START, trick17.SD_LISTEN_FDS_START + nfds)
+ assert len(fds) == nfds
+
+ # check names
+ names: List[str]
+ if trick17.SD_LISTEN_FDS_NAMES_ENV not in os.environ:
+ names = [
+ "",
+ ] * nfds
+ else:
+ names = os.environ[trick17.SD_LISTEN_FDS_NAMES_ENV].split(os.pathsep)
+ if len(names) > nfds:
+ names = names[:nfds]
+ elif len(names) < nfds:
+ names.extend("" for _ in range(nfds - len(names)))
+
+ return zip(fds, names)
+
+
 def notify(state: str) -> bool:
  """notify 'state' to systemd; returns
  - True if notification sent to socket,

tests/test_daemon.py

@@ -1,3 +1,5 @@
+import pytest
+
 from trick17 import daemon
 
 
@@ -8,3 +10,8 @@ def test_booted():
 def test_notify():
  ret = daemon.notify("READY=1")
  assert ret is False
+
+
+def test_listen_fds():
+ with pytest.raises(StopIteration):
+ next(daemon.listen_fds())