1 # -*- Mode: Python; tab-width: 4 -*- 2 # Id: asynchat.py,v 2.26 2000/09/07 22:29:26 rushing Exp 3 # Author: Sam Rushing <rushing (at] nightmare.com> 4 5 # ====================================================================== 6 # Copyright 1996 by Sam Rushing 7 # 8 # All Rights Reserved 9 # 10 # Permission to use, copy, modify, and distribute this software and 11 # its documentation for any purpose and without fee is hereby 12 # granted, provided that the above copyright notice appear in all 13 # copies and that both that copyright notice and this permission 14 # notice appear in supporting documentation, and that the name of Sam 15 # Rushing not be used in advertising or publicity pertaining to 16 # distribution of the software without specific, written prior 17 # permission. 18 # 19 # SAM RUSHING DISCLAIMS ALL WARRANTIES WITH REGARD TO THIS SOFTWARE, 20 # INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS, IN 21 # NO EVENT SHALL SAM RUSHING BE LIABLE FOR ANY SPECIAL, INDIRECT OR 22 # CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM LOSS 23 # OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, 24 # NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF OR IN 25 # CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE. 26 # ====================================================================== 27 28 r"""A class supporting chat-style (command/response) protocols. 29 30 This class adds support for 'chat' style protocols - where one side 31 sends a 'command', and the other sends a response (examples would be 32 the common internet protocols - smtp, nntp, ftp, etc..). 33 34 The handle_read() method looks at the input stream for the current 35 'terminator' (usually '\r\n' for single-line responses, '\r\n.\r\n' 36 for multi-line output), calling self.found_terminator() on its 37 receipt. 38 39 for example: 40 Say you build an async nntp client using this class. At the start 41 of the connection, you'll have self.terminator set to '\r\n', in 42 order to process the single-line greeting. Just before issuing a 43 'LIST' command you'll set it to '\r\n.\r\n'. The output of the LIST 44 command will be accumulated (using your own 'collect_incoming_data' 45 method) up to the terminator, and then control will be returned to 46 you - by calling your self.found_terminator() method. 47 """ 48 49 import socket 50 import asyncore 51 from collections import deque 52 from sys import py3kwarning 53 from warnings import filterwarnings, catch_warnings 54 55 class async_chat (asyncore.dispatcher): 56 """This is an abstract class. You must derive from this class, and add 57 the two methods collect_incoming_data() and found_terminator()""" 58 59 # these are overridable defaults 60 61 ac_in_buffer_size = 4096 62 ac_out_buffer_size = 4096 63 64 def __init__ (self, sock=None, map=None): 65 # for string terminator matching 66 self.ac_in_buffer = '' 67 68 # we use a list here rather than cStringIO for a few reasons... 69 # del lst[:] is faster than sio.truncate(0) 70 # lst = [] is faster than sio.truncate(0) 71 # cStringIO will be gaining unicode support in py3k, which 72 # will negatively affect the performance of bytes compared to 73 # a ''.join() equivalent 74 self.incoming = [] 75 76 # we toss the use of the "simple producer" and replace it with 77 # a pure deque, which the original fifo was a wrapping of 78 self.producer_fifo = deque() 79 asyncore.dispatcher.__init__ (self, sock, map) 80 81 def collect_incoming_data(self, data): 82 raise NotImplementedError("must be implemented in subclass") 83 84 def _collect_incoming_data(self, data): 85 self.incoming.append(data) 86 87 def _get_data(self): 88 d = ''.join(self.incoming) 89 del self.incoming[:] 90 return d 91 92 def found_terminator(self): 93 raise NotImplementedError("must be implemented in subclass") 94 95 def set_terminator (self, term): 96 "Set the input delimiter. Can be a fixed string of any length, an integer, or None" 97 self.terminator = term 98 99 def get_terminator (self): 100 return self.terminator 101 102 # grab some more data from the socket, 103 # throw it to the collector method, 104 # check for the terminator, 105 # if found, transition to the next state. 106 107 def handle_read (self): 108 109 try: 110 data = self.recv (self.ac_in_buffer_size) 111 except socket.error, why: 112 self.handle_error() 113 return 114 115 self.ac_in_buffer = self.ac_in_buffer + data 116 117 # Continue to search for self.terminator in self.ac_in_buffer, 118 # while calling self.collect_incoming_data. The while loop 119 # is necessary because we might read several data+terminator 120 # combos with a single recv(4096). 121 122 while self.ac_in_buffer: 123 lb = len(self.ac_in_buffer) 124 terminator = self.get_terminator() 125 if not terminator: 126 # no terminator, collect it all 127 self.collect_incoming_data (self.ac_in_buffer) 128 self.ac_in_buffer = '' 129 elif isinstance(terminator, int) or isinstance(terminator, long): 130 # numeric terminator 131 n = terminator 132 if lb < n: 133 self.collect_incoming_data (self.ac_in_buffer) 134 self.ac_in_buffer = '' 135 self.terminator = self.terminator - lb 136 else: 137 self.collect_incoming_data (self.ac_in_buffer[:n]) 138 self.ac_in_buffer = self.ac_in_buffer[n:] 139 self.terminator = 0 140 self.found_terminator() 141 else: 142 # 3 cases: 143 # 1) end of buffer matches terminator exactly: 144 # collect data, transition 145 # 2) end of buffer matches some prefix: 146 # collect data to the prefix 147 # 3) end of buffer does not match any prefix: 148 # collect data 149 terminator_len = len(terminator) 150 index = self.ac_in_buffer.find(terminator) 151 if index != -1: 152 # we found the terminator 153 if index > 0: 154 # don't bother reporting the empty string (source of subtle bugs) 155 self.collect_incoming_data (self.ac_in_buffer[:index]) 156 self.ac_in_buffer = self.ac_in_buffer[index+terminator_len:] 157 # This does the Right Thing if the terminator is changed here. 158 self.found_terminator() 159 else: 160 # check for a prefix of the terminator 161 index = find_prefix_at_end (self.ac_in_buffer, terminator) 162 if index: 163 if index != lb: 164 # we found a prefix, collect up to the prefix 165 self.collect_incoming_data (self.ac_in_buffer[:-index]) 166 self.ac_in_buffer = self.ac_in_buffer[-index:] 167 break 168 else: 169 # no prefix, collect it all 170 self.collect_incoming_data (self.ac_in_buffer) 171 self.ac_in_buffer = '' 172 173 def handle_write (self): 174 self.initiate_send() 175 176 def handle_close (self): 177 self.close() 178 179 def push (self, data): 180 sabs = self.ac_out_buffer_size 181 if len(data) > sabs: 182 for i in xrange(0, len(data), sabs): 183 self.producer_fifo.append(data[i:i+sabs]) 184 else: 185 self.producer_fifo.append(data) 186 self.initiate_send() 187 188 def push_with_producer (self, producer): 189 self.producer_fifo.append(producer) 190 self.initiate_send() 191 192 def readable (self): 193 "predicate for inclusion in the readable for select()" 194 # cannot use the old predicate, it violates the claim of the 195 # set_terminator method. 196 197 # return (len(self.ac_in_buffer) <= self.ac_in_buffer_size) 198 return 1 199 200 def writable (self): 201 "predicate for inclusion in the writable for select()" 202 return self.producer_fifo or (not self.connected) 203 204 def close_when_done (self): 205 "automatically close this channel once the outgoing queue is empty" 206 self.producer_fifo.append(None) 207 208 def initiate_send(self): 209 while self.producer_fifo and self.connected: 210 first = self.producer_fifo[0] 211 # handle empty string/buffer or None entry 212 if not first: 213 del self.producer_fifo[0] 214 if first is None: 215 self.handle_close() 216 return 217 218 # handle classic producer behavior 219 obs = self.ac_out_buffer_size 220 try: 221 with catch_warnings(): 222 if py3kwarning: 223 filterwarnings("ignore", ".*buffer", DeprecationWarning) 224 data = buffer(first, 0, obs) 225 except TypeError: 226 data = first.more() 227 if data: 228 self.producer_fifo.appendleft(data) 229 else: 230 del self.producer_fifo[0] 231 continue 232 233 # send the data 234 try: 235 num_sent = self.send(data) 236 except socket.error: 237 self.handle_error() 238 return 239 240 if num_sent: 241 if num_sent < len(data) or obs < len(first): 242 self.producer_fifo[0] = first[num_sent:] 243 else: 244 del self.producer_fifo[0] 245 # we tried to send some actual data 246 return 247 248 def discard_buffers (self): 249 # Emergencies only! 250 self.ac_in_buffer = '' 251 del self.incoming[:] 252 self.producer_fifo.clear() 253 254 class simple_producer: 255 256 def __init__ (self, data, buffer_size=512): 257 self.data = data 258 self.buffer_size = buffer_size 259 260 def more (self): 261 if len (self.data) > self.buffer_size: 262 result = self.data[:self.buffer_size] 263 self.data = self.data[self.buffer_size:] 264 return result 265 else: 266 result = self.data 267 self.data = '' 268 return result 269 270 class fifo: 271 def __init__ (self, list=None): 272 if not list: 273 self.list = deque() 274 else: 275 self.list = deque(list) 276 277 def __len__ (self): 278 return len(self.list) 279 280 def is_empty (self): 281 return not self.list 282 283 def first (self): 284 return self.list[0] 285 286 def push (self, data): 287 self.list.append(data) 288 289 def pop (self): 290 if self.list: 291 return (1, self.list.popleft()) 292 else: 293 return (0, None) 294 295 # Given 'haystack', see if any prefix of 'needle' is at its end. This 296 # assumes an exact match has already been checked. Return the number of 297 # characters matched. 298 # for example: 299 # f_p_a_e ("qwerty\r", "\r\n") => 1 300 # f_p_a_e ("qwertydkjf", "\r\n") => 0 301 # f_p_a_e ("qwerty\r\n", "\r\n") => <undefined> 302 303 # this could maybe be made faster with a computed regex? 304 # [answer: no; circa Python-2.0, Jan 2001] 305 # new python: 28961/s 306 # old python: 18307/s 307 # re: 12820/s 308 # regex: 14035/s 309 310 def find_prefix_at_end (haystack, needle): 311 l = len(needle) - 1 312 while l and not haystack.endswith(needle[:l]): 313 l -= 1 314 return l 315
