Scott.Idem/OSIT-AE-API-FastAPI
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Work on websockets. Finally 98.25% working?

Browse Source
This commit is contained in:
Scott Idem
2023-04-07 18:09:02 -04:00
parent 6172626254
commit 9f8157d338
2 changed files with 31 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
admin/requirements.txt
View File

@@ -4,8 +4,9 @@ fastapi[all]
pydantic pydantic
SQLAlchemy SQLAlchemy
mysqlclient mysqlclient
redis # redis
aioredis redis[hiredis]
# aioredis
html2text html2text
pytz pytz
stripe stripe

43
app/routers/websockets_redis.py
View File

@@ -68,21 +68,15 @@ async def redis_connector(
data['client_id'] = client_id data['client_id'] = client_id
if group_id: if group_id:
data['group_id'] = group_id data['group_id'] = group_id
# log.setLevel(logging.DEBUG)
log.debug(data) log.debug(data)
if data: if data:
# logging.info(ws)
# logging.info(dir(message))
# log.debug(await r_conn.get('channel:ws'))
# log.debug(await r_conn.get_message())
await r_conn.publish('channel:ws', json.dumps(data)) await r_conn.publish('channel:ws', json.dumps(data))
except WebSocketDisconnect as exc: except WebSocketDisconnect as exc:
# TODO: This needs handling better? # TODO: This needs handling better?
# FUTURE: Remove from list of connected clients? # FUTURE: Remove from list of connected clients?
log.error(exc) # usually 1001? log.error(exc) # usually 1001?
# await r_conn.close() # Uncommenting causes an exception below (sender_handler) # await r_conn.close() # Uncommenting causes an exception below in sender_handler()
except Exception as e: except Exception as e:
log.exception('Exception in Aether!') log.exception('Exception in Aether!')
log.error(e) log.error(e)
@@ -101,7 +95,6 @@ async def redis_connector(
try: try:
while True: while True:
message = await r_channel.get_message(ignore_subscribe_messages=True) message = await r_channel.get_message(ignore_subscribe_messages=True)
if message is not None: if message is not None:
log.debug(f'(Redis Reader) Message Received: {message}') log.debug(f'(Redis Reader) Message Received: {message}')
@@ -113,22 +106,42 @@ async def redis_connector(
# log.debug(ws_conn.client_state) # log.debug(ws_conn.client_state)
log.debug(ws_conn.client_state.name) log.debug(ws_conn.client_state.name)
# We only want to send to the WS marked as connected.
# 'WebSocketState.CONNECTED' or 'WebSocketState.DISCONNECTED' # 'WebSocketState.CONNECTED' or 'WebSocketState.DISCONNECTED'
# if ws_conn.client_state.CONNECTED:
# if ws_conn.client_state.value == 'WebSocketState.CONNECTED':
# if ws_conn.client_state == WebSocketState.CONNECTED:
if ws_conn.client_state.name == 'CONNECTED': if ws_conn.client_state.name == 'CONNECTED':
log.info(f'WS client ({client_id}) connected!') log.info(f'WS client ({client_id}) connected!')
if group_id and group_id == data.get('group_id'): send_flag = False
# Figure out which WS client(s) get the data.
if data.get('target') == 'group' and group_id and group_id == data.get('group_id'):
log.info('Sending to matching group ID') log.info('Sending to matching group ID')
await ws_conn.send_json(data) # await ws_conn.send_json(data)
send_flag = True
elif group_id: elif group_id:
log.info('Skip sending to client') log.info('Skip sending to client')
pass send_flag = False
else: # continue # Do not send
if data.get('target') == 'echo' and client_id == data.get('client_id'):
log.info('Echo')
send_flag = True
elif data.get('target') == 'echo':
log.info('Skip sending to client')
send_flag = False
# continue
if data.get('target') == 'all':
send_flag = True
log.info('Sending to all') log.info('Sending to all')
# await ws_conn.send_json(data)
if send_flag:
log.info('Sending!')
await ws_conn.send_json(data) await ws_conn.send_json(data)
else:
log.info('Not sending')
continue
else: else:
log.warning(f'WS client ({client_id}) no longer connected!? They have just missed a message.') log.warning(f'WS client ({client_id}) no longer connected!? They have just missed a message.')