bin/netsvc.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                         logger = Logger()
 214                         logger.notifyChannel("web-services", LOG_ERROR, 'Exception in call: ' + reduce(lambda x, y: x+y, traceback.format_exc()))
 215                         s=str(e)
 216                         import tools
 217                         if tools.config['debug_mode']:
 218                                 import pdb
 219                                 tb = sys.exc_info()[2]
 220                                 pdb.post_mortem(tb)
 221                         raise xmlrpclib.Fault(1,s)
 222
 223 class SimpleXMLRPCRequestHandler(GenericXMLRPCRequestHandler, SimpleXMLRPCServer.SimpleXMLRPCRequestHandler):
 224         pass
 225
 226 if HAS_SSL:
 227         class SecureXMLRPCRequestHandler(GenericXMLRPCRequestHandler, SecureXMLRPCServer.SecureXMLRPCRequestHandler):
 228                 pass
 229 else:
 230         pass
 231
 232 class SimpleThreadedXMLRPCServer(SocketServer.ThreadingMixIn, SimpleXMLRPCServer.SimpleXMLRPCServer):
 233         def server_bind(self):
 234                 self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
 235                 SimpleXMLRPCServer.SimpleXMLRPCServer.server_bind(self)
 236
 237 if HAS_SSL:
 238         class SecureThreadedXMLRPCServer(SocketServer.ThreadingMixIn, SecureXMLRPCServer.SecureXMLRPCServer):
 239                 def server_bind(self):
 240                         self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
 241                         SecureXMLRPCServer.SecureXMLRPCServer.server_bind(self)
 242 else:
 243         pass
 244
 245 class HttpDaemon(threading.Thread):
 246         def __init__(self, interface,port, secure=False):
 247                 threading.Thread.__init__(self)
 248                 self.__port=port
 249                 self.__interface=interface
 250                 if secure and HAS_SSL:
 251                         self.server = SecureThreadedXMLRPCServer((interface, port), SecureXMLRPCRequestHandler,0)
 252                 else:
 253                         self.server = SimpleThreadedXMLRPCServer((interface, port), SimpleXMLRPCRequestHandler,0)
 254
 255         def attach(self,path,gw):
 256                 pass
 257
 258         def stop(self):
 259                 self.running = False
 260                 if hasattr(socket, 'SHUT_RDWR'):
 261                         self.server.socket.shutdown(socket.SHUT_RDWR)
 262                 else:
 263                         self.server.socket.shutdown(2)
 264                 self.server.socket.close()
 265 #               self.server.socket.close()
 266 #               del self.server
 267
 268         def run(self):
 269                 self.server.register_introspection_functions()
 270
 271 #               self.server.serve_forever()
 272                 self.running = True
 273                 while self.running:
 274                         self.server.handle_request()
 275                 return True
 276
 277                 # If the server need to be run recursively
 278                 #
 279                 #signal.signal(signal.SIGALRM, self.my_handler)
 280                 #signal.alarm(6)
 281                 #while True:
 282                 #       self.server.handle_request()
 283                 #signal.alarm(0)          # Disable the alarm
 284
 285 import tiny_socket
 286 class TinySocketClientThread(threading.Thread):
 287         def __init__(self, sock, threads):
 288                 threading.Thread.__init__(self)
 289                 self.sock = sock
 290                 self.threads = threads
 291
 292         def run(self):
 293                 import traceback
 294                 import time
 295                 import select
 296                 try:
 297                         self.running = True
 298                         ts = tiny_socket.mysocket(self.sock)
 299                         while self.running:
 300                                 msg = ts.myreceive()
 301
 302                                 try:
 303                                         s=LocalService(msg[0])
 304                                         m=getattr(s,msg[1])
 305                                         s._service._response=None
 306                                         r=m(*msg[2:])
 307                                         res=s._service._response
 308                                         if res!=None:
 309                                                 r=res
 310                                         result = r
 311                                         ts.mysend(result)
 312                                 except Exception, e:
 313                                         logger = Logger()
 314                                         logger.notifyChannel("web-services", LOG_ERROR, 'Exception in call: ' + reduce(lambda x, y: x+y, traceback.format_exc()))
 315                                         s=str(e)
 316                                         import tools
 317                                         if tools.config['debug_mode']:
 318                                                 import pdb
 319                                                 tb = sys.exc_info()[2]
 320                                                 pdb.post_mortem(tb)
 321                                         ts.mysend(e, exception=True)
 322                                 self.sock.close()
 323                                 self.threads.remove(self)
 324                                 return True
 325                 except Exception, e:
 326                         self.sock.close()
 327                         return False
 328         def stop(self):
 329                 self.running = False
 330 #               self.sock.shutdown(socket.SHUT_RDWR)
 331 #               self.sock.close()
 332
 333 class TinySocketServerThread(threading.Thread):
 334         def __init__(self, interface, port, secure=False):
 335                 threading.Thread.__init__(self)
 336                 self.__port=port
 337                 self.__interface=interface
 338                 self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
 339                 self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
 340                 self.socket.bind((self.__interface, self.__port))
 341                 self.socket.listen(5)
 342                 self.threads = []
 343
 344         def run(self):
 345                 import select
 346                 try:
 347                         self.running = True
 348                         while self.running:
 349                                 #accept connections from outside
 350                                 (clientsocket, address) = self.socket.accept()
 351                                 #now do something with the clientsocket
 352                                 #in this case, we'll pretend this is a threaded server
 353                                 ct = TinySocketClientThread(clientsocket, self.threads)
 354                                 ct.start()
 355                                 self.threads.append(ct)
 356 #                               print "threads size:", len(self.threads)
 357                         self.socket.close()
 358                 except Exception, e:
 359                         self.socket.close()
 360                         return False
 361
 362         def stop(self):
 363                 self.running=False
 364                 for t in self.threads:
 365                         t.stop()
 366                 try:
 367                         if hasattr(socket, 'SHUT_RDWR'):
 368                                 self.socket.shutdown(socket.SHUT_RDWR)
 369                         else:
 370                                 self.socket.shutdown(2)
 371                         self.socket.close()
 372                 except:
 373                         return False
 374
 375 # vim:noexpandtab:
 376
 377