1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
|
# ========================================================================== #
# #
# KVMD - The main PiKVM daemon. #
# #
# Copyright (C) 2018-2024 Maxim Devaev <[email protected]> #
# #
# This program is free software: you can redistribute it and/or modify #
# it under the terms of the GNU General Public License as published by #
# the Free Software Foundation, either version 3 of the License, or #
# (at your option) any later version. #
# #
# This program is distributed in the hope that it will be useful, #
# but WITHOUT ANY WARRANTY; without even the implied warranty of #
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the #
# GNU General Public License for more details. #
# #
# You should have received a copy of the GNU General Public License #
# along with this program. If not, see <https://www.gnu.org/licenses/>. #
# #
# ========================================================================== #
import os
import asyncio
from aiohttp.web import Request
from aiohttp.web import WebSocketResponse
from ...logging import get_logger
from ... import tools
from ... import aiotools
from ... import aiohelpers
from ... import fstab
from ...htserver import exposed_http
from ...htserver import exposed_ws
from ...htserver import WsSession
from ...htserver import HttpServer
# =====
class PstServer(HttpServer): # pylint: disable=too-many-arguments,too-many-instance-attributes
def __init__( # pylint: disable=too-many-arguments,too-many-locals
self,
ro_retries_delay: float,
ro_cleanup_delay: float,
remount_cmd: list[str],
) -> None:
super().__init__()
self.__data_path = fstab.find_pst().root_path
self.__ro_retries_delay = ro_retries_delay
self.__ro_cleanup_delay = ro_cleanup_delay
self.__remount_cmd = remount_cmd
self.__notifier = aiotools.AioNotifier()
# ===== WEBSOCKET
@exposed_http("GET", "/ws")
async def __ws_handler(self, req: Request) -> WebSocketResponse:
async with self._ws_session(req) as ws:
await ws.send_event("loop", {})
return (await self._ws_loop(ws))
@exposed_ws("ping")
async def __ws_ping_handler(self, ws: WsSession, _: dict) -> None:
await ws.send_event("pong", {})
# ===== SYSTEM STUFF
async def _init_app(self) -> None:
if (await self.__remount_storage(rw=True)):
await self.__remount_storage(rw=False)
aiotools.create_deadly_task("Controller", self.__controller())
self._add_exposed(self)
async def _on_shutdown(self) -> None:
logger = get_logger(0)
logger.info("Stopping system tasks ...")
await aiotools.stop_all_deadly_tasks()
logger.info("Disconnecting clients ...")
await self.__broadcast_storage_state(len(self._get_wss()), False)
if (await self._close_all_wss()):
await asyncio.sleep(self.__ro_cleanup_delay)
logger.info("On-Shutdown complete")
async def _on_cleanup(self) -> None:
logger = get_logger(0)
await self.__remount_storage(rw=False)
logger.info("On-Cleanup complete")
async def _on_ws_opened(self) -> None:
self.__notifier.notify()
async def _on_ws_closed(self) -> None:
self.__notifier.notify()
# ===== SYSTEM TASKS
async def __controller(self) -> None:
prev: int = 0
while True:
cur = len(self._get_wss())
if cur > 0:
if not self.__is_write_available():
await self.__remount_storage(rw=True)
elif prev > 0 and cur == 0:
while not (await self.__remount_storage(rw=False)):
if len(self._get_wss()) > 0:
continue
await asyncio.sleep(self.__ro_retries_delay)
await self.__broadcast_storage_state(cur, self.__is_write_available())
prev = cur
await self.__notifier.wait()
async def __broadcast_storage_state(self, clients: int, write_allowed: bool) -> None:
await self._broadcast_ws_event("storage_state", {
"clients": clients,
"data": {
"path": self.__data_path,
"write_allowed": write_allowed,
},
})
def __is_write_available(self) -> bool:
try:
return (not (os.statvfs(self.__data_path).f_flag & os.ST_RDONLY))
except Exception as ex:
get_logger(0).info("Can't get filesystem state of PST (%s): %s",
self.__data_path, tools.efmt(ex))
return False
async def __remount_storage(self, rw: bool) -> bool:
return (await aiohelpers.remount("PST", self.__remount_cmd, rw))
|
