bin/netsvc_socket.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                 s=_service[name]
  36                 for m in s._method:
  37                         self._meth[m]=s._method[m]
  38         def __getattr__(self, name):
  39                 return ServiceEndPointCall(self._id, self._meth[name])
  40
  41 class Service(object):
  42         _serviceEndPointID = 0
  43         def __init__(self, name, audience=''):
  44                 _service[name]=self
  45                 self.__name=name
  46                 self._method={}
  47                 self.exportedMethods=None
  48                 self._response_process=None
  49                 self._response_process_id=None
  50                 self._response=None
  51
  52         def joinGroup(self,name):
  53                 if not name in _group:
  54                         _group[name]={}
  55                 _group[name][self.__name]=self
  56
  57         def exportMethod(self, m):
  58                 if callable(m):
  59                         self._method[m.__name__]=m
  60
  61         def serviceEndPoint(self,s):
  62                 if Service._serviceEndPointID >= 2**16:
  63                         Service._serviceEndPointID = 0
  64                 Service._serviceEndPointID += 1
  65                 return ServiceEndPoint(s, self._serviceEndPointID)
  66
  67         def conversationId(self):
  68                 return 1
  69
  70         def processResponse(self,s,id):
  71                 self._response_process, self._response_process_id = s, id
  72
  73         def processFailure(self,s,id):
  74                 pass
  75
  76         def resumeResponse(self,s):
  77                 pass
  78
  79         def cancelResponse(self,s):
  80                 pass
  81
  82         def suspendResponse(self,s):
  83                 if self._response_process:
  84                         self._response_process(self._response_process_id,
  85                                                                    _res[self._response_process_id])
  86                 self._response_process=None
  87                 self._response=s(self._response_process_id)
  88
  89         def abortResponse(self,error, description, origin, details):
  90                 import tools
  91                 if not tools.config['debug_mode']:
  92                         raise Exception("%s -- %s\n\n%s"%(origin,description,details))
  93                 else:
  94                         raise
  95
  96         def currentFailure(self,s):
  97                 pass
  98
  99 class LocalService(Service):
 100         def __init__(self, name):
 101                 self.__name=name
 102                 s=_service[name]
 103                 self._service=s
 104                 for m in s._method:
 105                         setattr(self,m,s._method[m])
 106
 107 class ServiceUnavailable(Exception):
 108         pass
 109
 110 LOG_DEBUG='debug'
 111 LOG_INFO='info'
 112 LOG_WARNING='warn'
 113 LOG_ERROR='error'
 114 LOG_CRITICAL='critical'
 115
 116 def init_logger():
 117         from tools import config
 118         import os
 119
 120         if config['logfile']:
 121                 logf = config['logfile']
 122                 # test if the directories exist, else create them
 123                 try:
 124                         if not os.path.exists(os.path.dirname(logf)):
 125                                 os.makedirs(os.path.dirname(logf))
 126                         try:
 127                                 fd = open(logf, 'a')
 128                                 handler = logging.StreamHandler(fd)
 129                         except IOError:
 130                                 sys.stderr.write("ERROR: couldn't open the logfile\n")
 131                                 handler = logging.StreamHandler(sys.stdout)
 132                 except OSError:
 133                         sys.stderr.write("ERROR: couldn't create the logfile directory\n")
 134                         handler = logging.StreamHandler(sys.stdout)
 135         else:
 136                 handler = logging.StreamHandler(sys.stdout)
 137
 138         # create a format for log messages and dates
 139         formatter = logging.Formatter('%(asctime)s %(levelname)s:%(name)s:%(message)s', '%a, %d %b %Y %H:%M:%S')
 140
 141         # tell the handler to use this format
 142         handler.setFormatter(formatter)
 143
 144         # add the handler to the root logger
 145         logging.getLogger().addHandler(handler)
 146         logging.getLogger().setLevel(logging.INFO)
 147
 148
 149 class Logger(object):
 150         def notifyChannel(self,name,level,msg):
 151                 log = logging.getLogger(name)
 152                 getattr(log,level)(msg)
 153
 154 class Agent(object):
 155         _timers = []
 156         _logger = Logger()
 157
 158         def setAlarm(self, fn, dt, args=[], kwargs={}):
 159                 wait = dt - time.time()
 160                 if wait > 0:
 161                         self._logger.notifyChannel(
 162                                         'timers', LOG_DEBUG,
 163                                         "Job scheduled in %s seconds for %s.%s" % (wait,
 164                                                                                                                            fn.im_class.__name__,
 165                                                                                                 fn.func_name))
 166                         timer = threading.Timer(wait, fn, args, kwargs)
 167                         timer.start()
 168                         self._timers.append(timer)
 169                 for timer in self._timers[:]:
 170                         if not timer.isAlive():
 171                                 self._timers.remove(timer)
 172
 173         def quit(cls):
 174                 for timer in cls._timers:
 175                         timer.cancel()
 176         quit=classmethod(quit)
 177
 178 class RpcGateway(object):
 179         def __init__(self, name):
 180                 self.name=name
 181
 182 class Dispatcher(object):
 183         def __init__(self):
 184                 pass
 185         def monitor(self,signal):
 186                 pass
 187         def run(self):
 188                 pass
 189
 190 class xmlrpc(object):
 191         class RpcGateway(object):
 192                 def __init__(self, name):
 193                         self.name=name
 194
 195 class GenericXMLRPCRequestHandler:
 196         def _dispatch(self, method, params):
 197                 import traceback
 198                 try:
 199                         n=self.path.split("/")[-1]
 200 #                       print "TERP-CALLING:",n,method,params
 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 #                               print "RESPONSE FOUND"
 208                                 r=res
 209 #                       print "TERP-RETURN :",r
 210                         return r
 211                 except Exception,e:
 212                         print "Exception in call:"
 213                         print '-'*60
 214                         traceback.print_exc(file=sys.stdout)
 215                         print '-'*60
 216                         s=str(e)
 217                         import tools
 218                         if tools.config['debug_mode']:
 219                                 import pdb
 220                                 tb = sys.exc_info()[2]
 221                                 pdb.post_mortem(tb)
 222                         raise xmlrpclib.Fault(1,s)
 223
 224 import tiny_socket
 225
 226 class HttpDaemon(object):
 227         def __init__(self, interface,port, secure=False):
 228                 self.__port=port
 229                 self.__interface=interface
 230                 self.server = socket.socket( socket.AF_INET, socket.SOCK_STREAM)
 231                 self.server.bind((socket.gethostname(), 8085))
 232                 self.server.listen(5)
 233
 234         def attach(self,path,gw):
 235                 pass
 236
 237         def handler(self,signum, frame):
 238                 from tools import config
 239                 self.server.close()
 240                 self.server.close()
 241                 Agent.quit()
 242                 if config['pidfile']:
 243                         os.unlink(config['pidfile'])
 244                 del self.server
 245                 sys.exit(0)
 246
 247         def start(self):
 248                 signal.signal(signal.SIGINT, self.handler)
 249                 signal.signal(signal.SIGTERM, self.handler)
 250                 class client_thread(threading.Thread):
 251                         def run(self, sock):
 252                                 ts = tiny_socket.mysocket(sock)
 253                                 msg = ts.myreceive()
 254
 255                                 s=LocalService(msg[0])
 256                                 m=getattr(s,msg[1])
 257                                 s._service._response=None
 258                                 r=m(*msg[2:])
 259                                 res=s._service._response
 260                                 if res!=None:
 261                                         r=res
 262                                 result = r
 263
 264                                 ts.mysend(result)
 265
 266                                 return msg
 267
 268                 while 1:
 269                         #accept connections from outside
 270                         (clientsocket, address) = self.server.accept()
 271                         #now do something with the clientsocket
 272                         #in this case, we'll pretend this is a threaded server
 273                         ct = client_thread()
 274                         ct.run(clientsocket)
 275