Skip to content

Instantly share code, notes, and snippets.

@lwzm
Last active June 17, 2021 03:38
Show Gist options
  • Save lwzm/dd883227543c24c5b32851124386aead to your computer and use it in GitHub Desktop.
Save lwzm/dd883227543c24c5b32851124386aead to your computer and use it in GitHub Desktop.
from json import dumps
from traceback import print_exc
class TcpLog:
def __init__(self, addr):
host, port = addr.split(":")
self._address = host, int(port)
self._socket = None
self._message = b''
def __call__(self, **kw):
msg = dumps(kw, ensure_ascii=False, default=str)
self._message += msg.encode() + b'\n'
if self._socket is None:
from socket import AF_INET, SOCK_STREAM, socket
s = socket(AF_INET, SOCK_STREAM)
try:
s.connect(self._address)
self._socket = s
except Exception as e:
print_exc()
return
try:
self._socket.sendall(self._message)
self._socket.sendall(b'') # raise if remote is closed
self._message = b''
except Exception:
self._socket.close()
self._socket = None
print_exc()
from json import dumps
from sys import stderr
from traceback import print_exc
class TcpLog:
def __init__(self, addr=None):
self._address = None
self._socket = None
self._message = b''
if addr is None:
from os import getenv
addr = getenv("TCPLOG")
if addr:
host, port = addr.split(":")
self._address = host, int(port)
def __call__(self, **kw):
msg = dumps(kw, ensure_ascii=False, default=str)
if self._address is None:
print(msg, file=stderr, flush=True)
return
self._message += msg.encode() + b'\n'
if self._socket is None:
from socket import AF_INET, SOCK_STREAM, socket
s = socket(AF_INET, SOCK_STREAM)
try:
s.connect(self._address)
self._socket = s
except Exception as e:
print_exc()
return
try:
self._socket.sendall(self._message)
self._socket.sendall(b'') # raise if remote is closed
self._message = b''
except Exception:
self._socket.close()
self._socket = None
print_exc()
log = TcpLog("tyio.net:1111")
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment