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

Passed

Push — master ( fbdd84...45bad2 )

by P.R.

created 2016-08-03 14:18 UTC

MessageController._receive_message() A

↳ Parent: MessageController

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
cc	2
c	1
b	0
f	0
dl	0
loc	18
rs	9.4285

"""
Enarksh

Copyright 2013-2016 Set Based IT Consultancy

Licence MIT
"""
import zmq

from enarksh.event.Event import Event
from enarksh.event.EventActor import EventActor


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

    # ------------------------------------------------------------------------------------------------------------------
    def __init__(self):
        """
        Object constructor.
        """
        EventActor.__init__(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]
        """

    # ------------------------------------------------------------------------------------------------------------------
    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)
        socket.bind(end_point)
        self._end_points[name] = socket

    # ------------------------------------------------------------------------------------------------------------------
    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.
        """
        message = socket.recv_pyobj()
        """: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 read 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):
        """
        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.
        """
        socket = self._end_points[end_point]
        socket.send_pyobj(message)

    # ------------------------------------------------------------------------------------------------------------------
    def destroy(self):
        """
        Removes this object from the event system. This as preparation for removing this object such that there aren't
        references to this object and the garbage collector can remove this object.
        """
        EventActor.destroy()

        self._message_types = {}

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


1			"""
2			Enarksh
3
4			Copyright 2013-2016 Set Based IT Consultancy
5
6			Licence MIT
7			"""
8			import zmq
9
10			from enarksh.event.Event import Event
11			from enarksh.event.EventActor import EventActor
12
13
14			class MessageController(EventActor):
15			"""
16			A message controller for receiving messages and firing the appropriate events.
17			"""
18
19			# ------------------------------------------------------------------------------------------------------------------
20			def __init__(self):
21			"""
22			Object constructor.
23			"""
24			EventActor.__init__(self)
25
26			self._zmq_context = zmq.Context()
27			"""
28			The ZMQ context.
29
30			:type: None\|zmq.Context
31			"""
32
33			self._message_types = {}
34			"""
35			All registered message types. A dict from message type to the event that must fire when a message of
36			that message type has been received.
37
38			:type: dict[str,enarksh.event.Event.Event]
39			"""
40
41			self._end_points = {}
42			"""
43			All registered end points.
44
45			:type: dict[str,zmq.sugar.socket.Socket]
46			"""
47
48			# ------------------------------------------------------------------------------------------------------------------
49			def register_end_point(self, name, socket_type, end_point):
50			"""
51			Registers an end point.
52
53			:param str name: The name of the end point.
54			:param int socket_type: The socket type, one of
55			- zmq.PULL for asynchronous incoming messages
56			- zmq.REP for lockstep incoming messages
57			- zmq.PUSH for asynchronous outgoing messages
58			:param str end_point: The end point.
59			"""
60			socket = self._zmq_context.socket(socket_type)
61			socket.bind(end_point)
62			self._end_points[name] = socket
63
64			# ------------------------------------------------------------------------------------------------------------------
65			def register_message_type(self, message_type):
66			"""
67			Registers a message type together with the event and message constructor.
68
69			:param str message_type: The message type to register.
70			"""
71			self._message_types[message_type] = Event(self)
72
73			# ------------------------------------------------------------------------------------------------------------------
74			def get_event_by_message_type(self, message_type):
75			"""
76			Returns the event that will be fired when a message a certain message type is been received.
77
78			:param str message_type: The messaged type
79
80			:rtype: enarksh.event.Event.Event
81			"""
82			if message_type not in self._message_types:
83			raise ValueError("Unknown message type '{0}'".format(message_type))
84
85			return self._message_types[message_type]
86
87			# ------------------------------------------------------------------------------------------------------------------
88			def register_listener(self, message_type, listener, listener_data=None):
89			"""
90			Registers an object as a listener for the event fired when a message has been received.
91
92			:param str message_type: The message type.
93			:param callable listener: An object that listen for an event.
94			:param * listener_data: Additional data supplied by the listener destination.
95			"""
96			self.get_event_by_message_type(message_type).register_listener(listener, listener_data)
97
98			# ------------------------------------------------------------------------------------------------------------------
99			def _receive_message(self, name, socket):
100			"""
101			Receives an incoming message from a ZMQ socket.
102
103			:param str name: The name of the end point of source of the message.
104			:param zmq.sugar.socket.Socket socket: The ZMQ socket.
105
106			XXX @todo Support JSON messages for communication with PHP front end.
107			"""
108			message = socket.recv_pyobj()
109			""":type: enarksh.message.Message.Message"""
110			message.message_source = name
111
112			if message.message_type not in self._message_types:
113			raise ValueError("Received message with unknown message type '{0}'".format(message.message_type))
114
115			event = self._message_types[message.message_type]
116			event.fire(message)
117
118			# ------------------------------------------------------------------------------------------------------------------
119			def receive_message(self, event, event_data, listener_data):
120			"""
121			Receives a messages from another processes.
122
123			:param * event: Not used.
124			:param * event_data: Not used.
125			:param * listener_data: Not used.
126			"""
127			del event, event_data, listener_data
128
129			# Make a poller for all incoming sockets.
130			poller = zmq.Poller()
131			for socket in self._end_points.values():
132			if socket.type in [zmq.PULL, zmq.REP]:
133			poller.register(socket, zmq.POLLIN)
134
135			# Wait for socket is read for reading.
136			socks = dict(poller.poll())
137
138			for name, socket in self._end_points.items():
139			if socket in socks:
140			self._receive_message(name, socket)
141
142			# ------------------------------------------------------------------------------------------------------------------
143			def send_message(self, end_point, message):
144			"""
145			Sends a message to an end point.
146
147			:param str end_point: The name of the end point.
148			:param enarksh.message.Message.Message message: The message.
149			"""
150			socket = self._end_points[end_point]
151			socket.send_pyobj(message)
152
153			# ------------------------------------------------------------------------------------------------------------------
154			def destroy(self):
155			"""
156			Removes this object from the event system. This as preparation for removing this object such that there aren't
157			references to this object and the garbage collector can remove this object.
158			"""
159			EventActor.destroy()
160
161			self._message_types = {}
162
163			# ----------------------------------------------------------------------------------------------------------------------
164

SetBased / py-enarksh

GitHub Access Token became invalid

Push — master ( fbdd84...45bad2 )

MessageController._receive_message() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like