Add a server for events mechanism.

2 files changed, 358 insertions, 0 deletions

diff --git a/src/leap/common/events/daemon.py b/src/leap/common/events/daemon.py
new file mode 100644
index 0000000..40242c9
--- /dev/null
+++ b/src/leap/common/events/daemon.py
@@ -0,0 +1,208 @@
+# -*- coding: utf-8 -*-
+# daemon.py
+# Copyright (C) 2013 LEAP
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program. If not, see <http://www.gnu.org/licenses/>.
+
+"""
+A singleton daemon for running RPC services using protobuf.socketrpc.
+"""
+
+
+import logging
+import threading
+
+
+from protobuf.socketrpc.server import (
+    SocketRpcServer,
+    ThreadedTCPServer,
+    SocketHandler,
+)
+
+
+logger = logging.getLogger(__name__)
+
+
+class ServiceAlreadyRunningException(Exception):
+    """
+    Raised whenever a service is already running in this process but someone
+    attemped to start it in a different port.
+    """
+
+
+class EventsRpcServer(SocketRpcServer):
+    """
+    RPC server used in server and component interfaces to receive messages.
+    """
+
+    def __init__(self, port, host='localhost'):
+        """
+        Initialize a RPC server.
+
+        @param port: the port in which to listen for incoming messages
+        @type port: int
+        @param host: the address to bind to
+        @type host: str
+        """
+        SocketRpcServer.__init__(self, port, host)
+        self._server = None
+
+    def run(self):
+        """
+        Run the server.
+        """
+        logger.info('Running server on port %d.' % self.port)
+        # parent implementation does not hold the server instance, so we do it
+        # here.
+        self._server = ThreadedTCPServer((self.host, self.port),
+                                        SocketHandler, self)
+        # if we chose to use a random port, fetch the port number info.
+        if self.port is 0:
+            self.port = self._server.socket.getsockname()[1]
+        self._server.serve_forever()
+
+    def stop(self):
+        """
+        Stop the server.
+        """
+        self._server.shutdown()
+
+
+class EventsSingletonDaemon(threading.Thread):
+    """
+    Singleton class for for launching and terminating a daemon.
+
+    This class is used so every part of the mechanism that needs to listen for
+    messages can launch its own daemon (thread) to do the job.
+    """
+
+    # Singleton instance
+    __instance = None
+
+    def __new__(cls, *args, **kwargs):
+        """
+        Return a singleton instance if it exists or create and initialize one.
+        """
+        if len(args) is not 2:
+            raise TypeError("__init__() takes exactly 2 arguments (%d given)"
+                            % len(args))
+        if cls.__instance is None:
+            cls.__instance = object.__new__(
+                EventsSingletonDaemon, *args, **kwargs)
+            cls.__initialize(cls.__instance, args[0], args[1])
+        return cls.__instance
+
+    @staticmethod
+    def __initialize(self, port, service):
+        """
+        Initialize a singleton daemon.
+
+        This is a static method disguised as instance method that actually
+        does the initialization of the daemon instance.
+
+        @param port: the port in which to listen for incoming messages
+        @type port: int
+        @param service: the service to provide in this daemon
+        @type service: google.protobuf.service.Service
+        """
+        threading.Thread.__init__(self)
+        self._port = port
+        self._service = service
+        self._server = EventsRpcServer(self._port)
+        self._server.registerService(self._service)
+        self.daemon = True
+
+    def __init__(self):
+        """
+        Singleton placeholder initialization method.
+
+        Initialization is made in __new__ so we can always return the same
+        instance upon object creation.
+        """
+        pass
+
+    @classmethod
+    def ensure(cls, port):
+        """
+        Make sure the daemon instance is running.
+
+        Each implementation of this method should call `self.ensure_service`
+        with the appropriate service from the `events.proto` definitions, and
+        return the daemon instance.
+
+        @param port: the port in which the daemon should be listening
+        @type port: int
+
+        @return: a daemon instance
+        @rtype: EventsSingletonDaemon
+        """
+        raise NotImplementedError(self.ensure)
+
+    @classmethod
+    def ensure_service(cls, port, service):
+        """
+        Start the singleton instance if not already running.
+
+        Might return ServiceAlreadyRunningException
+
+        @param port: the port in which the daemon should be listening
+        @type port: int
+
+        @return: a daemon instance
+        @rtype: EventsSingletonDaemon
+        """
+        daemon = cls(port, service)
+        if not daemon.is_alive():
+            daemon.start()
+        elif port and port != cls.__instance._port:
+            # service is running in this process but someone is trying to
+            # start it in another port
+            raise ServiceAlreadyRunningException(
+                "Service is already running in this process on port %d."
+                % self.__instance._port)
+        return daemon
+
+    @classmethod
+    def get_instance(cls):
+        """
+        Retrieve singleton instance of this daemon.
+
+        @return: a daemon instance
+        @rtype: EventsSingletonDaemon
+        """
+        return cls.__instance
+
+    def run(self):
+        """
+        Run the server.
+        """
+        self._server.run()
+
+    def stop(self):
+        """
+        Stop the daemon.
+        """
+        self._server.stop()
+
+    def get_port(self):
+        """
+        Retrieve the value of the port to which the service running in this
+        daemon is binded to.
+
+        @return: the port to which the daemon is binded to
+        @rtype: int
+        """
+        if self._port is 0:
+            self._port = self._server.port
+        return self._port
diff --git a/src/leap/common/events/server.py b/src/leap/common/events/server.py
new file mode 100644
index 0000000..5cc1add
--- /dev/null
+++ b/src/leap/common/events/server.py
@@ -0,0 +1,150 @@
+# -*- coding: utf-8 -*-
+# server.py
+# Copyright (C) 2013 LEAP
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program. If not, see <http://www.gnu.org/licenses/>.
+
+"""
+A server for the events mechanism.
+
+A server can receive different kinds of requests from components:
+
+  1. Registration request: store component port number to be notified when
+     a specific signal arrives.
+
+  2. Signal request: redistribute the signal to registered components.
+"""
+
+
+import logging
+import sets
+
+
+from protobuf.socketrpc import RpcService
+from leap.common.events import (
+    events_pb2 as proto,
+    daemon,
+)
+
+
+logger = logging.getLogger(__name__)
+
+
+SERVER_PORT = 8090
+
+# the `registered_components` dictionary below should have the following
+# format:
+#
+#     { event_signal: [ port, ... ], ... }
+#
+registered_components = {}
+
+
+def ensure_server(port=SERVER_PORT):
+    """
+    Make sure the server is running on the given port.
+
+    Attempt to connect to given local port. Upon success, assume that the
+    events server has already been started. Upon failure, start events server.
+
+    @param port: the port in which server should be listening
+    @type port: int
+
+    @return: the daemon instance or nothing
+    @rtype: EventsServerDaemon or None
+    """
+    try:
+        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
+    except socket.error:
+        logger.info('Launching server on port %d.', port)
+        return EventsServerDaemon.ensure(port)
+
+
+class EventsServerService(proto.EventsServerService):
+    """
+    Service for receiving events in components.
+    """
+
+    def register(self, controller, request, done):
+        """
+        Register a component 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
+        @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))
+        # add component port to signal list
+        if request.event not in registered_components:
+            registered_components[request.event] = sets.Set()
+        registered_components[request.event].add(request.port)
+        # send response back to component
+        response = proto.EventResponse()
+        response.status = proto.EventResponse.OK
+        done.run(response)
+
+    def signal(self, controller, request, done):
+        """
+        Perform an RPC call to signal all components 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
+        @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))
+        # send signal to all registered components
+        # TODO: verify signal auth
+        if request.event in registered_components:
+            for port in registered_components[request.event]:
+
+                def callback(req, resp):
+                    logger.info("Signal received by " + str(port))
+
+                service = RpcService(proto.EventsComponentService_Stub,
+                                     port, 'localhost')
+                service.signal(request, callback=callback)
+        # send response back to component
+        response = proto.EventResponse()
+        response.status = proto.EventResponse.OK
+        done.run(response)
+
+
+class EventsServerDaemon(daemon.EventsSingletonDaemon):
+    """
+    Singleton class for starting an events server daemon.
+    """
+
+    @classmethod
+    def ensure(cls, port):
+        """
+        Make sure the daemon is running on the given port.
+
+        @param port: the port in which the daemon should listen
+        @type port: int
+
+        @return: a daemon instance
+        @rtype: EventsServerDaemon
+        """
+        return cls.ensure_service(port, EventsServerService())