sv/tsun-gen3-proxy
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Revert "fix memory leak on connection aborts"

Browse Source 
This reverts commit f097b3350b.
This commit is contained in:
Stefan Allius
2023-10-03 11:45:17 +02:00
parent 38dacf2b97
commit d308c3a9fa
1 changed files with 23 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files

56
app/src/async_stream.py
View File

@@ -1,24 +1,19 @@
import logging, traceback, json import logging, traceback, aiomqtt, json
#import gc, ctypes
from config import Config from config import Config
from messages import Message, hex_dump_memory from messages import Message, hex_dump_memory
from mqtt import Mqtt from mqtt import Mqtt
logger = logging.getLogger('conn') logger = logging.getLogger('conn')
logger_mqtt = logging.getLogger('mqtt')
#def ref_count(address):
# return ctypes.c_long.from_address(address).value
class AsyncStream(Message): class AsyncStream(Message):
def __init__(self, proxy, reader, writer, addr, weak_stream=None, server_side=True): def __init__(self, proxy, reader, writer, addr, stream=None, server_side=True):
logger.debug (f"AsyncStream __init__ {self}")
super().__init__() super().__init__()
self.proxy = proxy self.proxy = proxy
self.reader = reader self.reader = reader
self.writer = writer self.writer = writer
self.__WkRemoteStream = weak_stream self.remoteStream = stream
self.addr = addr self.addr = addr
self.server_side = server_side self.server_side = server_side
self.mqtt = Mqtt() self.mqtt = Mqtt()
@@ -29,26 +24,26 @@ class AsyncStream(Message):
Our puplic methods Our puplic methods
''' '''
def set_serial_no(self, serial_no : str): def set_serial_no(self, serial_no : str):
logger.debug(f'SerialNo: {serial_no}') logger_mqtt.info(f'SerialNo: {serial_no}')
if self.unique_id != serial_no: if self.unique_id != serial_no:
inverters = Config.get('inverters') inverters = Config.get('inverters')
#logger.debug(f'Inverters: {inverters}') #logger_mqtt.debug(f'Inverters: {inverters}')
if serial_no in inverters: if serial_no in inverters:
logger.debug(f'SerialNo {serial_no} allowed!') logger_mqtt.debug(f'SerialNo {serial_no} allowed!')
inv = inverters[serial_no] inv = inverters[serial_no]
self.node_id = inv['node_id'] self.node_id = inv['node_id']
self.sug_area = inv['suggested_area'] self.sug_area = inv['suggested_area']
else: else:
logger.debug(f'SerialNo {serial_no} not known!') logger_mqtt.debug(f'SerialNo {serial_no} not known!')
self.node_id = '' self.node_id = ''
self.sug_area = '' self.sug_area = ''
if not inverters['allow_all']: if not inverters['allow_all']:
self.unique_id = None self.unique_id = None
logger.error('ignore message from unknow inverter!') logger_mqtt.error('ignore message from unknow inverter!')
return return
self.unique_id = serial_no self.unique_id = serial_no
@@ -63,7 +58,7 @@ class AsyncStream(Message):
if self.server_side: if self.server_side:
try: try:
for data_json, component, id in self.db.ha_confs(self.entitiy_prfx + self.node_id, self.unique_id, self.sug_area): for data_json, component, id in self.db.ha_confs(self.entitiy_prfx + self.node_id, self.unique_id, self.sug_area):
logger.debug(f'Register MQTT: {data_json}') logger_mqtt.debug(f'Register: {data_json}')
await self.mqtt.publish(f"{self.discovery_prfx}{component}/{self.node_id}{id}/config", data_json) await self.mqtt.publish(f"{self.discovery_prfx}{component}/{self.node_id}{id}/config", data_json)
except Exception: except Exception:
@@ -105,14 +100,11 @@ class AsyncStream(Message):
self.writer.close() self.writer.close()
def close(self) -> None: def close(self):
logger.debug(f'in AsyncStream.close() {self.addr}') logger.info(f'in async_stream.close() {self.addr}')
super().close()
self.writer.close() self.writer.close()
del self.proxy self.proxy = None
self.remoteStream = None
#logger.info (f'refcount: {ref_count(id (self))}')
#logger.info (f'AsyncStream refs: {gc.get_referrers(self)}')
''' '''
@@ -135,17 +127,15 @@ class AsyncStream(Message):
async def __async_forward(self) -> None: async def __async_forward(self) -> None:
if self._forward_buffer: if self._forward_buffer:
if not self.__WkRemoteStream: if not self.remoteStream:
tsun = Config.get('tsun') tsun = Config.get('tsun')
self.__WkRemoteStream = await self.proxy.CreateClientStream (tsun['host'], tsun['port']) self.remoteStream = await self.proxy.CreateClientStream (self, tsun['host'], tsun['port'])
if self.__WkRemoteStream: if self.remoteStream:
remoteStream = self.__WkRemoteStream() hex_dump_memory(logging.DEBUG, f'Forward to {self.remoteStream.addr}:', self._forward_buffer, len(self._forward_buffer))
if remoteStream: self.remoteStream.writer.write (self._forward_buffer)
hex_dump_memory(logging.DEBUG, f'Forward to {remoteStream.addr}:', self._forward_buffer, len(self._forward_buffer)) await self.remoteStream.writer.drain()
remoteStream.writer.write (self._forward_buffer) self._forward_buffer = bytearray(0)
await remoteStream.writer.drain()
self._forward_buffer = bytearray(0)
async def __async_publ_mqtt(self) -> None: async def __async_publ_mqtt(self) -> None:
if self.server_side: if self.server_side:
@@ -159,11 +149,11 @@ class AsyncStream(Message):
for key in self.new_data: for key in self.new_data:
if self.new_data[key] and key in db: if self.new_data[key] and key in db:
data_json = json.dumps(db[key]) data_json = json.dumps(db[key])
#logger.info(f'MQTT publish {key}: {data_json}') logger_mqtt.info(f'{key}: {data_json}')
await self.mqtt.publish(f"{self.entitiy_prfx}{self.node_id}{key}", data_json) await self.mqtt.publish(f"{self.entitiy_prfx}{self.node_id}{key}", data_json)
self.new_data[key] = False self.new_data[key] = False
def __del__ (self): def __del__ (self):
logger.debug ("AsyncStream __del__") logger.debug ("AsyncStream __del__")
super().__del__()