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