Skip to content

Instantly share code, notes, and snippets.

@bencord0
Created December 12, 2012 23:36
Show Gist options
  • Save bencord0/4272720 to your computer and use it in GitHub Desktop.
Save bencord0/4272720 to your computer and use it in GitHub Desktop.
ListeningSocketHandler ====================== A python logging handler. logging.handlers.SocketHandler is a TCP client that sends log records to a TCP server. This class is the opposite. When a TCP client connects (e.g. telnet or netcat), new log records are sent through the tcp connection.
#!/usr/bin/env python
"""
ListeningSocketHandler
======================
A python logging handler.
logging.handlers.SocketHandler is a TCP Socket client that sends log
records to a tcp server.
This class is the opposite.
When a TCP client connects (e.g. telnet or netcat), new log records
are sent through the tcp connection.
"""
import logging
import socket
import sys
import threading
import unittest
class ListeningSocketHandler(logging.Handler):
def __init__(self, port=0, ipv6=False):
super(ListeningSocketHandler, self).__init__()
if ipv6:
a = socket.socket(socket.AF_INET6)
a.bind(("::", port))
else:
a = socket.socket(socket.AF_INET)
a.bind(("0.0.0.0", port))
self._acceptor = a
self._acceptor.listen(1)
self._conn = None
def start_listening(tsh):
while True:
try:
conn, addr = tsh._acceptor.accept()
tsh._conn = conn.makefile('w')
except socket.error:
pass
self._accept_thread = threading.Thread(target=start_listening, args=(self,))
self._accept_thread.daemon = True
self._accept_thread.start()
def emit(self, record):
if self._conn is None:
# Silently drop the log
return
try:
self._conn.write(record.getMessage())
self._conn.write("\n")
self._conn.flush()
except socket.error:
self._conn = None
def flush(self):
if self._conn:
self._conn.flush()
def getsockname(self):
return self._acceptor.getsockname()
class TestListeningSocketHandler(unittest.TestCase):
def setUp(self):
self.log = logging.getLogger()
# default will bind to any unsed port.
self.lsh = ListeningSocketHandler()
self.log.addHandler(self.lsh)
def test_send_message(self):
self.log.warn("Sending a warning")
def test_getsockport(self):
self.assertIsInstance(self.lsh.getsockname()[1], int)
def test_recieve_message(self):
client = socket.socket()
client.connect(("localhost", self.lsh.getsockname()[1]))
self.log.warn("Sending a warning")
buf = client.recv(4096)
self.assertEqual(buf, "Sending a warning\n")
if __name__ == '__main__':
unittest.main()
import logging
from listeningsockethandler import ListeningSocketHandler
log = logging.getLogger()
log.setLevel(logging.DEBUG)
sh = logging.StreamHandler()
sh.setLevel(logging.ERROR)
lh1 = ListeningSocketHandler(1234)
lh1.setLevel(logging.INFO)
lh2 = ListeningSocketHandler(1235)
lh2.setLevel(logging.DEBUG)
print ("ListeningSocketHandler: bound to {0[0]} : {0[1]}".format(lh1.getsockname()))
print ("ListeningSocketHandler: bound to {0[0]} : {0[1]}".format(lh2.getsockname()))
log.addHandler(lh1)
log.addHandler(lh2)
log.addHandler(sh)
def test_log():
log.debug("debug")
log.info("info")
log.warn("warn")
log.error("error")
log.critical("critical")
from time import sleep
while True:
test_log()
sleep(3)
@leadscloud
Copy link

how to set a server ?

@houming818
Copy link

nc -kl 1234

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment