MessageController._receive_message() - Code Metrics - Inspection of "Refactor controller." - SetBased/py-enarksh - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( d4663b...83e50d )

by P.R.

created 2016-09-08 20:02 UTC

MessageController._receive_message() B

↳ Parent: MessageController

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
cc	4
c	1
b	0
f	0
dl	0
loc	25
rs	8.5806

"""
Enarksh

Copyright 2013-2016 Set Based IT Consultancy

Licence MIT
"""
import pickle
from time import sleep

import zmq
from zmq.utils import jsonapi

from enarksh.event.Event import Event
from enarksh.event.EventActor import EventActor
from enarksh.message.Message import Message


class MessageController(EventActor):
    """
    A message controller for receiving messages and firing the appropriate events.
    """

    _json_message_creators = {}
    """
    All registered message creates from JSON data.

    :type: dict[str,callable]
    """

    # ------------------------------------------------------------------------------------------------------------------
    def __init__(self):
        """
        Object constructor.
        """
        EventActor.__init__(self)

        Message.message_controller = self

        self._zmq_context = zmq.Context()
        """
        The ZMQ context.

        :type: None|zmq.Context
        """

        self._message_types = {}
        """
        All registered message types. A dict from message type to the event that must fire when a message of
        that message type has been received.

        :type: dict[str,enarksh.event.Event.Event]
        """

        self._end_points = {}
        """
        All registered end points.

        :type: dict[str,zmq.sugar.socket.Socket]
        """

    # ------------------------------------------------------------------------------------------------------------------
    @property
    def end_points(self):
        """
        Returns all end points.

        :rtype: dict[str,zmq.sugar.socket.Socket]
        """
        return self._end_points

    # ------------------------------------------------------------------------------------------------------------------
    def register_end_point(self, name, socket_type, end_point):
        """
        Registers an end point.

        :param str name: The name of the end point.
        :param int socket_type: The socket type, one of
                                - zmq.PULL for asynchronous incoming messages
                                - zmq.REP for lockstep incoming messages
                                - zmq.PUSH for asynchronous outgoing messages
        :param str end_point: The end point.
        """
        socket = self._zmq_context.socket(socket_type)
        self._end_points[name] = socket
        if socket_type in [zmq.PULL, zmq.REP]:
            socket.bind(end_point)
        elif socket_type == zmq.PUSH:
            socket.connect(end_point)
        else:
            raise ValueError("Unknown socket type {0}".format(socket_type))

    # ------------------------------------------------------------------------------------------------------------------
    @staticmethod
    def register_json_message_creator(message_type, creator):
        """
        Registers a function that creates a message from JSOn encode data.

        :param str message_type: The message type to register.
        :param callable creator: The function for creating a message from JSON encode data.
        """
        MessageController._json_message_creators[message_type] = creator

    # ------------------------------------------------------------------------------------------------------------------
    def register_message_type(self, message_type):
        """
        Registers a message type together with the event and message constructor.

        :param str message_type: The message type to register.
        """
        self._message_types[message_type] = Event(self)

    # ------------------------------------------------------------------------------------------------------------------
    def get_event_by_message_type(self, message_type):
        """
        Returns the event that will be fired when a message a certain message type is been received.

        :param str message_type: The messaged type

        :rtype: enarksh.event.Event.Event
        """
        if message_type not in self._message_types:
            raise ValueError("Unknown message type '{0}'".format(message_type))

        return self._message_types[message_type]

    # ------------------------------------------------------------------------------------------------------------------
    def register_listener(self, message_type, listener, listener_data=None):
        """
        Registers an object as a listener for the event fired when a message has been received.

        :param str message_type: The message type.
        :param callable listener: An object that listen for an event.
        :param * listener_data: Additional data supplied by the listener destination.
        """
        self.get_event_by_message_type(message_type).register_listener(listener, listener_data)

    # ------------------------------------------------------------------------------------------------------------------
    def _receive_message(self, name, socket):
        """
        Receives an incoming message from a ZMQ socket.

        :param str name: The name of the end point of source of the message.
        :param zmq.sugar.socket.Socket socket: The ZMQ socket.

        XXX @todo Support JSON messages for communication with PHP front end.
        """
        buffer = socket.recv()
        if buffer[:1] == b'{':
            tmp = jsonapi.loads(buffer)
            if tmp['type'] not in self._json_message_creators:
                raise ValueError("Received JSON message with unknown message type '{0}'".format(tmp['type']))
            message = self._json_message_creators[tmp['type']](tmp)
        else:
            message = pickle.loads(buffer)
            """:type: enarksh.message.Message.Message"""
            message.message_source = name

        if message.message_type not in self._message_types:
            raise ValueError("Received message with unknown message type '{0}'".format(message.message_type))

        event = self._message_types[message.message_type]
        event.fire(message)

    # ------------------------------------------------------------------------------------------------------------------
    def receive_message(self, event, event_data, listener_data):
        """
        Receives a messages from another processes.

        :param * event: Not used.
        :param * event_data: Not used.
        :param * listener_data: Not used.
        """
        del event, event_data, listener_data

        # Make a poller for all incoming sockets.
        poller = zmq.Poller()
        for socket in self._end_points.values():
            if socket.type in [zmq.PULL, zmq.REP]:
                poller.register(socket, zmq.POLLIN)

        # Wait for socket is ready for reading.
        socks = dict(poller.poll())

        for name, socket in self._end_points.items():
            if socket in socks:
                self._receive_message(name, socket)

    # ------------------------------------------------------------------------------------------------------------------
    def send_message(self, end_point, message, json=False):
        """
        Sends a message to an end point.

        :param str end_point: The name of the end point.
        :param enarksh.message.Message.Message|* message: The message.
        :param bool json: If True the message will be send as a JSON encode string.
        """
        socket = self._end_points[end_point]
        if json:
            socket.send_json(message)
        else:
            socket.send_pyobj(message)

    # ------------------------------------------------------------------------------------------------------------------
    def no_barking(self, seconds):
        """
        During start up of ZMQ the incoming file descriptors become 'ready for reading' while there is no message on
        the socket. This method prevent incoming sockets barking that the are ready the for reading.

        :param int seconds: The number of seconds the give the other ZMQ thread to start up.
        """
        sleep(seconds)

        for _ in range(1, len(self.end_points)):
            poller = zmq.Poller()
            for socket in self.end_points.values():
                if socket.type in [zmq.PULL, zmq.REP]:
                    poller.register(socket, zmq.POLLIN)

            poller.poll(1)

# ----------------------------------------------------------------------------------------------------------------------


1			"""
2			Enarksh
3
4			Copyright 2013-2016 Set Based IT Consultancy
5
6			Licence MIT
7			"""
8			import pickle
9			from time import sleep
10
11			import zmq
12			from zmq.utils import jsonapi
13
14			from enarksh.event.Event import Event
15			from enarksh.event.EventActor import EventActor
16			from enarksh.message.Message import Message
17
18
19			class MessageController(EventActor):
20			"""
21			A message controller for receiving messages and firing the appropriate events.
22			"""
23
24			_json_message_creators = {}
25			"""
26			All registered message creates from JSON data.
27
28			:type: dict[str,callable]
29			"""
30
31			# ------------------------------------------------------------------------------------------------------------------
32			def __init__(self):
33			"""
34			Object constructor.
35			"""
36			EventActor.__init__(self)
37
38			Message.message_controller = self
39
40			self._zmq_context = zmq.Context()
41			"""
42			The ZMQ context.
43
44			:type: None\|zmq.Context
45			"""
46
47			self._message_types = {}
48			"""
49			All registered message types. A dict from message type to the event that must fire when a message of
50			that message type has been received.
51
52			:type: dict[str,enarksh.event.Event.Event]
53			"""
54
55			self._end_points = {}
56			"""
57			All registered end points.
58
59			:type: dict[str,zmq.sugar.socket.Socket]
60			"""
61
62			# ------------------------------------------------------------------------------------------------------------------
63			@property
64			def end_points(self):
65			"""
66			Returns all end points.
67
68			:rtype: dict[str,zmq.sugar.socket.Socket]
69			"""
70			return self._end_points
71
72			# ------------------------------------------------------------------------------------------------------------------
73			def register_end_point(self, name, socket_type, end_point):
74			"""
75			Registers an end point.
76
77			:param str name: The name of the end point.
78			:param int socket_type: The socket type, one of
79			- zmq.PULL for asynchronous incoming messages
80			- zmq.REP for lockstep incoming messages
81			- zmq.PUSH for asynchronous outgoing messages
82			:param str end_point: The end point.
83			"""
84			socket = self._zmq_context.socket(socket_type)
85			self._end_points[name] = socket
86			if socket_type in [zmq.PULL, zmq.REP]:
87			socket.bind(end_point)
88			elif socket_type == zmq.PUSH:
89			socket.connect(end_point)
90			else:
91			raise ValueError("Unknown socket type {0}".format(socket_type))
92
93			# ------------------------------------------------------------------------------------------------------------------
94			@staticmethod
95			def register_json_message_creator(message_type, creator):
96			"""
97			Registers a function that creates a message from JSOn encode data.
98
99			:param str message_type: The message type to register.
100			:param callable creator: The function for creating a message from JSON encode data.
101			"""
102			MessageController._json_message_creators[message_type] = creator
103
104			# ------------------------------------------------------------------------------------------------------------------
105			def register_message_type(self, message_type):
106			"""
107			Registers a message type together with the event and message constructor.
108
109			:param str message_type: The message type to register.
110			"""
111			self._message_types[message_type] = Event(self)
112
113			# ------------------------------------------------------------------------------------------------------------------
114			def get_event_by_message_type(self, message_type):
115			"""
116			Returns the event that will be fired when a message a certain message type is been received.
117
118			:param str message_type: The messaged type
119
120			:rtype: enarksh.event.Event.Event
121			"""
122			if message_type not in self._message_types:
123			raise ValueError("Unknown message type '{0}'".format(message_type))
124
125			return self._message_types[message_type]
126
127			# ------------------------------------------------------------------------------------------------------------------
128			def register_listener(self, message_type, listener, listener_data=None):
129			"""
130			Registers an object as a listener for the event fired when a message has been received.
131
132			:param str message_type: The message type.
133			:param callable listener: An object that listen for an event.
134			:param * listener_data: Additional data supplied by the listener destination.
135			"""
136			self.get_event_by_message_type(message_type).register_listener(listener, listener_data)
137
138			# ------------------------------------------------------------------------------------------------------------------
139			def _receive_message(self, name, socket):
140			"""
141			Receives an incoming message from a ZMQ socket.
142
143			:param str name: The name of the end point of source of the message.
144			:param zmq.sugar.socket.Socket socket: The ZMQ socket.
145
146			XXX @todo Support JSON messages for communication with PHP front end.
147			"""
148			buffer = socket.recv()
149			if buffer[:1] == b'{':
150			tmp = jsonapi.loads(buffer)
151			if tmp['type'] not in self._json_message_creators:
152			raise ValueError("Received JSON message with unknown message type '{0}'".format(tmp['type']))
153			message = self._json_message_creators[tmp['type']](tmp)
154			else:
155			message = pickle.loads(buffer)
156			""":type: enarksh.message.Message.Message"""
157			message.message_source = name
158
159			if message.message_type not in self._message_types:
160			raise ValueError("Received message with unknown message type '{0}'".format(message.message_type))
161
162			event = self._message_types[message.message_type]
163			event.fire(message)
164
165			# ------------------------------------------------------------------------------------------------------------------
166			def receive_message(self, event, event_data, listener_data):
167			"""
168			Receives a messages from another processes.
169
170			:param * event: Not used.
171			:param * event_data: Not used.
172			:param * listener_data: Not used.
173			"""
174			del event, event_data, listener_data
175
176			# Make a poller for all incoming sockets.
177			poller = zmq.Poller()
178			for socket in self._end_points.values():
179			if socket.type in [zmq.PULL, zmq.REP]:
180			poller.register(socket, zmq.POLLIN)
181
182			# Wait for socket is ready for reading.
183			socks = dict(poller.poll())
184
185			for name, socket in self._end_points.items():
186			if socket in socks:
187			self._receive_message(name, socket)
188
189			# ------------------------------------------------------------------------------------------------------------------
190			def send_message(self, end_point, message, json=False):
191			"""
192			Sends a message to an end point.
193
194			:param str end_point: The name of the end point.
195			:param enarksh.message.Message.Message\|* message: The message.
196			:param bool json: If True the message will be send as a JSON encode string.
197			"""
198			socket = self._end_points[end_point]
199			if json:
200			socket.send_json(message)
201			else:
202			socket.send_pyobj(message)
203
204			# ------------------------------------------------------------------------------------------------------------------
205			def no_barking(self, seconds):
206			"""
207			During start up of ZMQ the incoming file descriptors become 'ready for reading' while there is no message on
208			the socket. This method prevent incoming sockets barking that the are ready the for reading.
209
210			:param int seconds: The number of seconds the give the other ZMQ thread to start up.
211			"""
212			sleep(seconds)
213
214			for _ in range(1, len(self.end_points)):
215			poller = zmq.Poller()
216			for socket in self.end_points.values():
217			if socket.type in [zmq.PULL, zmq.REP]:
218			poller.register(socket, zmq.POLLIN)
219
220			poller.poll(1)
221
222			# ----------------------------------------------------------------------------------------------------------------------
223

SetBased / py-enarksh

GitHub Access Token became invalid

Push — master ( d4663b...83e50d )

MessageController._receive_message() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like