selfprivacy-rest-api/selfprivacy_api/graphql/subscriptions/logs.py

from typing import AsyncGenerator
from systemd import journal
import asyncio

from selfprivacy_api.graphql.queries.logs import LogEntry


async def log_stream() -> AsyncGenerator[LogEntry, None]:
    j = journal.Reader()

    j.seek_tail()
    j.get_previous()

    queue = asyncio.Queue()

    async def callback():
        if j.process() != journal.APPEND:
            return
        for entry in j:
            await queue.put(entry)

    asyncio.get_event_loop().add_reader(j, lambda: asyncio.ensure_future(callback()))

    try:
        while True:
            entry = await queue.get()
            try:
                yield LogEntry(entry)
            except Exception:
                asyncio.get_event_loop().remove_reader(j)
                j.close()
                return
            queue.task_done()
    except asyncio.CancelledError:
        asyncio.get_event_loop().remove_reader(j)
        j.close()
        return