Complete Phase 1: parallel sync, IPC, theme colors, lazy CLI loading
- Sync: Parallelize message downloads with asyncio.gather (batch size 5) - Sync: Increase HTTP semaphore from 2 to 5 concurrent requests - Sync: Add IPC notifications to sync daemon after sync completes - Mail: Replace all hardcoded RGB colors with theme variables - Mail: Remove envelope icon/checkbox gap (padding cleanup) - Mail: Add IPC listener for refresh notifications from sync - Calendar: Style current time line with error color and solid line - Tasks: Fix table not displaying (CSS grid to horizontal layout) - CLI: Implement lazy command loading for faster startup (~12s to ~0.3s) - Add PROJECT_PLAN.md with full improvement roadmap - Add src/utils/ipc.py for Unix socket cross-app communication
This commit is contained in:
Bendt
318
src/utils/ipc.py
Normal file
318
src/utils/ipc.py
Normal file
@@ -0,0 +1,318 @@
|
||||
"""Inter-Process Communication using Unix Domain Sockets.
|
||||
|
||||
This module provides a simple pub/sub mechanism for cross-app notifications.
|
||||
The sync daemon can broadcast messages when data changes, and TUI apps can
|
||||
listen for these messages to refresh their displays.
|
||||
|
||||
Usage:
|
||||
# In sync daemon (publisher):
|
||||
from src.utils.ipc import notify_refresh
|
||||
await notify_refresh("mail") # Notify mail app to refresh
|
||||
await notify_refresh("calendar") # Notify calendar app to refresh
|
||||
await notify_refresh("tasks") # Notify tasks app to refresh
|
||||
|
||||
# In TUI apps (subscriber):
|
||||
from src.utils.ipc import IPCListener
|
||||
|
||||
class MyApp(App):
|
||||
def on_mount(self):
|
||||
self.ipc_listener = IPCListener("mail", self.on_refresh)
|
||||
self.ipc_listener.start()
|
||||
|
||||
def on_unmount(self):
|
||||
self.ipc_listener.stop()
|
||||
|
||||
async def on_refresh(self, message):
|
||||
# Refresh the app's data
|
||||
await self.refresh_data()
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import os
|
||||
import socket
|
||||
import threading
|
||||
from pathlib import Path
|
||||
from typing import Callable, Optional, Any, Dict
|
||||
|
||||
# Socket paths for each app type
|
||||
SOCKET_DIR = Path("~/.local/share/luk/ipc").expanduser()
|
||||
SOCKET_PATHS = {
|
||||
"mail": SOCKET_DIR / "mail.sock",
|
||||
"calendar": SOCKET_DIR / "calendar.sock",
|
||||
"tasks": SOCKET_DIR / "tasks.sock",
|
||||
}
|
||||
|
||||
|
||||
def ensure_socket_dir():
|
||||
"""Ensure the socket directory exists."""
|
||||
SOCKET_DIR.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
|
||||
def get_socket_path(app_type: str) -> Path:
|
||||
"""Get the socket path for a given app type."""
|
||||
if app_type not in SOCKET_PATHS:
|
||||
raise ValueError(
|
||||
f"Unknown app type: {app_type}. Must be one of: {list(SOCKET_PATHS.keys())}"
|
||||
)
|
||||
return SOCKET_PATHS[app_type]
|
||||
|
||||
|
||||
class IPCMessage:
|
||||
"""A message sent via IPC."""
|
||||
|
||||
def __init__(self, event: str, data: Optional[Dict[str, Any]] = None):
|
||||
self.event = event
|
||||
self.data = data or {}
|
||||
|
||||
def to_json(self) -> str:
|
||||
return json.dumps({"event": self.event, "data": self.data})
|
||||
|
||||
@classmethod
|
||||
def from_json(cls, json_str: str) -> "IPCMessage":
|
||||
parsed = json.loads(json_str)
|
||||
return cls(event=parsed["event"], data=parsed.get("data", {}))
|
||||
|
||||
|
||||
async def notify_refresh(app_type: str, data: Optional[Dict[str, Any]] = None) -> bool:
|
||||
"""Send a refresh notification to a specific app.
|
||||
|
||||
Args:
|
||||
app_type: The type of app to notify ("mail", "calendar", "tasks")
|
||||
data: Optional data to include with the notification
|
||||
|
||||
Returns:
|
||||
True if the notification was sent successfully, False otherwise
|
||||
"""
|
||||
socket_path = get_socket_path(app_type)
|
||||
|
||||
if not socket_path.exists():
|
||||
# No listener, that's okay
|
||||
return False
|
||||
|
||||
try:
|
||||
message = IPCMessage("refresh", data)
|
||||
|
||||
# Connect to the socket and send the message
|
||||
reader, writer = await asyncio.open_unix_connection(str(socket_path))
|
||||
|
||||
writer.write((message.to_json() + "\n").encode())
|
||||
await writer.drain()
|
||||
|
||||
writer.close()
|
||||
await writer.wait_closed()
|
||||
|
||||
return True
|
||||
except (ConnectionRefusedError, FileNotFoundError, OSError):
|
||||
# Socket exists but no one is listening, or other error
|
||||
return False
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
|
||||
async def notify_all(data: Optional[Dict[str, Any]] = None) -> Dict[str, bool]:
|
||||
"""Send a refresh notification to all apps.
|
||||
|
||||
Args:
|
||||
data: Optional data to include with the notification
|
||||
|
||||
Returns:
|
||||
Dictionary of app_type -> success status
|
||||
"""
|
||||
results = {}
|
||||
for app_type in SOCKET_PATHS:
|
||||
results[app_type] = await notify_refresh(app_type, data)
|
||||
return results
|
||||
|
||||
|
||||
class IPCListener:
|
||||
"""Listens for IPC messages on a Unix socket.
|
||||
|
||||
Usage:
|
||||
listener = IPCListener("mail", on_message_callback)
|
||||
listener.start()
|
||||
# ... later ...
|
||||
listener.stop()
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
app_type: str,
|
||||
callback: Callable[[IPCMessage], Any],
|
||||
):
|
||||
"""Initialize the IPC listener.
|
||||
|
||||
Args:
|
||||
app_type: The type of app ("mail", "calendar", "tasks")
|
||||
callback: Function to call when a message is received.
|
||||
Can be sync or async.
|
||||
"""
|
||||
self.app_type = app_type
|
||||
self.callback = callback
|
||||
self.socket_path = get_socket_path(app_type)
|
||||
self._server: Optional[asyncio.AbstractServer] = None
|
||||
self._running = False
|
||||
self._loop: Optional[asyncio.AbstractEventLoop] = None
|
||||
self._thread: Optional[threading.Thread] = None
|
||||
|
||||
async def _handle_client(
|
||||
self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter
|
||||
):
|
||||
"""Handle an incoming client connection."""
|
||||
try:
|
||||
data = await reader.readline()
|
||||
if data:
|
||||
message_str = data.decode().strip()
|
||||
if message_str:
|
||||
message = IPCMessage.from_json(message_str)
|
||||
|
||||
# Call the callback (handle both sync and async)
|
||||
result = self.callback(message)
|
||||
if asyncio.iscoroutine(result):
|
||||
await result
|
||||
except Exception:
|
||||
pass # Ignore errors from malformed messages
|
||||
finally:
|
||||
writer.close()
|
||||
try:
|
||||
await writer.wait_closed()
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
async def _run_server(self):
|
||||
"""Run the Unix socket server."""
|
||||
ensure_socket_dir()
|
||||
|
||||
# Remove stale socket file if it exists
|
||||
if self.socket_path.exists():
|
||||
self.socket_path.unlink()
|
||||
|
||||
self._server = await asyncio.start_unix_server(
|
||||
self._handle_client, path=str(self.socket_path)
|
||||
)
|
||||
|
||||
# Set socket permissions (readable/writable by owner only)
|
||||
os.chmod(self.socket_path, 0o600)
|
||||
|
||||
async with self._server:
|
||||
await self._server.serve_forever()
|
||||
|
||||
def _run_in_thread(self):
|
||||
"""Run the event loop in a separate thread."""
|
||||
self._loop = asyncio.new_event_loop()
|
||||
asyncio.set_event_loop(self._loop)
|
||||
try:
|
||||
self._loop.run_until_complete(self._run_server())
|
||||
except asyncio.CancelledError:
|
||||
pass
|
||||
finally:
|
||||
self._loop.close()
|
||||
|
||||
def start(self):
|
||||
"""Start listening for IPC messages in a background thread."""
|
||||
if self._running:
|
||||
return
|
||||
|
||||
self._running = True
|
||||
self._thread = threading.Thread(target=self._run_in_thread, daemon=True)
|
||||
self._thread.start()
|
||||
|
||||
def stop(self):
|
||||
"""Stop listening for IPC messages."""
|
||||
if not self._running:
|
||||
return
|
||||
|
||||
self._running = False
|
||||
|
||||
# Cancel the server
|
||||
if self._server and self._loop:
|
||||
self._loop.call_soon_threadsafe(self._server.close)
|
||||
|
||||
# Stop the event loop
|
||||
if self._loop:
|
||||
self._loop.call_soon_threadsafe(self._loop.stop)
|
||||
|
||||
# Wait for thread to finish
|
||||
if self._thread:
|
||||
self._thread.join(timeout=1.0)
|
||||
|
||||
# Clean up socket file
|
||||
if self.socket_path.exists():
|
||||
try:
|
||||
self.socket_path.unlink()
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
class AsyncIPCListener:
|
||||
"""Async version of IPCListener for use within an existing event loop.
|
||||
|
||||
Usage in a Textual app:
|
||||
class MyApp(App):
|
||||
async def on_mount(self):
|
||||
self.ipc_listener = AsyncIPCListener("mail", self.on_refresh)
|
||||
await self.ipc_listener.start()
|
||||
|
||||
async def on_unmount(self):
|
||||
await self.ipc_listener.stop()
|
||||
|
||||
async def on_refresh(self, message):
|
||||
self.refresh_data()
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
app_type: str,
|
||||
callback: Callable[[IPCMessage], Any],
|
||||
):
|
||||
self.app_type = app_type
|
||||
self.callback = callback
|
||||
self.socket_path = get_socket_path(app_type)
|
||||
self._server: Optional[asyncio.AbstractServer] = None
|
||||
|
||||
async def _handle_client(
|
||||
self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter
|
||||
):
|
||||
"""Handle an incoming client connection."""
|
||||
try:
|
||||
data = await reader.readline()
|
||||
if data:
|
||||
message_str = data.decode().strip()
|
||||
if message_str:
|
||||
message = IPCMessage.from_json(message_str)
|
||||
|
||||
result = self.callback(message)
|
||||
if asyncio.iscoroutine(result):
|
||||
await result
|
||||
except Exception:
|
||||
pass
|
||||
finally:
|
||||
writer.close()
|
||||
try:
|
||||
await writer.wait_closed()
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
async def start(self):
|
||||
"""Start the Unix socket server."""
|
||||
ensure_socket_dir()
|
||||
|
||||
if self.socket_path.exists():
|
||||
self.socket_path.unlink()
|
||||
|
||||
self._server = await asyncio.start_unix_server(
|
||||
self._handle_client, path=str(self.socket_path)
|
||||
)
|
||||
os.chmod(self.socket_path, 0o600)
|
||||
|
||||
async def stop(self):
|
||||
"""Stop the server and clean up."""
|
||||
if self._server:
|
||||
self._server.close()
|
||||
await self._server.wait_closed()
|
||||
|
||||
if self.socket_path.exists():
|
||||
try:
|
||||
self.socket_path.unlink()
|
||||
except Exception:
|
||||
pass
|
||||
Reference in New Issue
Block a user