# Copyright (c) 2023 Nordic Semiconductor ASA
#
# SPDX-License-Identifier: Apache-2.0

from __future__ import annotations

import io
import logging
import os
import threading
import time
from pathlib import Path

logger = logging.getLogger(__name__)


class FifoHandler:
    """
    Class dedicated for handling communication over POSIX system FIFO (named
    pipes).
    """

    def __init__(self, fifo_path: str | Path, timeout: float):
        """
        :param fifo_path: path to basic fifo name
        :param timeout: timeout for establishing connection over FIFO
        """
        self._fifo_out_path = str(fifo_path) + '.out'
        self._fifo_in_path = str(fifo_path) + '.in'
        self._fifo_out_file: io.FileIO | None = None
        self._fifo_in_file: io.FileIO | None = None
        self._open_fifo_thread: threading.Thread | None = None
        self._opening_monitor_thread: threading.Thread | None = None
        self._fifo_opened: threading.Event = threading.Event()
        self._stop_waiting_for_opening: threading.Event = threading.Event()
        self._timeout = timeout

    def initiate_connection(self) -> None:
        """
        Opening FIFO could be a blocking operation (it requires also opening
        FIFO on the other side - by separate program/process). So, to avoid
        blockage, execute opening FIFO in separate thread and additionally run
        in second thread opening time monitor to alternatively unblock first
        thread when timeout will expire.
        """
        self._stop_waiting_for_opening.clear()
        self._make_fifo_file(self._fifo_out_path)
        self._make_fifo_file(self._fifo_in_path)
        if self._open_fifo_thread is None:
            self._open_fifo_thread = threading.Thread(target=self._open_fifo, daemon=True)
            self._open_fifo_thread.start()
        if self._opening_monitor_thread is None:
            self._opening_monitor_thread = threading.Thread(target=self._opening_monitor, daemon=True)
            self._opening_monitor_thread.start()

    @staticmethod
    def _make_fifo_file(filename: str) -> None:
        if not os.path.exists(filename):
            os.mkfifo(filename)

    def _open_fifo(self) -> None:
        self._fifo_out_file = open(self._fifo_out_path, 'rb', buffering=0)
        self._fifo_in_file = open(self._fifo_in_path, 'wb', buffering=0)
        if not self._stop_waiting_for_opening.is_set():
            self._fifo_opened.set()

    def _opening_monitor(self) -> None:
        """
        Monitor opening FIFO operation - if timeout was expired (or disconnect
        was called in the meantime), then interrupt opening FIFO in other
        thread.
        """
        timeout_time: float = time.time() + self._timeout
        while time.time() < timeout_time and not self._stop_waiting_for_opening.is_set():
            if self._fifo_opened.is_set():
                return
            time.sleep(0.1)
        self._stop_waiting_for_opening.set()
        self._unblock_open_fifo_operation()

    def _unblock_open_fifo_operation(self) -> None:
        """
        This is workaround for unblocking opening FIFO operation - imitate
        opening FIFO "on the other side".
        """
        if os.path.exists(self._fifo_out_path):
            open(self._fifo_out_path, 'wb', buffering=0)
        if os.path.exists(self._fifo_in_path):
            open(self._fifo_in_path, 'rb', buffering=0)

    def disconnect(self) -> None:
        self._stop_waiting_for_opening.set()
        if self._open_fifo_thread and self._open_fifo_thread.is_alive():
            self._open_fifo_thread.join(timeout=1)
        self._open_fifo_thread = None
        if self._opening_monitor_thread and self._opening_monitor_thread.is_alive():
            self._opening_monitor_thread.join(timeout=1)
        self._opening_monitor_thread = None
        self._fifo_opened.clear()

        if self._fifo_out_file:
            self._fifo_out_file.close()
        if self._fifo_in_file:
            self._fifo_in_file.close()

        if os.path.exists(self._fifo_out_path):
            os.unlink(self._fifo_out_path)
        if os.path.exists(self._fifo_in_path):
            os.unlink(self._fifo_in_path)

    @property
    def is_open(self) -> bool:
        try:
            return bool(
                self._fifo_opened.is_set()
                and self._fifo_in_file is not None and self._fifo_out_file is not None
                and self._fifo_in_file.fileno() and self._fifo_out_file.fileno()
            )
        except ValueError:
            return False

    def read(self, __size: int = -1) -> bytes:
        return self._fifo_out_file.read(__size)  # type: ignore[union-attr]

    def readline(self, __size: int | None = None) -> bytes:
        return self._fifo_out_file.readline(__size)  # type: ignore[union-attr]

    def write(self, __buffer: bytes) -> int:
        return self._fifo_in_file.write(__buffer)  # type: ignore[union-attr]

    def flush_write(self) -> None:
        if self._fifo_in_file:
            self._fifo_in_file.flush()

    def flush_read(self) -> None:
        if self._fifo_out_file:
            self._fifo_out_file.flush()