diff options
| author | Tomás Touceda <chiiph@leap.se> | 2013-07-25 10:32:06 -0300 | 
|---|---|---|
| committer | Tomás Touceda <chiiph@leap.se> | 2013-07-25 10:32:06 -0300 | 
| commit | c22072f6a12a9d896eb324a848043011b0ea30c5 (patch) | |
| tree | 62626661cf063af8bf4143e981c489f3e73a2b2f /src | |
| parent | f659de6e98073dc738b8d245e0ddb897f5bd3779 (diff) | |
| parent | e7fa419f13e0afbb3f2653a4b0d8330dd45bbfd0 (diff) | |
Merge remote-tracking branch 'drebs/feature/3246-refactor-events-so-components-become-clients' into develop
Diffstat (limited to 'src')
| -rw-r--r-- | src/leap/common/events/README.rst | 12 | ||||
| -rw-r--r-- | src/leap/common/events/__init__.py | 109 | ||||
| -rw-r--r-- | src/leap/common/events/client.py (renamed from src/leap/common/events/component.py) | 114 | ||||
| -rw-r--r-- | src/leap/common/events/daemon.py | 2 | ||||
| -rw-r--r-- | src/leap/common/events/events.proto | 36 | ||||
| -rw-r--r-- | src/leap/common/events/events_pb2.py | 93 | ||||
| -rw-r--r-- | src/leap/common/events/server.py | 117 | ||||
| -rw-r--r-- | src/leap/common/tests/test_events.py | 301 | 
8 files changed, 617 insertions, 167 deletions
| diff --git a/src/leap/common/events/README.rst b/src/leap/common/events/README.rst index 813be8b..2e7f254 100644 --- a/src/leap/common/events/README.rst +++ b/src/leap/common/events/README.rst @@ -1,19 +1,19 @@  Events mechanism  ================ -The events mechanism allows for "components" to send signal events to each -other by means of a centralized server. Components can register with the +The events mechanism allows for clients to send signal events to each +other by means of a centralized server. Clients can register with the  server to receive signals of certain types, and they can also send signals to -the server that will then redistribute these signals to registered components. +the server that will then redistribute these signals to registered clients.  Listening daemons  ----------------- -Both components and the server listen for incoming messages by using a +Both clients and the server listen for incoming messages by using a  listening daemon that runs in its own thread. The server daemon has to be -started explicitly, while components daemon will be started whenever a -component registers with the server to receive messages. +started explicitly, while clients daemon will be started whenever a +client registers with the server to receive messages.  How to use it diff --git a/src/leap/common/events/__init__.py b/src/leap/common/events/__init__.py index 12416e4..a6fe7c3 100644 --- a/src/leap/common/events/__init__.py +++ b/src/leap/common/events/__init__.py @@ -16,7 +16,58 @@  # along with this program. If not, see <http://www.gnu.org/licenses/>.  """ -An events mechanism that allows for signaling of events between components. +This is an events mechanism that uses a server to allow for signaling of +events between clients. + +Application components should use the interface available in this file to +register callbacks to be executed upon receiving specific signals, and to send +signals to other components. + +To register a callback to be executed when a specific event is signaled, use +leap.common.events.register(): + +>>> from leap.common.events import register +>>> from leap.common.events.proto import CLIENT_UID +>>> register(CLIENT_UID, lambda req: do_something(req)) + +To signal an event, use leap.common.events.signal(): + +>>> from leap.common.events import signal +>>> from leap.common.events.proto import CLIENT_UID +>>> signal(CLIENT_UID) + + +NOTE ABOUT SYNC/ASYNC REQUESTS: + +Clients always communicate with the server, and never between themselves. +When a client registers a callback for an event, the callback is saved locally +in the client and the server stores the client socket port in a list +associated with that event. When a client signals an event, the server +forwards the signal to all registered client ports, and then each client +executes its callbacks associated with that event locally. + +Each RPC call from a client to the server is followed by a response from the +server to the client. Calls to register() and signal() (and all other RPC +calls) can be synchronous or asynchronous meaning if they will or not wait for +the server's response before returning. + +This mechanism was built on top of protobuf.socketrpc, and because of this RPC +calls are made synchronous or asynchronous in the following way: + +  * If RPC calls receive a parameter called `reqcbk`, then the call is made +    asynchronous. That means that: + +        - an eventual `timeout` parameter is not used, +        - the call returns immediatelly with value None, and +        - the `reqcbk` callback is executed asynchronously upon the arrival of +          a response from the server. + +  * Otherwise, if the `reqcbk` parameter is None, then the call is made in a +    synchronous manner: + +        - if a response from server arrives within `timeout` milliseconds, the +          RPC call returns it; +        - otherwise, the call returns None.  """  import logging @@ -26,7 +77,7 @@ import socket  from leap.common.events import (      events_pb2 as proto,      server, -    component, +    client,      daemon,  ) @@ -54,9 +105,8 @@ def register(signal, callback, uid=None, replace=False, reqcbk=None,      :param replace: should an existent callback with same uid be replaced?      :type replace: bool      :param reqcbk: a callback to be called when a response from server is -        received -    :type reqcbk: function -        callback(leap.common.events.events_pb2.EventResponse) +                   received +    :type reqcbk: function(leap.common.events.events_pb2.EventResponse)      :param timeout: the timeout for synch calls      :type timeout: int @@ -64,7 +114,7 @@ def register(signal, callback, uid=None, replace=False, reqcbk=None,          calls.      :rtype: leap.common.events.events_pb2.EventsResponse or None      """ -    return component.register(signal, callback, uid, replace, reqcbk, timeout) +    return client.register(signal, callback, uid, replace, reqcbk, timeout)  def unregister(signal, uid=None, reqcbk=None, timeout=1000): @@ -79,17 +129,16 @@ def unregister(signal, uid=None, reqcbk=None, timeout=1000):      :param uid: a unique id for the callback      :type uid: int      :param reqcbk: a callback to be called when a response from server is -        received -    :type reqcbk: function -        callback(leap.common.events.events_pb2.EventResponse) +                   received +    :type reqcbk: function(leap.common.events.events_pb2.EventResponse)      :param timeout: the timeout for synch calls      :type timeout: int      :return: the response from server for synch calls or nothing for asynch -        calls. +             calls.      :rtype: leap.common.events.events_pb2.EventsResponse or None      """ -    return component.unregister(signal, uid, reqcbk, timeout) +    return client.unregister(signal, uid, reqcbk, timeout)  def signal(signal, content="", mac_method="", mac="", reqcbk=None, @@ -112,14 +161,42 @@ def signal(signal, content="", mac_method="", mac="", reqcbk=None,      :param mac: the content of the auth mac      :type mac: str      :param reqcbk: a callback to be called when a response from server is -        received -    :type reqcbk: function -        callback(leap.common.events.events_pb2.EventResponse) +                   received +    :type reqcbk: function(leap.common.events.events_pb2.EventResponse)      :param timeout: the timeout for synch calls      :type timeout: int      :return: the response from server for synch calls or nothing for asynch -        calls. +             calls.      :rtype: leap.common.events.events_pb2.EventsResponse or None      """ -    return component.signal(signal, content, mac_method, mac, reqcbk, timeout) +    return client.signal(signal, content, mac_method, mac, reqcbk, timeout) + +def ping_client(port, reqcbk=None, timeout=1000): +    """ +    Ping a client running in C{port}. + +    :param port: the port in which the client should be listening +    :type port: int +    :param reqcbk: a callback to be called when a response from client is +                   received +    :type reqcbk: function(leap.common.events.events_pb2.EventResponse) +    :param timeout: the timeout for synch calls +    :type timeout: int +    """ +    return client.ping(port, reqcbk=reqcbk, timeout=timeout) + + +def ping_server(port=server.SERVER_PORT, reqcbk=None, timeout=1000): +    """ +    Ping the server. + +    :param port: the port in which server should be listening +    :type port: int +    :param reqcbk: a callback to be called when a response from server is +                   received +    :type reqcbk: function(leap.common.events.events_pb2.EventResponse) +    :param timeout: the timeout for synch calls +    :type timeout: int +    """ +    return server.ping(port, reqcbk=reqcbk, timeout=timeout) diff --git a/src/leap/common/events/component.py b/src/leap/common/events/client.py index 029d1ac..55f14ab 100644 --- a/src/leap/common/events/component.py +++ b/src/leap/common/events/client.py @@ -1,5 +1,5 @@  # -*- coding: utf-8 -*- -# component.py +# client.py  # Copyright (C) 2013 LEAP  #  # This program is free software: you can redistribute it and/or modify @@ -16,15 +16,15 @@  # along with this program. If not, see <http://www.gnu.org/licenses/>.  """ -The component end point of the events mechanism. +The client end point of the events mechanism. -Components are the communicating parties of the events mechanism. They +Clients are the communicating parties of the events mechanism. They  communicate by sending messages to a server, which in turn redistributes -messages to other components. +messages to other clients. -When a component registers a callback for a given signal, it also tells the +When a client registers a callback for a given signal, it also tells the  server that it wants to be notified whenever signals of that type are sent by -some other component. +some other client.  """ @@ -59,23 +59,23 @@ class CallbackAlreadyRegistered(Exception):      pass -def ensure_component_daemon(): +def ensure_client_daemon():      """ -    Ensure the component daemon is running and listening for incoming +    Ensure the client daemon is running and listening for incoming      messages.      :return: the daemon instance -    :rtype: EventsComponentDaemon +    :rtype: EventsClientDaemon      """      import time -    daemon = EventsComponentDaemon.ensure(0) -    logger.debug('ensure component daemon') +    daemon = EventsClientDaemon.ensure(0) +    logger.debug('ensure client daemon')      # Because we use a random port we want to wait until a port is assigned to -    # local component daemon. +    # local client daemon. -    while not (EventsComponentDaemon.get_instance() and -               EventsComponentDaemon.get_instance().get_port()): +    while not (EventsClientDaemon.get_instance() and +               EventsClientDaemon.get_instance().get_port()):          time.sleep(0.1)      return daemon @@ -95,16 +95,14 @@ def register(signal, callback, uid=None, replace=False, reqcbk=None,      :param signal: the signal that causes the callback to be launched      :type signal: int (see the `events.proto` file)      :param callback: the callback to be called when the signal is received -    :type callback: function -        callback(leap.common.events.events_pb2.SignalRequest) +    :type callback: function(leap.common.events.events_pb2.SignalRequest)      :param uid: a unique id for the callback      :type uid: int      :param replace: should an existent callback with same uid be replaced?      :type replace: bool      :param reqcbk: a callback to be called when a response from server is -        received -    :type reqcbk: function -        callback(leap.common.events.events_pb2.EventResponse) +                   received +    :type reqcbk: function(leap.common.events.events_pb2.EventResponse)      :param timeout: the timeout for synch calls      :type timeout: int @@ -112,10 +110,10 @@ def register(signal, callback, uid=None, replace=False, reqcbk=None,      callback identified by the given uid and replace is False.      :return: the response from server for synch calls or nothing for asynch -        calls. +             calls.      :rtype: leap.common.events.events_pb2.EventsResponse or None      """ -    ensure_component_daemon()  # so we can receive registered signals +    ensure_client_daemon()  # so we can receive registered signals      # register callback locally      if signal not in registered_callbacks:          registered_callbacks[signal] = [] @@ -130,7 +128,7 @@ def register(signal, callback, uid=None, replace=False, reqcbk=None,      # register callback on server      request = proto.RegisterRequest()      request.event = signal -    request.port = EventsComponentDaemon.get_instance().get_port() +    request.port = EventsClientDaemon.get_instance().get_port()      request.mac_method = mac_auth.MacMethod.MAC_NONE      request.mac = ""      service = RpcService(proto.EventsServerService_Stub, @@ -153,14 +151,14 @@ def unregister(signal, uid=None, reqcbk=None, timeout=1000):      :param uid: a unique id for the callback      :type uid: int      :param reqcbk: a callback to be called when a response from server is -        received -    :type reqcbk: function -        callback(leap.common.events.events_pb2.EventResponse) +                   received +    :type reqcbk: function(leap.common.events.events_pb2.EventResponse)      :param timeout: the timeout for synch calls      :type timeout: int      :return: the response from server for synch calls or nothing for asynch -        calls or None if no callback is registered for that signal or uid. +             calls or None if no callback is registered for that signal or +             uid.      :rtype: leap.common.events.events_pb2.EventsResponse or None      """      if signal not in registered_callbacks or not registered_callbacks[signal]: @@ -180,7 +178,7 @@ def unregister(signal, uid=None, reqcbk=None, timeout=1000):      if not registered_callbacks[signal]:          request = proto.UnregisterRequest()          request.event = signal -        request.port = EventsComponentDaemon.get_instance().get_port() +        request.port = EventsClientDaemon.get_instance().get_port()          request.mac_method = mac_auth.MacMethod.MAC_NONE          request.mac = ""          service = RpcService(proto.EventsServerService_Stub, @@ -212,14 +210,13 @@ def signal(signal, content="", mac_method="", mac="", reqcbk=None,      :param mac: the content of the auth mac      :type mac: str      :param reqcbk: a callback to be called when a response from server is -        received -    :type reqcbk: function -        callback(leap.common.events.events_pb2.EventResponse) +                   received +    :type reqcbk: function(leap.common.events.events_pb2.EventResponse)      :param timeout: the timeout for synch calls      :type timeout: int      :return: the response from server for synch calls or nothing for asynch -        calls. +             calls.      :rtype: leap.common.events.events_pb2.EventsResponse or None      """      request = proto.SignalRequest() @@ -233,13 +230,38 @@ def signal(signal, content="", mac_method="", mac="", reqcbk=None,      return service.signal(request, callback=reqcbk, timeout=timeout) -class EventsComponentService(proto.EventsComponentService): +def ping(port, reqcbk=None, timeout=1000):      """ -    Service for receiving signal events in components. +    Ping a client running in C{port}. + +    :param port: the port in which the client should be listening +    :type port: int +    :param reqcbk: a callback to be called when a response from client is +                   received +    :type reqcbk: function(leap.common.events.events_pb2.EventResponse) +    :param timeout: the timeout for synch calls +    :type timeout: int + +    :return: the response from client for synch calls or nothing for asynch +             calls. +    :rtype: leap.common.events.events_pb2.EventsResponse or None +    """ +    request = proto.PingRequest() +    service = RpcService( +        proto.EventsClientService_Stub, +        port, +        'localhost') +    logger.info("Pinging a client in port %d..." % port) +    return service.ping(request, callback=reqcbk, timeout=timeout) + + +class EventsClientService(proto.EventsClientService): +    """ +    Service for receiving signal events in clients.      """      def __init__(self): -        proto.EventsComponentService.__init__(self) +        proto.EventsClientService.__init__(self)      def signal(self, controller, request, done):          """ @@ -250,7 +272,7 @@ class EventsComponentService(proto.EventsComponentService):          :param controller: used to mediate a single method call          :type controller: protobuf.socketrpc.controller.SocketRpcController -        :param request: the request received from the component +        :param request: the request received from the client          :type request: leap.common.events.events_pb2.SignalRequest          :param done: callback to be called when done          :type done: protobuf.socketrpc.server.Callback @@ -270,8 +292,24 @@ class EventsComponentService(proto.EventsComponentService):          response.status = proto.EventResponse.OK          done.run(response) +    def ping(self, controller, request, done): +        """ +        Reply to a ping request. + +        :param controller: used to mediate a single method call +        :type controller: protobuf.socketrpc.controller.SocketRpcController +        :param request: the request received from the client +        :type request: leap.common.events.events_pb2.RegisterRequest +        :param done: callback to be called when done +        :type done: protobuf.socketrpc.server.Callback +        """ +        logger.info("Received ping request, sending response.") +        response = proto.EventResponse() +        response.status = proto.EventResponse.OK +        done.run(response) + -class EventsComponentDaemon(daemon.EventsSingletonDaemon): +class EventsClientDaemon(daemon.EventsSingletonDaemon):      """      A daemon that listens for incoming events from server.      """ @@ -285,6 +323,6 @@ class EventsComponentDaemon(daemon.EventsSingletonDaemon):          :type port: int          :return: a daemon instance -        :rtype: EventsComponentDaemon +        :rtype: EventsClientDaemon          """ -        return cls.ensure_service(port, EventsComponentService()) +        return cls.ensure_service(port, EventsClientService()) diff --git a/src/leap/common/events/daemon.py b/src/leap/common/events/daemon.py index c253948..c4a4189 100644 --- a/src/leap/common/events/daemon.py +++ b/src/leap/common/events/daemon.py @@ -43,7 +43,7 @@ class ServiceAlreadyRunningException(Exception):  class EventsRpcServer(SocketRpcServer):      """ -    RPC server used in server and component interfaces to receive messages. +    RPC server used in server and client interfaces to receive messages.      """      def __init__(self, port, host='localhost'): diff --git a/src/leap/common/events/events.proto b/src/leap/common/events/events.proto index a813ed1..b844f42 100644 --- a/src/leap/common/events/events.proto +++ b/src/leap/common/events/events.proto @@ -17,6 +17,9 @@  package leap.common.events;  option py_generic_services = true; + +// These are the events that can be signaled using the events mechanism. +  enum Event {    CLIENT_SESSION_ID = 1;    CLIENT_UID = 2; @@ -33,6 +36,10 @@ enum Event {    RAISE_WINDOW = 13;  } + +// A SignalRequest is the type of the message sent from one component to request +// that a signal be sent to every registered component. +  message SignalRequest {    required Event event = 1;    required string content = 2; @@ -42,6 +49,10 @@ message SignalRequest {    optional bool error_occurred = 6;  } + +// A RegisterRequest message tells the server that a component wants to +// be signaled whenever a specific event occurs. +  message RegisterRequest {    required Event event = 1;    required int32 port = 2; @@ -49,6 +60,10 @@ message RegisterRequest {    required bytes mac = 4;  } + +// An UnregisterRequest message tells the server that a component does not +// want to be signaled when a specific event occurs. +  message UnregisterRequest {    required Event event = 1;    required int32 port = 2; @@ -56,6 +71,17 @@ message UnregisterRequest {    required bytes mac = 4;  } + +// A PingRequest message is used to find out if a server or component is +// alive. + +message PingRequest { +} + + +// The EventResponse is the message sent back by server and components after +// they receive other kinds of requests. +  message EventResponse {    enum Status { @@ -68,12 +94,20 @@ message EventResponse {    optional string result = 2;  } + +// The EventsServerService is the service provided by the server. +  service EventsServerService { +  rpc ping(PingRequest) returns (EventResponse);    rpc register(RegisterRequest) returns (EventResponse);    rpc unregister(UnregisterRequest) returns (EventResponse);    rpc signal(SignalRequest) returns (EventResponse);  } -service EventsComponentService { + +// EventsComponentService is the service provided by components (clients). + +service EventsClientService { +  rpc ping(PingRequest) returns (EventResponse);    rpc signal(SignalRequest) returns (EventResponse);  } diff --git a/src/leap/common/events/events_pb2.py b/src/leap/common/events/events_pb2.py index 5b1c118..274514c 100644 --- a/src/leap/common/events/events_pb2.py +++ b/src/leap/common/events/events_pb2.py @@ -14,7 +14,7 @@ from google.protobuf import descriptor_pb2  DESCRIPTOR = _descriptor.FileDescriptor(      name='events.proto',      package='leap.common.events', -    serialized_pb='\n\x0c\x65vents.proto\x12\x12leap.common.events\"\x97\x01\n\rSignalRequest\x12(\n\x05\x65vent\x18\x01 \x02(\x0e\x32\x19.leap.common.events.Event\x12\x0f\n\x07\x63ontent\x18\x02 \x02(\t\x12\x12\n\nmac_method\x18\x03 \x02(\t\x12\x0b\n\x03mac\x18\x04 \x02(\x0c\x12\x12\n\nenc_method\x18\x05 \x01(\t\x12\x16\n\x0e\x65rror_occurred\x18\x06 \x01(\x08\"j\n\x0fRegisterRequest\x12(\n\x05\x65vent\x18\x01 \x02(\x0e\x32\x19.leap.common.events.Event\x12\x0c\n\x04port\x18\x02 \x02(\x05\x12\x12\n\nmac_method\x18\x03 \x02(\t\x12\x0b\n\x03mac\x18\x04 \x02(\x0c\"l\n\x11UnregisterRequest\x12(\n\x05\x65vent\x18\x01 \x02(\x0e\x32\x19.leap.common.events.Event\x12\x0c\n\x04port\x18\x02 \x02(\x05\x12\x12\n\nmac_method\x18\x03 \x02(\t\x12\x0b\n\x03mac\x18\x04 \x02(\x0c\"\x82\x01\n\rEventResponse\x12\x38\n\x06status\x18\x01 \x02(\x0e\x32(.leap.common.events.EventResponse.Status\x12\x0e\n\x06result\x18\x02 \x01(\t\"\'\n\x06Status\x12\x06\n\x02OK\x10\x01\x12\n\n\x06UNAUTH\x10\x02\x12\t\n\x05\x45RROR\x10\x03*\xe7\x02\n\x05\x45vent\x12\x15\n\x11\x43LIENT_SESSION_ID\x10\x01\x12\x0e\n\nCLIENT_UID\x10\x02\x12\x19\n\x15SOLEDAD_CREATING_KEYS\x10\x03\x12\x1e\n\x1aSOLEDAD_DONE_CREATING_KEYS\x10\x04\x12\x1a\n\x16SOLEDAD_UPLOADING_KEYS\x10\x05\x12\x1f\n\x1bSOLEDAD_DONE_UPLOADING_KEYS\x10\x06\x12\x1c\n\x18SOLEDAD_DOWNLOADING_KEYS\x10\x07\x12!\n\x1dSOLEDAD_DONE_DOWNLOADING_KEYS\x10\x08\x12\x1c\n\x18SOLEDAD_NEW_DATA_TO_SYNC\x10\t\x12\x1a\n\x16SOLEDAD_DONE_DATA_SYNC\x10\n\x12\x17\n\x13UPDATER_NEW_UPDATES\x10\x0b\x12\x19\n\x15UPDATER_DONE_UPDATING\x10\x0c\x12\x10\n\x0cRAISE_WINDOW\x10\r2\x91\x02\n\x13\x45ventsServerService\x12R\n\x08register\x12#.leap.common.events.RegisterRequest\x1a!.leap.common.events.EventResponse\x12V\n\nunregister\x12%.leap.common.events.UnregisterRequest\x1a!.leap.common.events.EventResponse\x12N\n\x06signal\x12!.leap.common.events.SignalRequest\x1a!.leap.common.events.EventResponse2h\n\x16\x45ventsComponentService\x12N\n\x06signal\x12!.leap.common.events.SignalRequest\x1a!.leap.common.events.EventResponseB\x03\x90\x01\x01') +    serialized_pb='\n\x0c\x65vents.proto\x12\x12leap.common.events\"\x97\x01\n\rSignalRequest\x12(\n\x05\x65vent\x18\x01 \x02(\x0e\x32\x19.leap.common.events.Event\x12\x0f\n\x07\x63ontent\x18\x02 \x02(\t\x12\x12\n\nmac_method\x18\x03 \x02(\t\x12\x0b\n\x03mac\x18\x04 \x02(\x0c\x12\x12\n\nenc_method\x18\x05 \x01(\t\x12\x16\n\x0e\x65rror_occurred\x18\x06 \x01(\x08\"j\n\x0fRegisterRequest\x12(\n\x05\x65vent\x18\x01 \x02(\x0e\x32\x19.leap.common.events.Event\x12\x0c\n\x04port\x18\x02 \x02(\x05\x12\x12\n\nmac_method\x18\x03 \x02(\t\x12\x0b\n\x03mac\x18\x04 \x02(\x0c\"l\n\x11UnregisterRequest\x12(\n\x05\x65vent\x18\x01 \x02(\x0e\x32\x19.leap.common.events.Event\x12\x0c\n\x04port\x18\x02 \x02(\x05\x12\x12\n\nmac_method\x18\x03 \x02(\t\x12\x0b\n\x03mac\x18\x04 \x02(\x0c\"\r\n\x0bPingRequest\"\x82\x01\n\rEventResponse\x12\x38\n\x06status\x18\x01 \x02(\x0e\x32(.leap.common.events.EventResponse.Status\x12\x0e\n\x06result\x18\x02 \x01(\t\"\'\n\x06Status\x12\x06\n\x02OK\x10\x01\x12\n\n\x06UNAUTH\x10\x02\x12\t\n\x05\x45RROR\x10\x03*\xe7\x02\n\x05\x45vent\x12\x15\n\x11\x43LIENT_SESSION_ID\x10\x01\x12\x0e\n\nCLIENT_UID\x10\x02\x12\x19\n\x15SOLEDAD_CREATING_KEYS\x10\x03\x12\x1e\n\x1aSOLEDAD_DONE_CREATING_KEYS\x10\x04\x12\x1a\n\x16SOLEDAD_UPLOADING_KEYS\x10\x05\x12\x1f\n\x1bSOLEDAD_DONE_UPLOADING_KEYS\x10\x06\x12\x1c\n\x18SOLEDAD_DOWNLOADING_KEYS\x10\x07\x12!\n\x1dSOLEDAD_DONE_DOWNLOADING_KEYS\x10\x08\x12\x1c\n\x18SOLEDAD_NEW_DATA_TO_SYNC\x10\t\x12\x1a\n\x16SOLEDAD_DONE_DATA_SYNC\x10\n\x12\x17\n\x13UPDATER_NEW_UPDATES\x10\x0b\x12\x19\n\x15UPDATER_DONE_UPDATING\x10\x0c\x12\x10\n\x0cRAISE_WINDOW\x10\r2\xdd\x02\n\x13\x45ventsServerService\x12J\n\x04ping\x12\x1f.leap.common.events.PingRequest\x1a!.leap.common.events.EventResponse\x12R\n\x08register\x12#.leap.common.events.RegisterRequest\x1a!.leap.common.events.EventResponse\x12V\n\nunregister\x12%.leap.common.events.UnregisterRequest\x1a!.leap.common.events.EventResponse\x12N\n\x06signal\x12!.leap.common.events.SignalRequest\x1a!.leap.common.events.EventResponse2\xb1\x01\n\x13\x45ventsClientService\x12J\n\x04ping\x12\x1f.leap.common.events.PingRequest\x1a!.leap.common.events.EventResponse\x12N\n\x06signal\x12!.leap.common.events.SignalRequest\x1a!.leap.common.events.EventResponseB\x03\x90\x01\x01')  _EVENT = _descriptor.EnumDescriptor(      name='Event', @@ -77,8 +77,8 @@ _EVENT = _descriptor.EnumDescriptor(      ],      containing_type=None,      options=None, -    serialized_start=542, -    serialized_end=901, +    serialized_start=557, +    serialized_end=916,  )  Event = enum_type_wrapper.EnumTypeWrapper(_EVENT) @@ -118,8 +118,8 @@ _EVENTRESPONSE_STATUS = _descriptor.EnumDescriptor(      ],      containing_type=None,      options=None, -    serialized_start=500, -    serialized_end=539, +    serialized_start=515, +    serialized_end=554,  ) @@ -284,6 +284,27 @@ _UNREGISTERREQUEST = _descriptor.Descriptor(  ) +_PINGREQUEST = _descriptor.Descriptor( +    name='PingRequest', +    full_name='leap.common.events.PingRequest', +    filename=None, +    file=DESCRIPTOR, +    containing_type=None, +    fields=[ +    ], +    extensions=[ +    ], +    nested_types=[], +    enum_types=[ +    ], +    options=None, +    is_extendable=False, +    extension_ranges=[], +    serialized_start=408, +    serialized_end=421, +) + +  _EVENTRESPONSE = _descriptor.Descriptor(      name='EventResponse',      full_name='leap.common.events.EventResponse', @@ -315,8 +336,8 @@ _EVENTRESPONSE = _descriptor.Descriptor(      options=None,      is_extendable=False,      extension_ranges=[], -    serialized_start=409, -    serialized_end=539, +    serialized_start=424, +    serialized_end=554,  )  _SIGNALREQUEST.fields_by_name['event'].enum_type = _EVENT @@ -327,6 +348,7 @@ _EVENTRESPONSE_STATUS.containing_type = _EVENTRESPONSE  DESCRIPTOR.message_types_by_name['SignalRequest'] = _SIGNALREQUEST  DESCRIPTOR.message_types_by_name['RegisterRequest'] = _REGISTERREQUEST  DESCRIPTOR.message_types_by_name['UnregisterRequest'] = _UNREGISTERREQUEST +DESCRIPTOR.message_types_by_name['PingRequest'] = _PINGREQUEST  DESCRIPTOR.message_types_by_name['EventResponse'] = _EVENTRESPONSE @@ -351,6 +373,13 @@ class UnregisterRequest(_message.Message):      # @@protoc_insertion_point(class_scope:leap.common.events.UnregisterRequest) +class PingRequest(_message.Message): +    __metaclass__ = _reflection.GeneratedProtocolMessageType +    DESCRIPTOR = _PINGREQUEST + +    # @@protoc_insertion_point(class_scope:leap.common.events.PingRequest) + +  class EventResponse(_message.Message):      __metaclass__ = _reflection.GeneratedProtocolMessageType      DESCRIPTOR = _EVENTRESPONSE @@ -368,13 +397,22 @@ _EVENTSSERVERSERVICE = _descriptor.ServiceDescriptor(      file=DESCRIPTOR,      index=0,      options=None, -    serialized_start=904, -    serialized_end=1177, +    serialized_start=919, +    serialized_end=1268,      methods=[          _descriptor.MethodDescriptor( +            name='ping', +            full_name='leap.common.events.EventsServerService.ping', +            index=0, +            containing_service=None, +            input_type=_PINGREQUEST, +            output_type=_EVENTRESPONSE, +            options=None, +        ), +        _descriptor.MethodDescriptor(              name='register',              full_name='leap.common.events.EventsServerService.register', -            index=0, +            index=1,              containing_service=None,              input_type=_REGISTERREQUEST,              output_type=_EVENTRESPONSE, @@ -383,7 +421,7 @@ _EVENTSSERVERSERVICE = _descriptor.ServiceDescriptor(          _descriptor.MethodDescriptor(              name='unregister',              full_name='leap.common.events.EventsServerService.unregister', -            index=1, +            index=2,              containing_service=None,              input_type=_UNREGISTERREQUEST,              output_type=_EVENTRESPONSE, @@ -392,7 +430,7 @@ _EVENTSSERVERSERVICE = _descriptor.ServiceDescriptor(          _descriptor.MethodDescriptor(              name='signal',              full_name='leap.common.events.EventsServerService.signal', -            index=2, +            index=3,              containing_service=None,              input_type=_SIGNALREQUEST,              output_type=_EVENTRESPONSE, @@ -411,20 +449,29 @@ class EventsServerService_Stub(EventsServerService):      DESCRIPTOR = _EVENTSSERVERSERVICE -_EVENTSCOMPONENTSERVICE = _descriptor.ServiceDescriptor( -    name='EventsComponentService', -    full_name='leap.common.events.EventsComponentService', +_EVENTSCLIENTSERVICE = _descriptor.ServiceDescriptor( +    name='EventsClientService', +    full_name='leap.common.events.EventsClientService',      file=DESCRIPTOR,      index=1,      options=None, -    serialized_start=1179, -    serialized_end=1283, +    serialized_start=1271, +    serialized_end=1448,      methods=[          _descriptor.MethodDescriptor( -            name='signal', -            full_name='leap.common.events.EventsComponentService.signal', +            name='ping', +            full_name='leap.common.events.EventsClientService.ping',              index=0,              containing_service=None, +            input_type=_PINGREQUEST, +            output_type=_EVENTRESPONSE, +            options=None, +        ), +        _descriptor.MethodDescriptor( +            name='signal', +            full_name='leap.common.events.EventsClientService.signal', +            index=1, +            containing_service=None,              input_type=_SIGNALREQUEST,              output_type=_EVENTRESPONSE,              options=None, @@ -432,13 +479,13 @@ _EVENTSCOMPONENTSERVICE = _descriptor.ServiceDescriptor(      ]) -class EventsComponentService(_service.Service): +class EventsClientService(_service.Service):      __metaclass__ = service_reflection.GeneratedServiceType -    DESCRIPTOR = _EVENTSCOMPONENTSERVICE +    DESCRIPTOR = _EVENTSCLIENTSERVICE -class EventsComponentService_Stub(EventsComponentService): +class EventsClientService_Stub(EventsClientService):      __metaclass__ = service_reflection.GeneratedServiceStubType -    DESCRIPTOR = _EVENTSCOMPONENTSERVICE +    DESCRIPTOR = _EVENTSCLIENTSERVICE  # @@protoc_insertion_point(module_scope) diff --git a/src/leap/common/events/server.py b/src/leap/common/events/server.py index d53c218..a7d4da9 100644 --- a/src/leap/common/events/server.py +++ b/src/leap/common/events/server.py @@ -17,12 +17,12 @@  """  A server for the events mechanism. -A server can receive different kinds of requests from components: +A server can receive different kinds of requests from clients: -  1. Registration request: store component port number to be notified when +  1. Registration request: store client port number to be notified when       a specific signal arrives. -  2. Signal request: redistribute the signal to registered components. +  2. Signal request: redistribute the signal to registered clients.  """  import logging  import socket @@ -40,12 +40,19 @@ logger = logging.getLogger(__name__)  SERVER_PORT = 8090 -# the `registered_components` dictionary below should have the following +# the `registered_clients` dictionary below should have the following  # format:  #  #     { event_signal: [ port, ... ], ... }  # -registered_components = {} +registered_clients = {} + + +class PortAlreadyTaken(Exception): +    """ +    Raised when trying to open a server in a port that is already taken. +    """ +    pass  def ensure_server(port=SERVER_PORT): @@ -60,40 +67,76 @@ def ensure_server(port=SERVER_PORT):      :return: the daemon instance or nothing      :rtype: EventsServerDaemon or None + +    :raise PortAlreadyTaken: Raised if C{port} is already taken by something +                             that is not an events server.      """      try: +        # check if port is available          s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)          s.connect(('localhost', port))          s.close() -        logger.info('Server is already running on port %d.', port) -        return None +        # port is taken, check if there's a server running there +        response = ping(port) +        if response is not None and response.status == proto.EventResponse.OK: +            logger.info('A server is already running on port %d.', port) +            return None +        # port is taken, and not by an events server +        logger.info('Port %d is taken by something not an events server.', port) +        raise PortAlreadyTaken(port)      except socket.error: +        # port is available, run a server          logger.info('Launching server on port %d.', port)          return EventsServerDaemon.ensure(port) +def ping(port=SERVER_PORT, reqcbk=None, timeout=1000): +    """ +    Ping the server. + +    :param port: the port in which server should be listening +    :type port: int +    :param reqcbk: a callback to be called when a response from server is +                   received +    :type reqcbk: function(leap.common.events.events_pb2.EventResponse) +    :param timeout: the timeout for synch calls +    :type timeout: int + +    :return: the response from server for synch calls or nothing for asynch +             calls. +    :rtype: leap.common.events.events_pb2.EventsResponse or None +    """ +    request = proto.PingRequest() +    service = RpcService( +        proto.EventsServerService_Stub, +        port, +        'localhost') +    logger.info("Pinging server in port %d..." % port) +    return service.ping(request, callback=reqcbk, timeout=timeout) + +  class EventsServerService(proto.EventsServerService):      """ -    Service for receiving events in components. +    Service for receiving events in clients.      """      def register(self, controller, request, done):          """ -        Register a component port to be signaled when specific events come in. +        Register a client port to be signaled when specific events come in.          :param controller: used to mediate a single method call          :type controller: protobuf.socketrpc.controller.SocketRpcController -        :param request: the request received from the component +        :param request: the request received from the client          :type request: leap.common.events.events_pb2.RegisterRequest          :param done: callback to be called when done          :type done: protobuf.socketrpc.server.Callback          """          logger.info("Received registration request: %s..." % str(request)[:40]) -        # add component port to signal list -        if request.event not in registered_components: -            registered_components[request.event] = set([]) -        registered_components[request.event].add(request.port) -        # send response back to component +        # add client port to signal list +        if request.event not in registered_clients: +            registered_clients[request.event] = set([]) +        registered_clients[request.event].add(request.port) +        # send response back to client          logger.debug('sending response back')          response = proto.EventResponse() @@ -102,56 +145,72 @@ class EventsServerService(proto.EventsServerService):      def unregister(self, controller, request, done):          """ -        Unregister a component port so it will not be signaled when specific +        Unregister a client port so it will not be signaled when specific          events come in.          :param controller: used to mediate a single method call          :type controller: protobuf.socketrpc.controller.SocketRpcController -        :param request: the request received from the component +        :param request: the request received from the client          :type request: leap.common.events.events_pb2.RegisterRequest          :param done: callback to be called when done          :type done: protobuf.socketrpc.server.Callback          """          logger.info(              "Received unregistration request: %s..." % str(request)[:40]) -        # remove component port from signal list +        # remove client port from signal list          response = proto.EventResponse() -        if request.event in registered_components: +        if request.event in registered_clients:              try: -                registered_components[request.event].remove(request.port) +                registered_clients[request.event].remove(request.port)                  response.status = proto.EventResponse.OK              except KeyError:                  response.status = proto.EventsResponse.ERROR                  response.result = 'Port %d not registered.' % request.port -        # send response back to component +        # send response back to client          logger.debug('sending response back')          done.run(response)      def signal(self, controller, request, done):          """ -        Perform an RPC call to signal all components registered to receive a +        Perform an RPC call to signal all clients registered to receive a          specific signal.          :param controller: used to mediate a single method call          :type controller: protobuf.socketrpc.controller.SocketRpcController -        :param request: the request received from the component +        :param request: the request received from the client          :type request: leap.common.events.events_pb2.SignalRequest          :param done: callback to be called when done          :type done: protobuf.socketrpc.server.Callback          """ -        logger.info('Received signal from component: %s...', str(request)[:40]) -        # send signal to all registered components +        logger.info('Received signal from client: %s...', str(request)[:40]) +        # send signal to all registered clients          # TODO: verify signal auth -        if request.event in registered_components: -            for port in registered_components[request.event]: +        if request.event in registered_clients: +            for port in registered_clients[request.event]:                  def callback(req, resp):                      logger.info("Signal received by " + str(port)) -                service = RpcService(proto.EventsComponentService_Stub, +                service = RpcService(proto.EventsClientService_Stub,                                       port, 'localhost')                  service.signal(request, callback=callback) -        # send response back to component +        # send response back to client +        response = proto.EventResponse() +        response.status = proto.EventResponse.OK +        done.run(response) + +    def ping(self, controller, request, done): +        """ +        Reply to a ping request. + +        :param controller: used to mediate a single method call +        :type controller: protobuf.socketrpc.controller.SocketRpcController +        :param request: the request received from the client +        :type request: leap.common.events.events_pb2.RegisterRequest +        :param done: callback to be called when done +        :type done: protobuf.socketrpc.server.Callback +        """ +        logger.info("Received ping request, sending response.")          response = proto.EventResponse()          response.status = proto.EventResponse.OK          done.run(response) diff --git a/src/leap/common/tests/test_events.py b/src/leap/common/tests/test_events.py index 7286bdc..bc04dd6 100644 --- a/src/leap/common/tests/test_events.py +++ b/src/leap/common/tests/test_events.py @@ -18,19 +18,27 @@  import unittest  import sets  import time +import socket +import threading +import random + + +from mock import Mock  from protobuf.socketrpc import RpcService  from leap.common import events  from leap.common.events import (      server, -    component, +    client,      mac_auth,  )  from leap.common.events.events_pb2 import (      EventsServerService,      EventsServerService_Stub, +    EventsClientService_Stub,      EventResponse,      SignalRequest,      RegisterRequest, +    PingRequest,      SOLEDAD_CREATING_KEYS,      CLIENT_UID,  ) @@ -39,11 +47,6 @@ from leap.common.events.events_pb2 import (  port = 8090  received = False -local_callback_executed = False - - -def callback(request, reponse): -    return True  class EventsTestCase(unittest.TestCase): @@ -51,7 +54,7 @@ class EventsTestCase(unittest.TestCase):      @classmethod      def setUpClass(cls):          server.EventsServerDaemon.ensure(8090) -        cls.callbacks = events.component.registered_callbacks +        cls.callbacks = events.client.registered_callbacks      @classmethod      def tearDownClass(cls): @@ -62,8 +65,8 @@ class EventsTestCase(unittest.TestCase):          super(EventsTestCase, self).setUp()      def tearDown(self): -        #events.component.registered_callbacks = {} -        server.registered_components = {} +        #events.client.registered_callbacks = {} +        server.registered_clients = {}          super(EventsTestCase, self).tearDown()      def test_service_singleton(self): @@ -76,24 +79,24 @@ class EventsTestCase(unittest.TestCase):          self.assertEqual(service1, service2,                           "Can't get singleton class for service.") -    def test_component_register(self): +    def test_client_register(self):          """ -        Ensure components can register callbacks. +        Ensure clients can register callbacks.          """          self.assertTrue(1 not in self.callbacks,                          'There should should be no callback for this signal.')          events.register(1, lambda x: True)          self.assertTrue(1 in self.callbacks, -                        'Could not register signal in local component.') +                        'Could not register signal in local client.')          events.register(2, lambda x: True)          self.assertTrue(1 in self.callbacks, -                        'Could not register signal in local component.') +                        'Could not register signal in local client.')          self.assertTrue(2 in self.callbacks, -                        'Could not register signal in local component.') +                        'Could not register signal in local client.')      def test_register_signal_replace(self):          """ -        Make sure components can replace already registered callbacks. +        Make sure clients can replace already registered callbacks.          """          sig = 3          cbk = lambda x: True @@ -120,21 +123,32 @@ class EventsTestCase(unittest.TestCase):          response = service.signal(request, timeout=1000)          self.assertEqual(EventResponse.OK, response.status,                           'Wrong response status.') -        # test asynch -        def local_callback(request, response): -            global local_callback_executed -            local_callback_executed = True +    def test_signal_executes_callback(self): +        """ +        Ensure callback is executed upon receiving signal. +        """ +        sig = CLIENT_UID +        request = SignalRequest() +        request.event = sig +        request.content = 'my signal contents' +        request.mac_method = mac_auth.MacMethod.MAC_NONE +        request.mac = "" +        service = RpcService(EventsServerService_Stub, port, 'localhost') -        events.register(sig, local_callback) -        service.signal(request, callback=local_callback) -        time.sleep(0.1) -        self.assertTrue(local_callback_executed, -                        'Local callback did not execute.') +        # register a callback +        flag = Mock() +        events.register(sig, lambda req: flag(req.event)) +        # signal +        response = service.signal(request) +        self.assertEqual(EventResponse.OK, response.status, +                         'Wrong response status.') +        time.sleep(1)  # wait for signal to arrive +        flag.assert_called_once_with(sig)      def test_events_server_service_register(self):          """ -        Ensure the server can register components to be signaled. +        Ensure the server can register clients to be signaled.          """          sig = 5          request = RegisterRequest() @@ -143,42 +157,39 @@ class EventsTestCase(unittest.TestCase):          request.mac_method = mac_auth.MacMethod.MAC_NONE          request.mac = ""          service = RpcService(EventsServerService_Stub, port, 'localhost') -        complist = server.registered_components +        complist = server.registered_clients          self.assertEqual({}, complist,                           'There should be no registered_ports when '                           'server has just been created.')          response = service.register(request, timeout=1000)          self.assertTrue(sig in complist, "Signal not registered succesfully.")          self.assertTrue(8091 in complist[sig], -                        'Failed registering component port.') +                        'Failed registering client port.') -    def test_component_request_register(self): +    def test_client_request_register(self):          """ -        Ensure components can register themselves with server. +        Ensure clients can register themselves with server.          """          sig = 6 -        complist = server.registered_components +        complist = server.registered_clients          self.assertTrue(sig not in complist, -                        'There should be no registered components for this ' +                        'There should be no registered clients for this '                          'signal.')          events.register(sig, lambda x: True)          time.sleep(0.1) -        port = component.EventsComponentDaemon.get_instance().get_port() -        self.assertTrue(sig in complist, 'Failed registering component.') +        port = client.EventsClientDaemon.get_instance().get_port() +        self.assertTrue(sig in complist, 'Failed registering client.')          self.assertTrue(port in complist[sig], -                        'Failed registering component port.') +                        'Failed registering client port.') -    def test_component_receives_signal(self): +    def test_client_receives_signal(self):          """ -        Ensure components can receive signals. +        Ensure clients can receive signals.          """          sig = 7 +        flag = Mock() -        def getsig(param=None): -            global received -            received = True - -        events.register(sig, getsig) +        events.register(sig, lambda req: flag(req.event))          request = SignalRequest()          request.event = sig          request.content = "" @@ -188,46 +199,230 @@ class EventsTestCase(unittest.TestCase):          response = service.signal(request, timeout=1000)          self.assertTrue(response is not None, 'Did not receive response.')          time.sleep(0.5) -        self.assertTrue(received, 'Did not receive signal back.') +        flag.assert_called_once_with(sig) -    def test_component_send_signal(self): +    def test_client_send_signal(self):          """ -        Ensure components can send signals. +        Ensure clients can send signals.          """          sig = 8          response = events.signal(sig)          self.assertTrue(response.status == response.OK,                          'Received wrong response status when signaling.') -    def test_component_unregister_all(self): +    def test_client_unregister_all(self):          """ -        Test that the component can unregister all events for one signal. +        Test that the client can unregister all events for one signal.          """          sig = CLIENT_UID -        complist = server.registered_components +        complist = server.registered_clients          events.register(sig, lambda x: True)          events.register(sig, lambda x: True)          time.sleep(0.1)          events.unregister(sig)          time.sleep(0.1) -        port = component.EventsComponentDaemon.get_instance().get_port() +        port = client.EventsClientDaemon.get_instance().get_port()          self.assertFalse(bool(complist[sig]))          self.assertTrue(port not in complist[sig]) -    def test_component_unregister_by_uid(self): +    def test_client_unregister_by_uid(self):          """ -        Test that the component can unregister an event by uid. +        Test that the client can unregister an event by uid.          """          sig = CLIENT_UID -        complist = server.registered_components +        complist = server.registered_clients          events.register(sig, lambda x: True, uid='cbkuid')          events.register(sig, lambda x: True, uid='cbkuid2')          time.sleep(0.1)          events.unregister(sig, uid='cbkuid')          time.sleep(0.1) -        port = component.EventsComponentDaemon.get_instance().get_port() +        port = client.EventsClientDaemon.get_instance().get_port()          self.assertTrue(sig in complist)          self.assertTrue(len(complist[sig]) == 1)          self.assertTrue( -            component.registered_callbacks[sig].pop()[0] == 'cbkuid2') +            client.registered_callbacks[sig].pop()[0] == 'cbkuid2')          self.assertTrue(port in complist[sig]) + +    def test_server_replies_ping(self): +        """ +        Ensure server replies to a ping. +        """ +        request = PingRequest() +        service = RpcService(EventsServerService_Stub, port, 'localhost') +        response = service.ping(request, timeout=1000) +        self.assertIsNotNone(response) +        self.assertEqual(EventResponse.OK, response.status, +                         'Wrong response status.') + +    def test_client_replies_ping(self): +        """ +        Ensure clients reply to a ping. +        """ +        daemon = client.ensure_client_daemon() +        port = daemon.get_port() +        request = PingRequest() +        service = RpcService(EventsClientService_Stub, port, 'localhost') +        response = service.ping(request, timeout=1000) +        self.assertEqual(EventResponse.OK, response.status, +                         'Wrong response status.') + +    def test_server_ping(self): +        """ +        Ensure the function from server module pings correctly. +        """ +        response = server.ping() +        self.assertIsNotNone(response) +        self.assertEqual(EventResponse.OK, response.status, +                         'Wrong response status.') + +    def test_client_ping(self): +        """ +        Ensure the function from client module pings correctly. +        """ +        daemon = client.ensure_client_daemon() +        response = client.ping(daemon.get_port()) +        self.assertIsNotNone(response) +        self.assertEqual(EventResponse.OK, response.status, +                         'Wrong response status.') + +    def test_module_ping_server(self): +        """ +        Ensure the function from main module pings server correctly. +        """ +        response = events.ping_server() +        self.assertIsNotNone(response) +        self.assertEqual(EventResponse.OK, response.status, +                         'Wrong response status.') + +    def test_module_ping_client(self): +        """ +        Ensure the function from main module pings clients correctly. +        """ +        daemon = client.ensure_client_daemon() +        response = events.ping_client(daemon.get_port()) +        self.assertIsNotNone(response) +        self.assertEqual(EventResponse.OK, response.status, +                         'Wrong response status.') + +    def test_ensure_server_raises_if_port_taken(self): +        """ +        Verify that server raises an exception if port is already taken. +        """ +        # get a random free port +        while True: +            port = random.randint(1024, 65535) +            try: +                s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) +                s.connect(('localhost', port)) +                s.close() +            except: +                break + +        class PortBlocker(threading.Thread): + +            def run(self): +                conns = 0 +                s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) +                s.bind(('localhost', port)) +                s.setblocking(1) +                s.listen(1) +                while conns < 2:  # blocks until rece +                    conns += 1 +                    s.accept() +                s.close() + +        # block the port +        taker = PortBlocker() +        taker.start() +        time.sleep(1)  # wait for thread to start. +        self.assertRaises( +            server.PortAlreadyTaken, server.ensure_server, port) + +    def test_async_register(self): +        """ +        Test asynchronous registering of callbacks. +        """ +        flag = Mock() + +        # executed after async register, when response is received from server +        def reqcbk(request, response): +            flag(request.event) + +        # callback registered by application +        def callback(request): +            pass + +        # passing a callback as reqcbk param makes the call asynchronous +        result = events.register(CLIENT_UID, callback, reqcbk=reqcbk) +        self.assertIsNone(result) +        events.signal(CLIENT_UID) +        time.sleep(1)  # wait for signal to arrive from server +        flag.assert_called_once_with(CLIENT_UID) + +    def test_async_signal(self): +        """ +        Test asynchronous registering of callbacks. +        """ +        flag = Mock() + +        # executed after async signal, when response is received from server +        def reqcbk(request, response): +            flag(request.event) + +        # passing a callback as reqcbk param makes the call asynchronous +        result = events.signal(CLIENT_UID, reqcbk=reqcbk) +        self.assertIsNone(result) +        time.sleep(1)  # wait for signal to arrive from server +        flag.assert_called_once_with(CLIENT_UID) + +    def test_async_unregister(self): +        """ +        Test asynchronous unregistering of callbacks. +        """ +        flag = Mock() + +        # executed after async signal, when response is received from server +        def reqcbk(request, response): +            flag(request.event) + +        # callback registered by application +        def callback(request): +            pass + +        # passing a callback as reqcbk param makes the call asynchronous +        events.register(CLIENT_UID, callback) +        result = events.unregister(CLIENT_UID, reqcbk=reqcbk) +        self.assertIsNone(result) +        time.sleep(1)  # wait for signal to arrive from server +        flag.assert_called_once_with(CLIENT_UID) + +    def test_async_ping_server(self): +        """ +        Test asynchronous pinging of server. +        """ +        flag = Mock() + +        # executed after async signal, when response is received from server +        def reqcbk(request, response): +            flag() + +        result = events.ping_server(reqcbk=reqcbk) +        self.assertIsNone(result) +        time.sleep(1)  # wait for response to arrive from server. +        flag.assert_called_once_with() + +    def test_async_ping_client(self): +        """ +        Test asynchronous pinging of client. +        """ +        flag = Mock() + +        # executed after async signal, when response is received from server +        def reqcbk(request, response): +            flag() + +        daemon = client.ensure_client_daemon() +        result = events.ping_client(daemon.get_port(), reqcbk=reqcbk) +        self.assertIsNone(result) +        time.sleep(1)  # wait for response to arrive from server. +        flag.assert_called_once_with() | 
