summaryrefslogtreecommitdiff
path: root/src/leap/common/events/server.py
diff options
context:
space:
mode:
authorMicah Anderson <micah@riseup.net>2013-08-13 14:25:57 -0400
committerMicah Anderson <micah@riseup.net>2013-08-13 14:27:01 -0400
commitddff9c2dda51fd28dcdc8e2db998d7254f69a41b (patch)
treeb000723d9c7e54f0ac163e54f8d0d6fcd829209a /src/leap/common/events/server.py
parentcc8dd1e7f3d64ae6d7d69ed08a4346bfabfe7b22 (diff)
parent0e721b1b47c3b94f6d4d6709e34b6b816f9fd810 (diff)
Merge tag '0.3.0' into debian
Tag leap.common version 0.3.0
Diffstat (limited to 'src/leap/common/events/server.py')
-rw-r--r--src/leap/common/events/server.py132
1 files changed, 109 insertions, 23 deletions
diff --git a/src/leap/common/events/server.py b/src/leap/common/events/server.py
index 1f3a874..59f3454 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,71 +67,150 @@ 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(proto.PingRequest, proto.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()
response.status = proto.EventResponse.OK
done.run(response)
+ def unregister(self, controller, request, done):
+ """
+ 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 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 client port from signal list
+ response = proto.EventResponse()
+ if request.event in registered_clients:
+ try:
+ 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 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)