mirror of
https://github.com/avatao-content/baseimage-tutorial-framework
synced 2024-12-23 04:01:56 +00:00
Move serialization in one place
This commit is contained in:
parent
91d5e560eb
commit
31fea13a9a
@ -1,5 +1,6 @@
|
||||
import json
|
||||
|
||||
from .networking.serialization import deserialize_all
|
||||
from .networking.server_connector import ServerConnector
|
||||
|
||||
|
||||
@ -13,12 +14,11 @@ class EventHandlerBase:
|
||||
self.server_connector.register_callback(self.event_handler_callback)
|
||||
|
||||
def event_handler_callback(self, msg_parts):
|
||||
anchor, message = msg_parts
|
||||
anchor, message = deserialize_all(*msg_parts)
|
||||
data_json = json.loads(message)
|
||||
response = self.handle_event(anchor, data_json) if anchor != b'reset' else self.handle_reset(data_json)
|
||||
response = self.handle_event(anchor, data_json) if anchor != 'reset' else self.handle_reset(data_json)
|
||||
if response is None: return
|
||||
encoded_response = json.dumps(response).encode('utf-8')
|
||||
self.server_connector.send(anchor, encoded_response)
|
||||
self.server_connector.send(anchor, json.dumps(response))
|
||||
|
||||
def handle_event(self, anchor, data_json):
|
||||
raise NotImplementedError
|
||||
@ -27,13 +27,11 @@ class EventHandlerBase:
|
||||
return None
|
||||
|
||||
def message_other(self, anchor, data):
|
||||
encoded_anchor = anchor.encode('utf-8')
|
||||
message = {
|
||||
'anchor': anchor,
|
||||
'data': data
|
||||
}
|
||||
encoded_message = json.dumps(message).encode('utf-8')
|
||||
self.server_connector.send(encoded_anchor, encoded_message)
|
||||
self.server_connector.send(anchor, json.dumps(message))
|
||||
|
||||
def subscribe(self, anchor):
|
||||
if anchor not in self.subscriptions:
|
||||
|
@ -5,10 +5,8 @@ from .networking.server_connector import ServerUplinkConnector
|
||||
|
||||
|
||||
class MessageSender:
|
||||
def __init__(self, custom_anchor=None):
|
||||
def __init__(self, custom_anchor: str = None):
|
||||
self.server_connector = ServerUplinkConnector()
|
||||
if isinstance(custom_anchor, bytes):
|
||||
custom_anchor = custom_anchor.decode('utf-8')
|
||||
self.anchor = custom_anchor or 'message'
|
||||
|
||||
def send(self, originator, message):
|
||||
@ -21,4 +19,4 @@ class MessageSender:
|
||||
'anchor': self.anchor,
|
||||
'data': data
|
||||
}
|
||||
self.server_connector.send(*[frame.encode('utf-8') for frame in (self.anchor, json.dumps(response))])
|
||||
self.server_connector.send(self.anchor, json.dumps(response))
|
||||
|
@ -2,6 +2,7 @@ import zmq
|
||||
from zmq.eventloop import ioloop
|
||||
from zmq.eventloop.zmqstream import ZMQStream
|
||||
|
||||
from .serialization import serialize_all
|
||||
from ..config import PUBLISHER_PORT, RECEIVER_PORT
|
||||
from ..config.logs import logging
|
||||
log = logging.getLogger(__name__)
|
||||
@ -40,5 +41,4 @@ class EventHandlerConnector(EventHandlerDownlinkConnector, EventHandlerUplinkCon
|
||||
def send_message(self, message: str, anchor: str = None):
|
||||
if not anchor:
|
||||
anchor = parse_anchor_from_message(message)
|
||||
encoded_message = [part.encode('utf-8') for part in (anchor, message)]
|
||||
self._zmq_pub_socket.send_multipart(encoded_message)
|
||||
self._zmq_pub_socket.send_multipart(serialize_all(anchor, message))
|
||||
|
18
lib/tfw/networking/serialization.py
Normal file
18
lib/tfw/networking/serialization.py
Normal file
@ -0,0 +1,18 @@
|
||||
def encode_if_needed(value):
|
||||
if isinstance(value, str):
|
||||
value = value.encode('utf-8')
|
||||
return value
|
||||
|
||||
|
||||
def decode_if_needed(value):
|
||||
if isinstance(value, (bytes, bytearray)):
|
||||
value = value.decode('utf-8')
|
||||
return value
|
||||
|
||||
|
||||
def serialize_all(*args):
|
||||
return [encode_if_needed(a) for a in args]
|
||||
|
||||
|
||||
def deserialize_all(*args):
|
||||
return [decode_if_needed(a) for a in args]
|
@ -3,6 +3,7 @@ from functools import partial
|
||||
from zmq.eventloop import ioloop
|
||||
from zmq.eventloop.zmqstream import ZMQStream
|
||||
|
||||
from .serialization import serialize_all
|
||||
from ..config import PUBLISHER_PORT, RECEIVER_PORT
|
||||
from ..util import ZMQConnectorBase
|
||||
|
||||
@ -29,7 +30,7 @@ class ServerUplinkConnector(ZMQConnectorBase):
|
||||
self._zmq_push_socket.connect('tcp://localhost:{}'.format(RECEIVER_PORT))
|
||||
|
||||
def send(self, anchor, response):
|
||||
self._zmq_push_socket.send_multipart([anchor, response])
|
||||
self._zmq_push_socket.send_multipart(serialize_all(anchor, response))
|
||||
|
||||
|
||||
class ServerConnector(ServerUplinkConnector, ServerDownlinkConnector):
|
||||
|
@ -1,6 +1,7 @@
|
||||
import json
|
||||
from tornado.websocket import WebSocketHandler
|
||||
|
||||
from .serialization import deserialize_all
|
||||
from ..util import parse_anchor_from_message
|
||||
from .event_handler_connector import EventHandlerConnector
|
||||
from ..config.logs import logging
|
||||
@ -20,9 +21,9 @@ class ZMQWebSocketHandler(WebSocketHandler):
|
||||
self._event_handler_connector.register_callback(self.zmq_callback)
|
||||
|
||||
def zmq_callback(self, msg_parts):
|
||||
anchor, data = msg_parts
|
||||
log.debug('Received on pull socket: {}'.format(data.decode()))
|
||||
self.write_message(data.decode())
|
||||
anchor, data = deserialize_all(*msg_parts)
|
||||
log.debug('Received on pull socket: {}'.format(data))
|
||||
self.write_message(data)
|
||||
|
||||
def on_message(self, message):
|
||||
log.debug('Received on WebSocket: {}'.format(message))
|
||||
|
Loading…
Reference in New Issue
Block a user