change docstring comments to use sphinx style
[leap_pycommon.git] / src / leap / common / events / component.py
1 # -*- coding: utf-8 -*-
2 # component.py
3 # Copyright (C) 2013 LEAP
4 #
5 # This program is free software: you can redistribute it and/or modify
6 # it under the terms of the GNU General Public License as published by
7 # the Free Software Foundation, either version 3 of the License, or
8 # (at your option) any later version.
9 #
10 # This program is distributed in the hope that it will be useful,
11 # but WITHOUT ANY WARRANTY; without even the implied warranty of
12 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
13 # GNU General Public License for more details.
14 #
15 # You should have received a copy of the GNU General Public License
16 # along with this program. If not, see <http://www.gnu.org/licenses/>.
17
18 """
19 The component end point of the events mechanism.
20
21 Components are the communicating parties of the events mechanism. They
22 communicate by sending messages to a server, which in turn redistributes
23 messages to other components.
24
25 When a component registers a callback for a given signal, it also tells the
26 server that it wants to be notified whenever signals of that type are sent by
27 some other component.
28 """
29
30
31 import logging
32 import threading
33
34
35 from protobuf.socketrpc import RpcService
36 from leap.common.events import (
37     events_pb2 as proto,
38     server,
39     daemon,
40     mac_auth,
41 )
42
43
44 logger = logging.getLogger(__name__)
45
46
47 # the `registered_callbacks` dictionary below should have the following
48 # format:
49 #
50 #     { event_signal: [ (uid, callback), ... ], ... }
51 #
52 registered_callbacks = {}
53
54
55 class CallbackAlreadyRegistered(Exception):
56     """
57     Raised when trying to register an already registered callback.
58     """
59
60
61 def ensure_component_daemon():
62     """
63     Ensure the component daemon is running and listening for incoming
64     messages.
65
66     :return: the daemon instance
67     :rtype: EventsComponentDaemon
68     """
69     import time
70     daemon = EventsComponentDaemon.ensure(0)
71     logger.debug('ensure component daemon')
72
73     # Because we use a random port we want to wait until a port is assigned to
74     # local component daemon.
75
76     while not (EventsComponentDaemon.get_instance() and
77                EventsComponentDaemon.get_instance().get_port()):
78         time.sleep(0.1)
79     return daemon
80
81
82 def register(signal, callback, uid=None, replace=False, reqcbk=None,
83              timeout=1000):
84     """
85     Registers a callback to be called when a specific signal event is
86     received.
87
88     Will timeout after timeout ms if response has not been received. The
89     timeout arg is only used for asynch requests. If a reqcbk callback has
90     been supplied the timeout arg is not used. The response value will be
91     returned for a synch request but nothing will be returned for an asynch
92     request.
93
94     :param signal: the signal that causes the callback to be launched
95     :type signal: int (see the `events.proto` file)
96     :param callback: the callback to be called when the signal is received
97     :type callback: function
98         callback(leap.common.events.events_pb2.SignalRequest)
99     :param uid: a unique id for the callback
100     :type uid: int
101     :param replace: should an existent callback with same uid be replaced?
102     :type replace: bool
103     :param reqcbk: a callback to be called when a response from server is
104         received
105     :type reqcbk: function
106         callback(leap.common.events.events_pb2.EventResponse)
107     :param timeout: the timeout for synch calls
108     :type timeout: int
109
110     Might raise a CallbackAlreadyRegistered exception if there's already a
111     callback identified by the given uid and replace is False.
112
113     :return: the response from server for synch calls or nothing for asynch
114         calls
115     :rtype: leap.common.events.events_pb2.EventsResponse or None
116     """
117     ensure_component_daemon()  # so we can receive registered signals
118     # register callback locally
119     if signal not in registered_callbacks:
120         registered_callbacks[signal] = []
121     cbklist = registered_callbacks[signal]
122     if uid and filter(lambda (x, y): x == uid, cbklist):
123         if not replace:
124             raise CallbackAlreadyRegisteredException()
125         else:
126             registered_callbacks[signal] = filter(lambda(x, y): x != uid,
127                                                   cbklist)
128     registered_callbacks[signal].append((uid, callback))
129     # register callback on server
130     request = proto.RegisterRequest()
131     request.event = signal
132     request.port = EventsComponentDaemon.get_instance().get_port()
133     request.mac_method = mac_auth.MacMethod.MAC_NONE
134     request.mac = ""
135     service = RpcService(proto.EventsServerService_Stub,
136                          server.SERVER_PORT, 'localhost')
137     logger.info(
138         "Sending registration request to server on port %s: %s",
139         server.SERVER_PORT,
140         str(request))
141     return service.register(request, callback=reqcbk, timeout=timeout)
142
143
144 def signal(signal, content="", mac_method="", mac="", reqcbk=None,
145            timeout=1000):
146     """
147     Send `signal` event to events server.
148
149     Will timeout after timeout ms if response has not been received. The
150     timeout arg is only used for asynch requests.  If a reqcbk callback has
151     been supplied the timeout arg is not used. The response value will be
152     returned for a synch request but nothing will be returned for an asynch
153     request.
154
155     :param signal: the signal that causes the callback to be launched
156     :type signal: int (see the `events.proto` file)
157     :param content: the contents of the event signal
158     :type content: str
159     :param mac_method: the method used for auth mac
160     :type mac_method: str
161     :param mac: the content of the auth mac
162     :type mac: str
163     :param reqcbk: a callback to be called when a response from server is
164         received
165     :type reqcbk: function
166         callback(leap.common.events.events_pb2.EventResponse)
167     :param timeout: the timeout for synch calls
168     :type timeout: int
169
170     :return: the response from server for synch calls or nothing for asynch
171         calls
172     :rtype: leap.common.events.events_pb2.EventsResponse or None
173     """
174     request = proto.SignalRequest()
175     request.event = signal
176     request.content = content
177     request.mac_method = mac_method
178     request.mac = mac
179     service = RpcService(proto.EventsServerService_Stub, server.SERVER_PORT,
180                          'localhost')
181     logger.info("Sending signal to server: %s", str(request))
182     return service.signal(request, callback=reqcbk, timeout=timeout)
183
184
185 class EventsComponentService(proto.EventsComponentService):
186     """
187     Service for receiving signal events in components.
188     """
189
190     def __init__(self):
191         proto.EventsComponentService.__init__(self)
192
193     def signal(self, controller, request, done):
194         """
195         Receive a signal and run callbacks registered for that signal.
196
197         This method is called whenever a signal request is received from
198         server.
199
200         :param controller: used to mediate a single method call
201         :type controller: protobuf.socketrpc.controller.SocketRpcController
202         :param request: the request received from the component
203         :type request: leap.common.events.events_pb2.SignalRequest
204         :param done: callback to be called when done
205         :type done: protobuf.socketrpc.server.Callback
206         """
207         logger.info('Received signal from server: %s' % str(request))
208
209         # run registered callbacks
210         # TODO: verify authentication using mac in incoming message
211         if request.event in registered_callbacks:
212             for (_, cbk) in registered_callbacks[request.event]:
213                 # callbacks should be prepared to receive a
214                 # events_pb2.SignalRequest.
215                 cbk(request)
216
217         # send response back to server
218         response = proto.EventResponse()
219         response.status = proto.EventResponse.OK
220         done.run(response)
221
222
223 class EventsComponentDaemon(daemon.EventsSingletonDaemon):
224     """
225     A daemon that listens for incoming events from server.
226     """
227
228     @classmethod
229     def ensure(cls, port):
230         """
231         Make sure the daemon is running on the given port.
232
233         :param port: the port in which the daemon should listen
234         :type port: int
235
236         :return: a daemon instance
237         :rtype: EventsComponentDaemon
238         """
239         return cls.ensure_service(port, EventsComponentService())