meerschaum.utils.daemon.StdinFile

  1#! /usr/bin/env python3
  2# -*- coding: utf-8 -*-
  3# vim:fenc=utf-8
  4
  5"""
  6Create a file manager to pass STDIN to the Daemon.
  7"""
  8
  9import io
 10import pathlib
 11import time
 12import os
 13import selectors
 14import traceback
 15
 16import meerschaum as mrsm
 17from meerschaum.utils.typing import Optional, Union
 18from meerschaum.utils.warnings import warn
 19
 20
 21class StdinFile(io.TextIOBase):
 22    """
 23    Redirect user input into a Daemon's context.
 24    """
 25    def __init__(
 26        self,
 27        file_path: Union[pathlib.Path, str],
 28        lock_file_path: Optional[pathlib.Path] = None,
 29        decode: bool = True,
 30        refresh_seconds: Union[int, float, None] = None,
 31    ):
 32        if isinstance(file_path, str):
 33            file_path = pathlib.Path(file_path)
 34
 35        self.file_path = file_path
 36        self.blocking_file_path = (
 37            lock_file_path
 38            if lock_file_path is not None
 39            else (file_path.parent / (file_path.name + '.block'))
 40        )
 41        self._file_handler = None
 42        self._fd = None
 43        self.sel = selectors.DefaultSelector()
 44        self.decode = decode
 45        self._write_fp = None
 46        self._refresh_seconds = refresh_seconds
 47
 48    @property
 49    def encoding(self):
 50        return 'utf-8'
 51
 52    @property
 53    def file_handler(self):
 54        """
 55        Return the read file handler to the provided file path.
 56        """
 57        if self._file_handler is not None:
 58            return self._file_handler
 59
 60        if not self.file_path.exists():
 61            self.file_path.parent.mkdir(parents=True, exist_ok=True)
 62            os.mkfifo(self.file_path.as_posix(), mode=0o600)
 63
 64        self._fd = os.open(self.file_path, os.O_RDONLY | os.O_NONBLOCK)
 65        self._file_handler = os.fdopen(self._fd, 'rb', buffering=0)
 66        self.sel.register(self._file_handler, selectors.EVENT_READ)
 67        return self._file_handler
 68
 69    def write(self, data):
 70        if self._write_fp is None:
 71            self.file_path.parent.mkdir(parents=True, exist_ok=True)
 72            if not self.file_path.exists():
 73                os.mkfifo(self.file_path.as_posix(), mode=0o600)
 74            self._write_fp = open(self.file_path, 'wb')
 75
 76        if isinstance(data, str):
 77            data = data.encode('utf-8')
 78        try:
 79            self._write_fp.write(data)
 80            self._write_fp.flush()
 81        except BrokenPipeError:
 82            pass
 83
 84    def fileno(self):
 85        fileno = self.file_handler.fileno()
 86        return fileno
 87
 88    def read(self, size=-1):
 89        """
 90        Read from the FIFO pipe, blocking on EOFError.
 91        """
 92        _ = self.file_handler
 93        while True:
 94            try:
 95                data = self._file_handler.read(size)
 96                if data:
 97                    try:
 98                        if self.blocking_file_path.exists():
 99                            self.blocking_file_path.unlink()
100                    except Exception:
101                        warn(traceback.format_exc())
102                    return data.decode('utf-8') if self.decode else data
103            except (OSError, EOFError):
104                pass
105
106            if not self.blocking_file_path.exists():
107                self.blocking_file_path.touch()
108            time.sleep(self.refresh_seconds)
109
110    def readline(self, size=-1):
111        line = '' if self.decode else b''
112        while True:
113            data = self.read(1)
114            if not data or ((data == '\n') if self.decode else (data == b'\n')):
115                break
116            line += data
117
118        return line
119
120    def close(self):
121        if self._file_handler is not None:
122            self.sel.unregister(self._file_handler)
123            self._file_handler.close()
124            try:
125                os.close(self._fd)
126            except OSError:
127                pass
128            self._file_handler = None
129            self._fd = None
130
131        if self._write_fp is not None:
132            try:
133                self._write_fp.close()
134            except BrokenPipeError:
135                pass
136            self._write_fp = None
137
138        try:
139            if self.blocking_file_path.exists():
140                self.blocking_file_path.unlink()
141        except Exception:
142            pass
143        super().close()
144
145    def is_open(self):
146        return self._file_handler is not None
147
148    def isatty(self) -> bool:
149        return False
150
151    @property
152    def refresh_seconds(self) -> Union[int, float]:
153        """
154        How many seconds between checking for blocking functions.
155        """
156        if not self._refresh_seconds:
157            self._refresh_seconds = mrsm.get_config('system', 'cli', 'refresh_seconds')
158        return self._refresh_seconds
159
160    def __str__(self) -> str:
161        return f"StdinFile('{self.file_path}')"
162
163    def __repr__(self) -> str:
164        return str(self)