268 lines
7.5 KiB
Python
268 lines
7.5 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
from pathlib import Path
|
|
import threading
|
|
from typing import Any
|
|
|
|
from services.shared.core import utc_now_iso
|
|
|
|
|
|
class BridgeState:
|
|
def __init__(self) -> None:
|
|
self._lock = threading.Lock()
|
|
self.ami_connected = False
|
|
self.last_event_at: str | None = None
|
|
self.last_error: str | None = None
|
|
self.reconnect_requested = False
|
|
|
|
def snapshot(self) -> dict[str, Any]:
|
|
with self._lock:
|
|
return {
|
|
"ami_connected": self.ami_connected,
|
|
"last_event_at": self.last_event_at,
|
|
"last_error": self.last_error,
|
|
"reconnect_requested": self.reconnect_requested,
|
|
}
|
|
|
|
def set_connected(self, value: bool) -> None:
|
|
with self._lock:
|
|
self.ami_connected = value
|
|
if value:
|
|
self.last_error = None
|
|
|
|
def set_last_event(self) -> None:
|
|
with self._lock:
|
|
self.last_event_at = utc_now_iso()
|
|
|
|
def set_error(self, message: str) -> None:
|
|
with self._lock:
|
|
self.last_error = message
|
|
|
|
def request_reconnect(self) -> None:
|
|
with self._lock:
|
|
self.reconnect_requested = True
|
|
|
|
def take_reconnect(self) -> bool:
|
|
with self._lock:
|
|
current = self.reconnect_requested
|
|
self.reconnect_requested = False
|
|
return current
|
|
|
|
|
|
_BACKGROUND_LOCK = threading.Lock()
|
|
_BACKGROUND_STOP_EVENT: threading.Event | None = None
|
|
_BACKGROUND_THREADS: list[threading.Thread] = []
|
|
_BRIDGE_SINGLETON_DB_CONNECTION = None
|
|
_BRIDGE_SINGLETON_FILE_HANDLE = None
|
|
|
|
|
|
def bridge_singleton_lock_path(*, database_url: str, cc_data_dir: str) -> Path:
|
|
if database_url.startswith("sqlite:///"):
|
|
db_path = Path(database_url[len("sqlite:///") :]).resolve()
|
|
db_path.parent.mkdir(parents=True, exist_ok=True)
|
|
return db_path.parent / "asterisk_bridge_service.singleton.lock"
|
|
data_dir = Path(cc_data_dir).resolve()
|
|
data_dir.mkdir(parents=True, exist_ok=True)
|
|
return data_dir / "asterisk_bridge_service.singleton.lock"
|
|
|
|
|
|
def try_lock_file(handle) -> bool:
|
|
try:
|
|
handle.seek(0)
|
|
handle.write(b"0")
|
|
handle.flush()
|
|
handle.seek(0)
|
|
except OSError:
|
|
return False
|
|
if os.name == "nt":
|
|
import msvcrt
|
|
|
|
try:
|
|
msvcrt.locking(handle.fileno(), msvcrt.LK_NBLCK, 1)
|
|
return True
|
|
except OSError:
|
|
return False
|
|
|
|
import fcntl
|
|
|
|
try:
|
|
fcntl.flock(handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
|
|
return True
|
|
except OSError:
|
|
return False
|
|
|
|
|
|
def unlock_file(handle) -> None:
|
|
handle.seek(0)
|
|
if os.name == "nt":
|
|
import msvcrt
|
|
|
|
msvcrt.locking(handle.fileno(), msvcrt.LK_UNLCK, 1)
|
|
return
|
|
|
|
import fcntl
|
|
|
|
fcntl.flock(handle.fileno(), fcntl.LOCK_UN)
|
|
|
|
|
|
def try_acquire_bridge_singleton_guard(
|
|
*,
|
|
database_url: str,
|
|
engine: Any,
|
|
lock_key: int,
|
|
cc_data_dir: str,
|
|
) -> bool:
|
|
global _BRIDGE_SINGLETON_DB_CONNECTION, _BRIDGE_SINGLETON_FILE_HANDLE
|
|
|
|
if _BRIDGE_SINGLETON_DB_CONNECTION is not None or _BRIDGE_SINGLETON_FILE_HANDLE is not None:
|
|
return True
|
|
|
|
if database_url.startswith("postgresql"):
|
|
connection = engine.raw_connection()
|
|
try:
|
|
cursor = connection.cursor()
|
|
try:
|
|
cursor.execute("SELECT pg_try_advisory_lock(%s)", (lock_key,))
|
|
acquired_row = cursor.fetchone()
|
|
finally:
|
|
cursor.close()
|
|
connection.commit()
|
|
except Exception:
|
|
connection.close()
|
|
raise
|
|
if acquired_row and bool(acquired_row[0]):
|
|
_BRIDGE_SINGLETON_DB_CONNECTION = connection
|
|
return True
|
|
connection.close()
|
|
return False
|
|
|
|
lock_path = bridge_singleton_lock_path(database_url=database_url, cc_data_dir=cc_data_dir)
|
|
lock_path.touch(exist_ok=True)
|
|
handle = lock_path.open("r+b")
|
|
if try_lock_file(handle):
|
|
_BRIDGE_SINGLETON_FILE_HANDLE = handle
|
|
return True
|
|
try:
|
|
handle.close()
|
|
except OSError:
|
|
pass
|
|
return False
|
|
|
|
|
|
def release_bridge_singleton_guard(*, lock_key: int) -> None:
|
|
global _BRIDGE_SINGLETON_DB_CONNECTION, _BRIDGE_SINGLETON_FILE_HANDLE
|
|
|
|
if _BRIDGE_SINGLETON_DB_CONNECTION is not None:
|
|
connection = _BRIDGE_SINGLETON_DB_CONNECTION
|
|
_BRIDGE_SINGLETON_DB_CONNECTION = None
|
|
try:
|
|
cursor = connection.cursor()
|
|
try:
|
|
cursor.execute("SELECT pg_advisory_unlock(%s)", (lock_key,))
|
|
finally:
|
|
cursor.close()
|
|
connection.commit()
|
|
finally:
|
|
connection.close()
|
|
|
|
if _BRIDGE_SINGLETON_FILE_HANDLE is not None:
|
|
handle = _BRIDGE_SINGLETON_FILE_HANDLE
|
|
_BRIDGE_SINGLETON_FILE_HANDLE = None
|
|
try:
|
|
unlock_file(handle)
|
|
finally:
|
|
handle.close()
|
|
|
|
|
|
def bridge_singleton_db_connection():
|
|
return _BRIDGE_SINGLETON_DB_CONNECTION
|
|
|
|
|
|
def bridge_singleton_file_handle():
|
|
return _BRIDGE_SINGLETON_FILE_HANDLE
|
|
|
|
|
|
def background_threads_alive() -> list[threading.Thread]:
|
|
return [thread for thread in _BACKGROUND_THREADS if thread.is_alive()]
|
|
|
|
|
|
def background_stop_event() -> threading.Event:
|
|
global _BACKGROUND_STOP_EVENT
|
|
if _BACKGROUND_STOP_EVENT is None or _BACKGROUND_STOP_EVENT.is_set():
|
|
_BACKGROUND_STOP_EVENT = threading.Event()
|
|
return _BACKGROUND_STOP_EVENT
|
|
|
|
|
|
def background_shutdown_timeout_seconds(*, poll_interval: float, failed_retry_interval_seconds: float) -> float:
|
|
return max(10.0, poll_interval, failed_retry_interval_seconds) + 1.0
|
|
|
|
|
|
def start_background_threads(
|
|
*,
|
|
ami_loop,
|
|
failed_retry_loop,
|
|
ensure_guard,
|
|
extra_loops: list[tuple[str, Any]] | None = None,
|
|
) -> None:
|
|
global _BACKGROUND_THREADS
|
|
|
|
with _BACKGROUND_LOCK:
|
|
alive = background_threads_alive()
|
|
if alive:
|
|
_BACKGROUND_THREADS = alive
|
|
ensure_guard()
|
|
return
|
|
|
|
stop_event = background_stop_event()
|
|
ensure_guard()
|
|
threads = [
|
|
threading.Thread(target=ami_loop, args=(stop_event,), daemon=True, name="asterisk-ami-loop"),
|
|
threading.Thread(
|
|
target=failed_retry_loop,
|
|
args=(stop_event,),
|
|
daemon=True,
|
|
name="asterisk-failed-retry-loop",
|
|
),
|
|
]
|
|
for name, target in extra_loops or []:
|
|
threads.append(threading.Thread(target=target, args=(stop_event,), daemon=True, name=name))
|
|
_BACKGROUND_THREADS = threads
|
|
try:
|
|
for thread in threads:
|
|
thread.start()
|
|
except Exception:
|
|
stop_event.set()
|
|
_BACKGROUND_THREADS = []
|
|
raise
|
|
|
|
|
|
def stop_background_threads(
|
|
*,
|
|
state: BridgeState,
|
|
release_guard,
|
|
shutdown_timeout_seconds: float,
|
|
) -> None:
|
|
global _BACKGROUND_STOP_EVENT, _BACKGROUND_THREADS
|
|
|
|
with _BACKGROUND_LOCK:
|
|
stop_event = _BACKGROUND_STOP_EVENT
|
|
threads = list(_BACKGROUND_THREADS)
|
|
if stop_event is not None:
|
|
stop_event.set()
|
|
state.request_reconnect()
|
|
|
|
for thread in threads:
|
|
if thread.is_alive():
|
|
thread.join(timeout=shutdown_timeout_seconds)
|
|
|
|
with _BACKGROUND_LOCK:
|
|
alive = background_threads_alive()
|
|
if alive:
|
|
_BACKGROUND_THREADS = alive
|
|
return
|
|
_BACKGROUND_THREADS = []
|
|
_BACKGROUND_STOP_EVENT = None
|
|
state.set_connected(False)
|
|
release_guard()
|