linse/frappy
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

frappy_psi.sea: auto connect

Browse Source 
on both .ssynio and /syncio try to reconnect after failure
This commit is contained in:
l_samenv 2023-06-26 14:45:53 +02:00 committed by Markus Zolliker
parent e36ff9bc41
commit 8e7e70c50e
1 changed files with 35 additions and 26 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
frappy_psi/sea.py
View File

@ -111,6 +111,7 @@ class SeaClient(ProxyClient, Module):
_connect_thread = None _connect_thread = None
_service_manager = None _service_manager = None
_instance = None _instance = None
_last_connect = 0
def __init__(self, name, log, opts, srv): def __init__(self, name, log, opts, srv):
nodename = srv.node_cfg.get('name') or srv.node_cfg.get('equipment_id') nodename = srv.node_cfg.get('name') or srv.node_cfg.get('equipment_id')
@ -135,6 +136,10 @@ class SeaClient(ProxyClient, Module):
ProxyClient.__init__(self) ProxyClient.__init__(self)
Module.__init__(self, name, log, opts, srv) Module.__init__(self, name, log, opts, srv)
def doPoll(self):
if not self.asynio and time.time() > self._last_connect + 10:
self._connect_thread = mkthread(self._connect, None)
def register_obj(self, module, obj): def register_obj(self, module, obj):
self.objects.add(obj) self.objects.add(obj)
for k, v in module.path2param.items(): for k, v in module.path2param.items():
@ -146,6 +151,7 @@ class SeaClient(ProxyClient, Module):
self._connect_thread = mkthread(self._connect, start_events.get_trigger()) self._connect_thread = mkthread(self._connect, start_events.get_trigger())
def _connect(self, started_callback): def _connect(self, started_callback):
self._last_connect = time.time()
if self._instance: if self._instance:
if not self._service_manager: if not self._service_manager:
if self._service_manager is None: if self._service_manager is None:
@ -192,18 +198,16 @@ class SeaClient(ProxyClient, Module):
self.syncio.writeline(b'seauser seaser') self.syncio.writeline(b'seauser seaser')
assert self.syncio.readline() == b'Login OK' assert self.syncio.readline() == b'Login OK'
self.log.info('connected to %s', self.uri) self.log.info('connected to %s', self.uri)
try:
self.syncio.flush_recv() self.syncio.flush_recv()
ft = 'fulltransAct' if quiet else 'fulltransact' ft = 'fulltransAct' if quiet else 'fulltransact'
self.syncio.writeline(('%s %s' % (ft, command)).encode()) self.syncio.writeline(('%s %s' % (ft, command)).encode())
result = None result = None
deadline = time.time() + 10 deadline = time.time() + 10
while time.time() < deadline: while time.time() < deadline:
try:
reply = self.syncio.readline() reply = self.syncio.readline()
if reply is None: if reply is None:
continue continue
except ConnectionClosed:
break
reply = reply.decode() reply = reply.decode()
if reply.startswith('TRANSACTIONSTART'): if reply.startswith('TRANSACTIONSTART'):
result = [] result = []
@ -222,6 +226,8 @@ class SeaClient(ProxyClient, Module):
result = [reply.split('=', 1)[-1]] result = [reply.split('=', 1)[-1]]
else: else:
result.append(reply) result.append(reply)
except ConnectionClosed:
self.syncio = None
raise TimeoutError('no response within 10s') raise TimeoutError('no response within 10s')
def _rxthread(self, started_callback): def _rxthread(self, started_callback):
@ -231,6 +237,10 @@ class SeaClient(ProxyClient, Module):
if reply is None: if reply is None:
continue continue
except ConnectionClosed: except ConnectionClosed:
try:
self.asynio.close()
except Exception:
self.asynio = None
break break
try: try:
msg = json.loads(reply) msg = json.loads(reply)
@ -463,7 +473,6 @@ class SeaModule(Module):
descr['params'].pop(0) descr['params'].pop(0)
else: else:
# filter by relative paths # filter by relative paths
# rel_paths = rel_paths.split()
result = [] result = []
is_running = None is_running = None
for rpath in rel_paths: for rpath in rel_paths: