openrvdas.logger.writers.socket_writer

 1#!/usr/bin/env python3
 2import os
 3import socket
 4import tempfile
 5import atexit
 6import threading
 7from typing import Any, Dict
 8
 9# Add parent directory to path
10from logger.writers.writer import Writer  # noqa: E402
11
12# Global reference counter for channels
13_channel_refs: Dict[str, int] = {}
14_lock = threading.Lock()
15
16
17class SocketWriter(Writer):
18    """Writer class for socket-based IPC mechanism.
19
20    Writes records to a Unix domain socket.
21    """
22
23    def __init__(self, channel: str, **kwargs):
24        """Initialize a Writer for the specified channel.
25
26        Args:
27            channel: A string identifier for the communication channel
28        """
29        super().__init__(**kwargs)  # processes 'quiet' and type hints
30
31        # Create a unique socket path based on the channel name
32        import hashlib
33        channel_hash = hashlib.md5(channel.encode()).hexdigest()[:8]
34        self.channel = channel
35
36        # Set socket path in temp directory to avoid permission issues
37        temp_dir = tempfile.gettempdir()
38        self.socket_path = os.path.join(temp_dir, f'ipc_socket_{channel_hash}')
39
40        # Socket for sending data
41        self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
42
43        # Update reference counter
44        with _lock:
45            _channel_refs[self.channel] = _channel_refs.get(self.channel, 0) + 1
46
47        # Register cleanup on exit
48        atexit.register(self.close)
49
50    def write(self, record: Any) -> bool:
51        """Write a record to the channel.
52
53        If no Reader is waiting, the record is discarded.
54
55        Args:
56            record: The data to write (will be converted to bytes)
57
58        Returns:
59            bool: True if data was written, False if it was discarded
60        """
61        if isinstance(record, str):
62            record = record.encode('utf-8')
63        elif not isinstance(record, bytes):
64            record = str(record).encode('utf-8')
65
66        # Check if the socket path exists (indicating a receiver)
67        if not os.path.exists(self.socket_path):
68            # No reader is waiting, discard the message
69            return False
70
71        # Send the record
72        try:
73            self.sock.sendto(record, self.socket_path)
74            return True
75        except (ConnectionRefusedError, FileNotFoundError):
76            # No reader is available
77            return False
78
79    def close(self):
80        """Clean up resources."""
81        if not hasattr(self, 'sock') or self.sock is None:
82            # Already closed
83            return
84
85        self.sock.close()
86        self.sock = None
87
88        # Decrement reference counter
89        with _lock:
90            _channel_refs[self.channel] = _channel_refs.get(self.channel, 1) - 1
91
92    def __del__(self):
93        """Ensure cleanup happens."""
94        self.close()