summaryrefslogtreecommitdiff
path: root/src/leap/common/events/daemon.py
diff options
context:
space:
mode:
Diffstat (limited to 'src/leap/common/events/daemon.py')
-rw-r--r--src/leap/common/events/daemon.py208
1 files changed, 208 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..c253948
--- /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)
+ 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