asynchat.py 11 KB

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