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

frappy_psi.sea: try to reconnect on failure

Browse Source 
both .asynio and .syncio connection should be tried to reopen.

Change-Id: I836ede1e5fff6f7ac670c92e769da8ec834d2b12
Reviewed-on: https://forge.frm2.tum.de/review/c/secop/frappy/+/31449
Tested-by: Jenkins Automated Tests <pedersen+jenkins@frm2.tum.de>
Reviewed-by: Markus Zolliker <markus.zolliker@psi.ch>
This commit is contained in:
zolliker
2023-06-26 16:18:47 +02:00
parent b47e5d270c
commit 9ece4f797b
1 changed files with 40 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

65
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,36 +198,40 @@ 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)
self.syncio.flush_recv() try:
ft = 'fulltransAct' if quiet else 'fulltransact' self.syncio.flush_recv()
self.syncio.writeline(('%s %s' % (ft, command)).encode()) ft = 'fulltransAct' if quiet else 'fulltransact'
result = None self.syncio.writeline(('%s %s' % (ft, command)).encode())
deadline = time.time() + 10 result = None
while time.time() < deadline: deadline = time.time() + 10
try: while time.time() < deadline:
reply = self.syncio.readline() reply = self.syncio.readline()
if reply is None: if reply is None:
continue continue
except ConnectionClosed: reply = reply.decode()
break if reply.startswith('TRANSACTIONSTART'):
reply = reply.decode() result = []
if reply.startswith('TRANSACTIONSTART'): continue
result = [] if reply == 'TRANSACTIONFINISHED':
continue if result is None:
if reply == 'TRANSACTIONFINISHED': self.log.info('missing TRANSACTIONSTART on: %s', command)
return ''
if not result:
return ''
return '\n'.join(result)
if result is None: if result is None:
self.log.info('missing TRANSACTIONSTART on: %s', command) self.log.info('swallow: %s', reply)
return '' continue
if not result: if not result:
return '' result = [reply.split('=', 1)[-1]]
return '\n'.join(result) else:
if result is None: result.append(reply)
self.log.info('swallow: %s', reply) except ConnectionClosed:
continue try:
if not result: self.syncio.disconnect()
result = [reply.split('=', 1)[-1]] except Exception:
else: pass
result.append(reply) 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 +241,11 @@ class SeaClient(ProxyClient, Module):
if reply is None: if reply is None:
continue continue
except ConnectionClosed: except ConnectionClosed:
try:
self.asynio.disconnect()
except Exception:
pass
self.asynio = None
break break
try: try:
msg = json.loads(reply) msg = json.loads(reply)