bin/netsvc_xmlrpc.py

   1 #!/usr/bin/python
   2
   3 import time
   4 import threading
   5
   6 import SimpleXMLRPCServer,signal,sys,xmlrpclib
   7 import SocketServer
   8 import socket
   9 import logging
  10 import os
  11
  12 try:
  13         from ssl import *
  14         HAS_SSL = True
  15 except ImportError:
  16         HAS_SSL = False
  17
  18 _service={}
  19 _group={}
  20 _res_id=1
  21 _res={}
  22
  23 class ServiceEndPointCall(object):
  24         def __init__(self,id,method):
  25                 self._id=id
  26                 self._meth=method
  27         def __call__(self,*args):
  28                 _res[self._id]=self._meth(*args)
  29                 return self._id
  30
  31 class ServiceEndPoint(object):
  32         def __init__(self, name, id):
  33                 self._id = id
  34                 self._meth={}
  35                 print _service
  36                 s=_service[name]
  37                 for m in s._method:
  38                         self._meth[m]=s._method[m]
  39         def __getattr__(self, name):
  40                 return ServiceEndPointCall(self._id, self._meth[name])
  41
  42 class Service(object):
  43         _serviceEndPointID = 0
  44         def __init__(self, name, audience=''):
  45                 _service[name]=self
  46                 self.__name=name
  47                 self._method={}
  48                 self.exportedMethods=None
  49                 self._response_process=None
  50                 self._response_process_id=None
  51                 self._response=None
  52
  53         def joinGroup(self,name):
  54                 if not name in _group:
  55                         _group[name]={}
  56                 _group[name][self.__name]=self
  57
  58         def exportMethod(self, m):
  59                 if callable(m):
  60                         self._method[m.__name__]=m
  61
  62         def serviceEndPoint(self,s):
  63                 if Service._serviceEndPointID >= 2**16:
  64                         Service._serviceEndPointID = 0
  65                 Service._serviceEndPointID += 1
  66                 return ServiceEndPoint(s, self._serviceEndPointID)
  67
  68         def conversationId(self):
  69                 return 1
  70
  71         def processResponse(self,s,id):
  72                 self._response_process, self._response_process_id = s, id
  73
  74         def processFailure(self,s,id):
  75                 pass
  76
  77         def resumeResponse(self,s):
  78                 pass
  79
  80         def cancelResponse(self,s):
  81                 pass
  82
  83         def suspendResponse(self,s):
  84                 if self._response_process:
  85                         self._response_process(self._response_process_id,
  86                                                                    _res[self._response_process_id])
  87                 self._response_process=None
  88                 self._response=s(self._response_process_id)
  89
  90         def abortResponse(self,error, description, origin, details):
  91                 import tools
  92                 if not tools.config['debug_mode']:
  93                         raise Exception("%s -- %s\n\n%s"%(origin,description,details))
  94                 else:
  95                         raise
  96
  97         def currentFailure(self,s):
  98                 pass
  99
 100 class LocalService(Service):
 101         def __init__(self, name):
 102                 self.__name=name
 103                 s=_service[name]
 104                 self._service=s
 105                 for m in s._method:
 106                         setattr(self,m,s._method[m])
 107
 108 class ServiceUnavailable(Exception):
 109         pass
 110
 111 LOG_DEBUG='debug'
 112 LOG_INFO='info'
 113 LOG_WARNING='warn'
 114 LOG_ERROR='error'
 115 LOG_CRITICAL='critical'
 116
 117 def init_logger():
 118         from tools import config
 119         import os
 120
 121         if config['logfile']:
 122                 logf = config['logfile']
 123                 # test if the directories exist, else create them
 124                 try:
 125                         if not os.path.exists(os.path.dirname(logf)):
 126                                 os.makedirs(os.path.dirname(logf))
 127                         try:
 128                                 fd = open(logf, 'a')
 129                                 handler = logging.StreamHandler(fd)
 130                         except IOError:
 131                                 sys.stderr.write("ERROR: couldn't open the logfile\n")
 132                                 handler = logging.StreamHandler(sys.stdout)
 133                 except OSError:
 134                         sys.stderr.write("ERROR: couldn't create the logfile directory\n")
 135                         handler = logging.StreamHandler(sys.stdout)
 136         else:
 137                 handler = logging.StreamHandler(sys.stdout)
 138
 139         # create a format for log messages and dates
 140         formatter = logging.Formatter('%(asctime)s %(levelname)s:%(name)s:%(message)s', '%a, %d %b %Y %H:%M:%S')
 141
 142         # tell the handler to use this format
 143         handler.setFormatter(formatter)
 144
 145         # add the handler to the root logger
 146         logging.getLogger().addHandler(handler)
 147         logging.getLogger().setLevel(logging.INFO)
 148
 149
 150 class Logger(object):
 151         def notifyChannel(self,name,level,msg):
 152                 log = logging.getLogger(name)
 153                 getattr(log,level)(msg)
 154
 155 class Agent(object):
 156         _timers = []
 157         _logger = Logger()
 158
 159         def setAlarm(self, fn, dt, args=[], kwargs={}):
 160                 wait = dt - time.time()
 161                 if wait > 0:
 162                         self._logger.notifyChannel(
 163                                         'timers', LOG_DEBUG,
 164                                         "Job scheduled in %s seconds for %s.%s" % (wait,
 165                                                                                                                            fn.im_class.__name__,
 166                                                                                                 fn.func_name))
 167                         timer = threading.Timer(wait, fn, args, kwargs)
 168                         timer.start()
 169                         self._timers.append(timer)
 170                 for timer in self._timers[:]:
 171                         if not timer.isAlive():
 172                                 self._timers.remove(timer)
 173
 174         def quit(cls):
 175                 for timer in cls._timers:
 176                         timer.cancel()
 177         quit=classmethod(quit)
 178
 179 class RpcGateway(object):
 180         def __init__(self, name):
 181                 self.name=name
 182
 183 class Dispatcher(object):
 184         def __init__(self):
 185                 pass
 186         def monitor(self,signal):
 187                 pass
 188         def run(self):
 189                 pass
 190
 191 class xmlrpc(object):
 192         class RpcGateway(object):
 193                 def __init__(self, name):
 194                         self.name=name
 195
 196 class GenericXMLRPCRequestHandler:
 197         def _dispatch(self, method, params):
 198                 import traceback
 199                 try:
 200                         n=self.path.split("/")[-1]
 201 #                       print "TERP-CALLING:",n,method,params
 202                         s=LocalService(n)
 203                         m=getattr(s,method)
 204                         s._service._response=None
 205                         r=m(*params)
 206                         res=s._service._response
 207                         if res!=None:
 208 #                               print "RESPONSE FOUND"
 209                                 r=res
 210 #                       print "TERP-RETURN :",r
 211                         return r
 212                 except Exception,e:
 213                         print "Exception in call:"
 214                         print '-'*60
 215                         traceback.print_exc(file=sys.stdout)
 216                         print '-'*60
 217                         s=str(e)
 218                         import tools
 219                         if tools.config['debug_mode']:
 220                                 import pdb
 221                                 tb = sys.exc_info()[2]
 222                                 pdb.post_mortem(tb)
 223                         raise xmlrpclib.Fault(1,s)
 224
 225 class SimpleXMLRPCRequestHandler(GenericXMLRPCRequestHandler, SimpleXMLRPCServer.SimpleXMLRPCRequestHandler):
 226         pass
 227
 228 if HAS_SSL:
 229         class SecureXMLRPCRequestHandler(GenericXMLRPCRequestHandler, SecureXMLRPCServer.SecureXMLRPCRequestHandler):
 230                 pass
 231 else:
 232         pass
 233
 234 class SimpleThreadedXMLRPCServer(SocketServer.ThreadingMixIn, SimpleXMLRPCServer.SimpleXMLRPCServer):
 235         def server_bind(self):
 236                 self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
 237                 SimpleXMLRPCServer.SimpleXMLRPCServer.server_bind(self)
 238
 239 if HAS_SSL:
 240         class SecureThreadedXMLRPCServer(SocketServer.ThreadingMixIn, SecureXMLRPCServer.SecureXMLRPCServer):
 241                 def server_bind(self):
 242                         self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
 243                         SecureXMLRPCServer.SecureXMLRPCServer.server_bind(self)
 244 else:
 245         pass
 246
 247 class HttpDaemon(object):
 248         def __init__(self, interface,port, secure=False):
 249                 self.__port=port
 250                 self.__interface=interface
 251                 if secure and HAS_SSL:
 252                         self.server = SecureThreadedXMLRPCServer((interface, port), SecureXMLRPCRequestHandler,0)
 253                 else:
 254                         self.server = SimpleThreadedXMLRPCServer((interface, port), SimpleXMLRPCRequestHandler,0)
 255
 256         def attach(self,path,gw):
 257                 pass
 258
 259         def handler(self,signum, frame):
 260                 from tools import config
 261                 self.server.socket.close()
 262                 self.server.socket.close()
 263                 Agent.quit()
 264                 if config['pidfile']:
 265                         os.unlink(config['pidfile'])
 266                 del self.server
 267                 sys.exit(0)
 268
 269         def start(self):
 270                 from tools import config
 271                 if config['pidfile']:
 272                         fd=open(config['pidfile'], 'w')
 273                         pidtext="%d" % (os.getpid())
 274                         fd.write(pidtext)
 275                         fd.close()
 276                 signal.signal(signal.SIGINT, self.handler)
 277                 signal.signal(signal.SIGTERM, self.handler)
 278                 self.server.register_introspection_functions()
 279
 280                 self.server.serve_forever()
 281
 282                 # If the server need to be run recursively
 283                 #
 284                 #signal.signal(signal.SIGALRM, self.my_handler)
 285                 #signal.alarm(6)
 286                 #while True:
 287                 #       self.server.handle_request()
 288                 #signal.alarm(0)          # Disable the alarm
 289
 290 # vim:noexpandtab:
 291
 292