Source Code for Module SpecClient.SpecServer

  1  import asyncore 
  2  import socket 
  3  import types 
  4   
  5  import SpecConnection 
  6  import SpecMessage 
  7   
  8   
  9 -class BaseSpecRequestHandler(asyncore.dispatcher): 
 10 -    def __init__(self, request, client_address, server): 
 11          asyncore.dispatcher.__init__(self, request) 
 12   
 13          self.client_address = client_address 
 14          self.server = server 
 15          self.sendq = [] 
 16          self.receivedStrings = [] 
 17          self.outputStrings = [] 
 18          self.message = None 
 19          self.clientVersion = None 
 20          self.clientOrder = "" 
 21   
 22   
 23 -    def handle_read(self): 
 24          self.receivedStrings.append(self.recv(32768)) 
 25          s = ''.join(self.receivedStrings) 
 26          sbuffer = buffer(s) 
 27          consumedBytes = 0 
 28          offset = 0 
 29   
 30          while offset < len(sbuffer): 
 31              if self.message is None: 
 32                  self.message = SpecMessage.message(version = self.clientVersion, order=self.clientOrder) 
 33   
 34              consumedBytes = self.message.readFromStream(sbuffer[offset:]) 
 35   
 36              if consumedBytes == 0: 
 37                  break 
 38   
 39              offset += consumedBytes 
 40   
 41              if self.message.isComplete(): 
 42                  # dispatch incoming message 
 43                  if self.message.cmd == SpecMessage.HELLO: 
 44                      self.clientOrder = self.message.packedHeaderDataFormat[0] 
 45                      print "client byte order: ", self.clientOrder 
 46                      self.clientVersion = self.message.vers 
 47                      self.send_hello_reply(self.message.sn, str(self.server.name)) 
 48                  else: 
 49                      if not self.dispatchIncomingMessage(self.message): 
 50                          self.send_error(self.message.sn, '', 'unsupported command type : %d' % self.message.cmd) 
 51   
 52                  self.message = None 
 53   
 54          self.receivedStrings = [ s[offset:] ] 
 55   
 56   
 57 -    def writable(self): 
 58          return len(self.sendq) > 0 or sum(map(len, self.outputStrings)) > 0 
 59   
 60   
 61 -    def handle_write(self): 
 62          # 
 63          # send all the messages from the queue 
 64          # 
 65          while len(self.sendq) > 0: 
 66              self.outputStrings.append(self.sendq.pop().sendingString()) 
 67   
 68          outputBuffer = ''.join(self.outputStrings) 
 69   
 70          sent = self.send(outputBuffer) 
 71          self.outputStrings = [ outputBuffer[sent:] ] 
 72   
 73   
 74 -    def handle_close(self): 
 75          self.close() 
 76          self.server.clients.remove(self) 
 77   
 78   
 79 -    def dispatchIncomingMessage(self, message): 
 80          pass 
 81   
 82   
 83 -    def parseCommandString(self, cmdstr): 
 84          if SpecMessage.NULL in cmdstr: 
 85              cmdparts = cmdstr.split(SpecMessage.NULL) 
 86              command = cmdparts[0] 
 87              args = tuple([ eval(cmdpart) for cmdpart in cmdparts[1:] ]) 
 88              return command, args 
 89   
 90          cmdpartLength = cmdstr.find('(') 
 91   
 92          if cmdpartLength < 0: 
 93              return cmdstr, () 
 94   
 95          try: 
 96              command = cmdstr[:cmdpartLength] 
 97              args = eval(cmdstr[cmdpartLength:]) 
 98          except: 
 99              print 'error parsing command string %s' % cmdstr 
100              return '', () 
101          else: 
102              if not type(args) == types.TupleType: 
103                  args = (args, ) 
104   
105              return command, args 
106   
107   
108 -    def executeCommandAndReply(self, replyID = None, cmd = '', *args): 
109          if len(cmd) == 0 or replyID is None: 
110              return 
111   
112          if len(args) == 0: 
113              cmdstr = str(cmd) 
114              command, args = self.parseCommandString(cmdstr) 
115          else: 
116              command = cmd 
117   
118   
119          func = None 
120   
121          if hasattr(self, command): 
122              func = getattr(self, command) 
123          elif hasattr(self.server, command): 
124              func = getattr(self.server, command) 
125          else: 
126              self.send_error(replyID, '', '"' + command + '" command does not exist.') 
127              return 
128   
129          print 'executeCommandAndReply: func=%s, args=%s' % (repr(func), args) 
130   
131          if callable(func): 
132              try: 
133                  ret = func(*args) 
134              except: 
135                  import traceback 
136                  traceback.print_exc() 
137   
138                  self.send_error(replyID, '', 'Failed to execute command "' + command) 
139              else: 
140                  if ret is None: 
141                      self.send_error(replyID, '', command + ' returned None.') 
142                  else: 
143                      self.send_reply(replyID, '', ret) 
144          else: 
145              self.send_error(replyID, '',  command + ' is not callable on server.') 
146   
147   
148 -    def send_hello_reply(self, replyID, serverName): 
149          self.sendq.append(SpecMessage.msg_hello_reply(replyID, serverName, version = self.clientVersion, order=self.clientOrder)) 
150   
151   
152 -    def send_reply(self, replyID, name, data): 
153          self.sendq.append(SpecMessage.reply_message(replyID, name, data, version = self.clientVersion, order=self.clientOrder)) 
154   
155   
156 -    def send_error(self, replyID, name, data): 
157          self.sendq.append(SpecMessage.error_message(replyID, name, data, version = self.clientVersion, order=self.clientOrder)) 
158   
159   
160 -    def send_msg_event(self, chanName, value): 
161          self.sendq.append(SpecMessage.msg_event(chanName, value, version = self.clientVersion, order=self.clientOrder)) 
162   
163   
164 -class SpecServer(asyncore.dispatcher): 
165 -    def __init__(self, host, name, handler = BaseSpecRequestHandler): 
166          asyncore.dispatcher.__init__(self) 
167   
168          self.name = name 
169          self.RequestHandlerClass = handler 
170          self.clients = [] 
171   
172          self.create_socket(socket.AF_INET, socket.SOCK_STREAM) 
173   
174          self.set_reuse_addr() 
175   
176          if type(name) == type(''): 
177              for p in range(SpecConnection.MIN_PORT, SpecConnection.MAX_PORT): 
178                  self.server_address = ( host, p ) 
179   
180                  try: 
181                      self.bind(self.server_address) 
182                  except: 
183                      continue 
184                  else: 
185                      break 
186          else: 
187              self.server_address = (host, name) 
188              self.bind(self.server_address) 
189   
190          #print self.server_address 
191          self.listen(5) 
192   
193   
194 -    def handle_accept(self): 
195          try: 
196              conn, addr = self.accept() 
197          except: 
198              return 
199          else: 
200              conn.setblocking(0) 
201              self.clients.append(self.RequestHandlerClass(conn, addr, self)) 
202   
203   
204 -    def serve_forever(self): 
205          asyncore.loop() 
206