#!/usr/bin/env python2 # -*- coding: utf-8 # Copyright (C) 2010 Stefan Hacker # All rights reserved. # # Redistribution and use in source and binary forms, with or without # modification, are permitted provided that the following conditions # are met: # - Redistributions of source code must retain the above copyright notice, # this list of conditions and the following disclaimer. # - Redistributions in binary form must reproduce the above copyright notice, # this list of conditions and the following disclaimer in the documentation # and/or other materials provided with the distribution. # - Neither the name of the Mumble Developers nor the names of its # contributors may be used to endorse or promote products derived from this # software without specific prior written permission. # THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS # `AS IS'' AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT # LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR # A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE FOUNDATION OR # CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, # EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, # PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR # PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF # LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING # NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS # SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. import Queue from worker import Worker, local_thread, local_thread_blocking from config import Config import sys import os import uuid class FailedLoadModuleException(Exception): pass class FailedLoadModuleConfigException(FailedLoadModuleException): pass class FailedLoadModuleImportException(FailedLoadModuleException): pass class FailedLoadModuleInitializationException(FailedLoadModuleException): pass def debug_log(enable = True): def new_dec(fu): def new_fu(*args, **kwargs): self = args[0] log = self.log() skwargs = ','.join(['%s=%s' % (karg,repr(arg)) for karg, arg in kwargs]) sargs = ','.join([str(arg) for arg in args[1:]]) + '' if not skwargs else (',' + str(skwargs)) call = "%s(%s)" % (fu.__name__, sargs) log.debug(call) res = fu(*args, **kwargs) log.debug("%s -> %s", call, repr(res)) return res return new_fu if enable else fu return new_dec debug_me = True class MumoManagerRemote(object): """ Manager object handed to MumoModules. This module acts as a remote for the MumoModule with which it can register/unregister to/from callbacks as well as do other signaling to the master MumoManager. """ SERVERS_ALL = [-1] ## Applies to all servers def __init__(self, master, name, queue): self.__master = master self.__name = name self.__queue = queue self.__context_callbacks = {} # server -> action -> callback def getQueue(self): return self.__queue def subscribeMetaCallbacks(self, handler, servers = SERVERS_ALL): """ Subscribe to meta callbacks. Subscribes the given handler to the following callbacks: >>> started(self, server, context = None) >>> stopped(self, server, context = None) @param servers: List of server IDs for which to subscribe. To subscribe to all servers pass SERVERS_ALL. @param handler: Object on which to call the callback functions """ return self.__master.subscribeMetaCallbacks(self.__queue, handler, servers) def unsubscribeMetaCallbacks(self, handler, servers = SERVERS_ALL): """ Unsubscribe from meta callbacks. Unsubscribes the given handler from callbacks for the given servers. @param servers: List of server IDs for which to unsubscribe. To unsubscribe from all servers pass SERVERS_ALL. @param handler: Subscribed handler """ return self.__master.unscubscribeMetaCallbacks(self.__queue, handler, servers) def subscribeServerCallbacks(self, handler, servers = SERVERS_ALL): """ Subscribe to server callbacks. Subscribes the given handler to the following callbacks: >>> userConnected(self, state, context = None) >>> userDisconnected(self, state, context = None) >>> userStateChanged(self, state, context = None) >>> channelCreated(self, state, context = None) >>> channelRemoved(self, state, context = None) >>> channelStateChanged(self, state, context = None) @param servers: List of server IDs for which to subscribe. To subscribe to all servers pass SERVERS_ALL. @param handler: Object on which to call the callback functions """ return self.__master.subscribeServerCallbacks(self.__queue, handler, servers) def unsubscribeServerCallbacks(self, handler, servers = SERVERS_ALL): """ Unsubscribe from server callbacks. Unsubscribes the given handler from callbacks for the given servers. @param servers: List of server IDs for which to unsubscribe. To unsubscribe from all servers pass SERVERS_ALL. @param handler: Subscribed handler """ return self.__master.unsubscribeServerCallbacks(self.__queue, handler, servers) def getUniqueAction(self): """ Returns a unique action string that can be used in addContextMenuEntry. :return: Unique action string """ return str(uuid.uuid4()) def addContextMenuEntry(self, server, user, action, text, handler, context): """ Adds a new context callback menu entry with the given text for the given user. You can use the same action identifier for multiple users entries to simplify your handling. However make sure an action identifier is unique to your module. The easiest way to achieve this is to use getUniqueAction to generate a guaranteed unique one. Your handler should be of form: >>> handler(self, server, action, user, target) Here server is the server the user who triggered the action resides on. Target identifies what the context action was invoked on. It can be either a User, Channel or None. @param server: Server the user resides on @param user: User to add entry for @param action: Action identifier passed to your callback (see above) @param text: Text for the menu entry @param handler: Handler function to call when the menu item is used @param context: Contexts to show entry in (can be a combination of ContextServer, ContextChannel and ContextUser) """ server_actions = self.__context_callbacks.get(server.id()) if not server_actions: server_actions = {} self.__context_callbacks[server.id()] = server_actions action_cb = server_actions.get(action) if not action_cb: # We need to create an register a new context callback action_cb = self.__master.createContextCallback(self.__handle_context_callback, handler, server) server_actions[action] = action_cb server.addContextCallback(user.session, action, text, action_cb, context) def __handle_context_callback(self, handler, server, action, user, target_session, target_channelid, current=None): """ Small callback wrapper for context menu operations. Translates the given target into the corresponding object and schedules a call to the actual user context menu handler which will be executed in the modules thread. """ if target_session != 0: target = server.getState(target_session) elif target_channelid != -1: target = server.getChannelState(target_channelid) else: target = None # Schedule a call to the handler self.__queue.put((None, handler, [server, action, user, target], {})) def removeContextMenuEntry(self, server, action): """ Removes a previously created context action callback from a server. Applies to all users that share the action on this server. @param server Server the action should be removed from. @param action Action to remove """ try: cb = self.__context_callbacks[server.id()].pop(action) except KeyError: # Nothing to unregister return server.removeContextCallback(cb) def getMurmurModule(self): """ Returns the Murmur module generated from the slice file """ return self.__master.getMurmurModule() def getMeta(self): """ Returns the connected servers meta module or None if it is not available """ return self.__master.getMeta() class MumoManager(Worker): MAGIC_ALL = -1 cfg_default = {'modules':(('mod_dir', str, "modules/"), ('cfg_dir', str, "modules-enabled/"), ('timeout', int, 2))} def __init__(self, murmur, context_callback_type, cfg = Config(default = cfg_default)): Worker.__init__(self, "MumoManager") self.queues = {} # {queue:module} self.modules = {} # {name:module} self.imports = {} # {name:import} self.cfg = cfg self.murmur = murmur self.meta = None self.client_adapter = None self.metaCallbacks = {} # {sid:{queue:[handler]}} self.serverCallbacks = {} self.context_callback_type = context_callback_type def setClientAdapter(self, client_adapter): """ Sets the ice adapter used for client-side callbacks. This is needed in case per-module callbacks have to be attached during run-time as is the case for context callbacks. :param client_adapter: Ice object adapter """ self.client_adapter = client_adapter def __add_to_dict(self, mdict, queue, handler, servers): for server in servers: if server in mdict: if queue in mdict[server]: if not handler in mdict[server][queue]: mdict[server][queue].append(handler) else: mdict[server][queue] = [handler] else: mdict[server] = {queue:[handler]} def __rem_from_dict(self, mdict, queue, handler, servers): for server in servers: try: mdict[server][queue].remove(handler) except KeyError, ValueError: pass def __announce_to_dict(self, mdict, server, function, *args, **kwargs): """ Call function on handlers for specific servers in one of our handler dictionaries. @param mdict Dictionary to announce to @param server Server to announce to, ALL is always implied @param function Function the handler should call @param args Arguments for the function @param kwargs Keyword arguments for the function """ # Announce to all handlers of the given serverlist if server == self.MAGIC_ALL: servers = mdict.iterkeys() else: servers = [self.MAGIC_ALL, server] for server in servers: try: for queue, handlers in mdict[server].iteritems(): for handler in handlers: self.__call_remote(queue, handler, function, *args, **kwargs) except KeyError: # No handler registered for that server pass def __call_remote(self, queue, handler, function, *args, **kwargs): try: func = getattr(handler, function) # Find out what to call on target queue.put((None, func, args, kwargs)) except AttributeError, e: mod = self.queues.get(queue, None) myname = "" for name, mymod in self.modules.iteritems(): if mod == mymod: myname = name if myname: self.log().error("Handler class registered by module '%s' does not handle function '%s'. Call failed.", myname, function) else: self.log().exception(e) # #-- Module multiplexing functionality # @local_thread def announceConnected(self, meta = None): """ Call connected handler on all handlers """ self.meta = meta for queue, module in self.queues.iteritems(): self.__call_remote(queue, module, "connected") @local_thread def announceDisconnected(self): """ Call disconnected handler on all handlers """ for queue, module in self.queues.iteritems(): self.__call_remote(queue, module, "disconnected") @local_thread def announceMeta(self, server, function, *args, **kwargs): """ Call a function on the meta handlers @param server Server to announce to @param function Name of the function to call on the handler @param args List of arguments @param kwargs List of keyword arguments """ self.__announce_to_dict(self.metaCallbacks, server, function, *args, **kwargs) @local_thread def announceServer(self, server, function, *args, **kwargs): """ Call a function on the server handlers @param server Server to announce to @param function Name of the function to call on the handler @param args List of arguments @param kwargs List of keyword arguments """ self.__announce_to_dict(self.serverCallbacks, server, function, *args, **kwargs) # #--- Module self management functionality # @local_thread def subscribeMetaCallbacks(self, queue, handler, servers): """ @param queue Target worker queue @see MumoManagerRemote """ return self.__add_to_dict(self.metaCallbacks, queue, handler, servers) @local_thread def unsubscribeMetaCallbacks(self, queue, handler, servers): """ @param queue Target worker queue @see MumoManagerRemote """ return self.__rem_from_dict(self.metaCallbacks, queue, handler, servers) @local_thread def subscribeServerCallbacks(self, queue, handler, servers): """ @param queue Target worker queue @see MumoManagerRemote """ return self.__add_to_dict(self.serverCallbacks, queue, handler, servers) @local_thread def unsubscribeServerCallbacks(self, queue, handler, servers): """ @param queue Target worker queue @see MumoManagerRemote """ return self.__rem_from_dict(self.serverCallbacks, queue, handler, servers) def getMurmurModule(self): """ Returns the Murmur module generated from the slice file """ return self.murmur def createContextCallback(self, callback, *ctx): """ Creates a new context callback handler class instance. @param callback Callback to set for handler @param *ctx Additional context parameters passed to callback before the actual parameters. @return Murmur ServerContextCallbackPrx object for the context callback handler class. """ contextcbprx = self.client_adapter.addWithUUID(self.context_callback_type(callback, *ctx)) contextcb = self.murmur.ServerContextCallbackPrx.uncheckedCast(contextcbprx) return contextcb def getMeta(self): """ Returns the connected servers meta module or None if it is not available """ return self.meta #--- Module load/start/stop/unload functionality # @local_thread_blocking @debug_log(debug_me) def loadModules(self, names = None): """ Loads a list of modules from the mumo directory structure by name. @param names List of names of modules to load @return: List of modules loaded """ loadedmodules = {} if not names: # If no names are given load all modules that have a configuration in the cfg_dir if not os.path.isdir(self.cfg.modules.cfg_dir): msg = "Module configuration directory '%s' not found" % self.cfg.modules.cfg_dir self.log().error(msg) raise FailedLoadModuleImportException(msg) names = [] for f in os.listdir(self.cfg.modules.cfg_dir): if os.path.isfile(self.cfg.modules.cfg_dir + f): base, ext = os.path.splitext(f) if not ext or ext.lower() == ".ini" or ext.lower() == ".conf": names.append(base) for name in names: try: modinst = self._loadModule_noblock(name) loadedmodules[name] = modinst except FailedLoadModuleException: pass return loadedmodules @local_thread_blocking def loadModuleCls(self, name, modcls, module_cfg = None): return self._loadModuleCls_noblock(name, modcls, module_cfg) @debug_log(debug_me) def _loadModuleCls_noblock(self, name, modcls, module_cfg = None): log = self.log() if name in self.modules: log.error("Module '%s' already loaded", name) return modqueue = Queue.Queue() modmanager = MumoManagerRemote(self, name, modqueue) try: modinst = modcls(name, modmanager, module_cfg) except Exception, e: msg = "Module '%s' failed to initialize" % name log.error(msg) log.exception(e) raise FailedLoadModuleInitializationException(msg) # Remember it self.modules[name] = modinst self.queues[modqueue] = modinst return modinst @local_thread_blocking def loadModule(self, name): """ Loads a single module either by name @param name Name of the module to load @return Module instance """ self._loadModule_noblock(name) @debug_log(debug_me) def _loadModule_noblock(self, name): # Make sure this module is not already loaded log = self.log() log.debug("loadModuleByName('%s')", name) if name in self.modules: log.warning("Tried to load already loaded module %s", name) return # Check whether there is a configuration file for this module confpath = self.cfg.modules.cfg_dir + name + '.ini' if not os.path.isfile(confpath): msg = "Module configuration file '%s' not found" % confpath log.error(msg) raise FailedLoadModuleConfigException(msg) # Make sure the module directory is in our python path and exists if not self.cfg.modules.mod_dir in sys.path: if not os.path.isdir(self.cfg.modules.mod_dir): msg = "Module directory '%s' not found" % self.cfg.modules.mod_dir log.error(msg) raise FailedLoadModuleImportException(msg) sys.path.insert(0, self.cfg.modules.mod_dir) # Import the module and instanciate it try: mod = __import__(name) self.imports[name] = mod except ImportError, e: msg = "Failed to import module '%s', reason: %s" % (name, str(e)) log.error(msg) raise FailedLoadModuleImportException(msg) try: try: modcls = mod.mumo_module_class # First check if there's a magic mumo_module_class variable log.debug("Magic mumo_module_class found") except AttributeError: modcls = getattr(mod, name) except AttributeError: msg = "Module does not contain required class '%s'" % name log.error(msg) raise FailedLoadModuleInitializationException(msg) return self._loadModuleCls_noblock(name, modcls, confpath) @local_thread_blocking @debug_log(debug_me) def startModules(self, names = None): """ Start a module by name @param names List of names of modules to start @return A dict of started module names and instances """ log = self.log() startedmodules = {} if not names: # If no names are given start all models names = self.modules.iterkeys() for name in names: try: modinst = self.modules[name] if not modinst.isAlive(): modinst.start() log.debug("Module '%s' started", name) else: log.debug("Module '%s' already running", name) startedmodules[name] = modinst except KeyError: log.error("Could not start unknown module '%s'", name) return startedmodules @local_thread_blocking @debug_log(debug_me) def stopModules(self, names = None, force = False): """ Stop a list of modules by name. Note that this only works for well behaved modules. At this point if a module is really going rampant you will have to restart mumo. @param names List of names of modules to unload @param force Unload the module asap dropping messages queued for it @return A dict of stopped module names and instances """ log = self.log() stoppedmodules = {} if not names: # If no names are given start all models names = self.modules.iterkeys() for name in names: try: modinst = self.modules[name] stoppedmodules[name] = modinst except KeyError: log.warning("Asked to stop unknown module '%s'", name) continue if force: # We will have to drain the modules queues for queue, module in self.queues.iteritems(): if module in self.modules: try: while queue.get_nowait(): pass except Queue.Empty: pass for modinst in stoppedmodules.itervalues(): if modinst.isAlive(): modinst.stop() log.debug("Module '%s' is being stopped", name) else: log.debug("Module '%s' already stopped", name) for modinst in stoppedmodules.itervalues(): modinst.join(timeout = self.cfg.modules.timeout) return stoppedmodules def stop(self, force = True): """ Stops all modules and shuts down the manager. """ self.log().debug("Stopping") self.stopModules() Worker.stop(self, force)