Test connection to Syslog server. Returns: True if connection successful
(self)
| 213 | def _send_tcp(self, message: bytes) -> None: |
| 214 | """Send message via TCP/TLS.""" |
| 215 | sock = self._get_socket() |
| 216 | sock.sendall(message) |
| 217 | |
| 218 | async def test_connection(self) -> bool: |
| 219 | """Test connection to Syslog server. |
| 220 | |
| 221 | Returns: |
| 222 | True if connection successful |
| 223 | """ |
| 224 | try: |
| 225 | if self.protocol == "udp": |
| 226 | # UDP is connectionless, just verify we can create socket |
| 227 | sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) |
| 228 | sock.close() |
| 229 | return True |
| 230 | else: |
| 231 | # Try to connect |
| 232 | loop = asyncio.get_event_loop() |
| 233 | await loop.run_in_executor(None, self._get_socket) |
| 234 | return True |
| 235 | |
| 236 | except Exception as e: |
| 237 | self.logger.error(f"Syslog connection test failed: {e}") |
| 238 | return False |