tiphys.ipc.server

  1"""
  2IPC Server implementation.
  3
  4Manages IPC connections and routes messages between host and clients.
  5"""
  6
  7import asyncio
  8import contextlib
  9import json
 10import os
 11from collections.abc import Callable
 12from typing import Any
 13
 14import structlog
 15
 16from tiphys.ipc.protocol import IPCRequest, IPCResponse, MessageType
 17
 18logger = structlog.get_logger(__name__)
 19
 20
 21class IPCServer:
 22    """
 23    Server for handling local IPC connections via Unix domain sockets.
 24    """
 25
 26    def __init__(self, socket_path: str = "/tmp/tiphys.sock"):
 27        self.socket_path = socket_path
 28        self._server: asyncio.AbstractServer | None = None
 29        self._handlers: dict[str, Callable[..., Any]] = {}
 30        self._running = False
 31
 32    async def start(self) -> None:
 33        """Start the IPC server."""
 34        if self._running:
 35            return
 36
 37        # Clean up old socket
 38        if os.path.exists(self.socket_path):
 39            os.remove(self.socket_path)
 40
 41        self._server = await asyncio.start_unix_server(self._handle_client, path=self.socket_path)
 42        self._running = True
 43        logger.info("IPC Server started", path=self.socket_path)
 44
 45    async def stop(self) -> None:
 46        """Stop the IPC server."""
 47        self._running = False
 48        if self._server:
 49            self._server.close()
 50            await self._server.wait_closed()
 51            self._server = None
 52
 53        if os.path.exists(self.socket_path):
 54            with contextlib.suppress(OSError):
 55                os.remove(self.socket_path)
 56        logger.info("IPC Server stopped")
 57
 58    def register_handler(self, method: str, handler: Callable[..., Any]) -> None:
 59        """Register a handler for a specific method."""
 60        self._handlers[method] = handler
 61
 62    async def _handle_client(
 63        self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter
 64    ) -> None:
 65        """Handle individual client connection."""
 66        peer = writer.get_extra_info("peername")
 67        logger.debug("IPC Client connected", peer=peer)
 68
 69        try:
 70            while self._running:
 71                data = await reader.readline()
 72                if not data:
 73                    break
 74
 75                try:
 76                    message_dict = json.loads(data.decode().strip())
 77                    await self._process_message(message_dict, writer)
 78                except json.JSONDecodeError:
 79                    logger.warning("Invalid JSON received")
 80                except Exception as e:
 81                    logger.error("Error processing message", error=str(e))
 82
 83        except Exception as e:
 84            logger.error("IPC Connection error", error=str(e))
 85        finally:
 86            writer.close()
 87            await writer.wait_closed()
 88
 89    async def _process_message(self, data: dict[str, Any], writer: asyncio.StreamWriter) -> None:
 90        """Process an incoming IPC message."""
 91        try:
 92            # Determine message type and validate
 93            msg_type = data.get("type")
 94
 95            if msg_type == MessageType.REQUEST:
 96                request = IPCRequest(**data)
 97                response = await self._handle_request(request)
 98
 99                # Send response
100                response_data = response.model_dump_json() + "\n"
101                writer.write(response_data.encode())
102                await writer.drain()
103
104            elif msg_type == MessageType.EVENT:
105                # Handle event (fire and forget)
106                # TODO: Implement event bus routing
107                pass
108
109        except Exception as e:
110            logger.error("Failed to process IPC message", error=str(e))
111            # Try to send error response if possible
112            if data.get("type") == MessageType.REQUEST and "id" in data:
113                error_response = IPCResponse(
114                    request_id=data["id"], type=MessageType.ERROR, error=str(e)
115                )
116                writer.write((error_response.model_dump_json() + "\n").encode())
117                await writer.drain()
118
119    async def _handle_request(self, request: IPCRequest) -> IPCResponse:
120        """Execute the requested method."""
121        handler = self._handlers.get(request.method)
122        if not handler:
123            return IPCResponse(request_id=request.id, error=f"Method '{request.method}' not found")
124
125        try:
126            # Call handler (support both sync and async)
127            if asyncio.iscoroutinefunction(handler):
128                result = await handler(**request.params)
129            else:
130                result = handler(**request.params)
131
132            return IPCResponse(request_id=request.id, result=result)
133        except Exception as e:
134            logger.error("Handler error", method=request.method, error=str(e))
135            return IPCResponse(request_id=request.id, error=str(e))