| # -*- Mode: Python; tab-width: 4 -*- |
| # Id: asynchat.py,v 2.26 2000/09/07 22:29:26 rushing Exp |
| # Author: Sam Rushing <rushing@nightmare.com> |
| |
| # ====================================================================== |
| # Copyright 1996 by Sam Rushing |
| # |
| # All Rights Reserved |
| # |
| # Permission to use, copy, modify, and distribute this software and |
| # its documentation for any purpose and without fee is hereby |
| # granted, provided that the above copyright notice appear in all |
| # copies and that both that copyright notice and this permission |
| # notice appear in supporting documentation, and that the name of Sam |
| # Rushing not be used in advertising or publicity pertaining to |
| # distribution of the software without specific, written prior |
| # permission. |
| # |
| # SAM RUSHING DISCLAIMS ALL WARRANTIES WITH REGARD TO THIS SOFTWARE, |
| # INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS, IN |
| # NO EVENT SHALL SAM RUSHING BE LIABLE FOR ANY SPECIAL, INDIRECT OR |
| # CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM LOSS |
| # OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, |
| # NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF OR IN |
| # CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE. |
| # ====================================================================== |
| |
| r"""A class supporting chat-style (command/response) protocols. |
| |
| This class adds support for 'chat' style protocols - where one side |
| sends a 'command', and the other sends a response (examples would be |
| the common internet protocols - smtp, nntp, ftp, etc..). |
| |
| The handle_read() method looks at the input stream for the current |
| 'terminator' (usually '\r\n' for single-line responses, '\r\n.\r\n' |
| for multi-line output), calling self.found_terminator() on its |
| receipt. |
| |
| for example: |
| Say you build an async nntp client using this class. At the start |
| of the connection, you'll have self.terminator set to '\r\n', in |
| order to process the single-line greeting. Just before issuing a |
| 'LIST' command you'll set it to '\r\n.\r\n'. The output of the LIST |
| command will be accumulated (using your own 'collect_incoming_data' |
| method) up to the terminator, and then control will be returned to |
| you - by calling your self.found_terminator() method. |
| """ |
| import asyncore |
| from collections import deque |
| |
| |
| class async_chat(asyncore.dispatcher): |
| """This is an abstract class. You must derive from this class, and add |
| the two methods collect_incoming_data() and found_terminator()""" |
| |
| # these are overridable defaults |
| |
| ac_in_buffer_size = 65536 |
| ac_out_buffer_size = 65536 |
| |
| # we don't want to enable the use of encoding by default, because that is a |
| # sign of an application bug that we don't want to pass silently |
| |
| use_encoding = 0 |
| encoding = 'latin-1' |
| |
| def __init__(self, sock=None, map=None): |
| # for string terminator matching |
| self.ac_in_buffer = b'' |
| |
| # we use a list here rather than io.BytesIO for a few reasons... |
| # del lst[:] is faster than bio.truncate(0) |
| # lst = [] is faster than bio.truncate(0) |
| self.incoming = [] |
| |
| # we toss the use of the "simple producer" and replace it with |
| # a pure deque, which the original fifo was a wrapping of |
| self.producer_fifo = deque() |
| asyncore.dispatcher.__init__(self, sock, map) |
| |
| def collect_incoming_data(self, data): |
| raise NotImplementedError("must be implemented in subclass") |
| |
| def _collect_incoming_data(self, data): |
| self.incoming.append(data) |
| |
| def _get_data(self): |
| d = b''.join(self.incoming) |
| del self.incoming[:] |
| return d |
| |
| def found_terminator(self): |
| raise NotImplementedError("must be implemented in subclass") |
| |
| def set_terminator(self, term): |
| """Set the input delimiter. |
| |
| Can be a fixed string of any length, an integer, or None. |
| """ |
| if isinstance(term, str) and self.use_encoding: |
| term = bytes(term, self.encoding) |
| elif isinstance(term, int) and term < 0: |
| raise ValueError('the number of received bytes must be positive') |
| self.terminator = term |
| |
| def get_terminator(self): |
| return self.terminator |
| |
| # grab some more data from the socket, |
| # throw it to the collector method, |
| # check for the terminator, |
| # if found, transition to the next state. |
| |
| def handle_read(self): |
| |
| try: |
| data = self.recv(self.ac_in_buffer_size) |
| except BlockingIOError: |
| return |
| except OSError as why: |
| self.handle_error() |
| return |
| |
| if isinstance(data, str) and self.use_encoding: |
| data = bytes(str, self.encoding) |
| self.ac_in_buffer = self.ac_in_buffer + data |
| |
| # Continue to search for self.terminator in self.ac_in_buffer, |
| # while calling self.collect_incoming_data. The while loop |
| # is necessary because we might read several data+terminator |
| # combos with a single recv(4096). |
| |
| while self.ac_in_buffer: |
| lb = len(self.ac_in_buffer) |
| terminator = self.get_terminator() |
| if not terminator: |
| # no terminator, collect it all |
| self.collect_incoming_data(self.ac_in_buffer) |
| self.ac_in_buffer = b'' |
| elif isinstance(terminator, int): |
| # numeric terminator |
| n = terminator |
| if lb < n: |
| self.collect_incoming_data(self.ac_in_buffer) |
| self.ac_in_buffer = b'' |
| self.terminator = self.terminator - lb |
| else: |
| self.collect_incoming_data(self.ac_in_buffer[:n]) |
| self.ac_in_buffer = self.ac_in_buffer[n:] |
| self.terminator = 0 |
| self.found_terminator() |
| else: |
| # 3 cases: |
| # 1) end of buffer matches terminator exactly: |
| # collect data, transition |
| # 2) end of buffer matches some prefix: |
| # collect data to the prefix |
| # 3) end of buffer does not match any prefix: |
| # collect data |
| terminator_len = len(terminator) |
| index = self.ac_in_buffer.find(terminator) |
| if index != -1: |
| # we found the terminator |
| if index > 0: |
| # don't bother reporting the empty string |
| # (source of subtle bugs) |
| self.collect_incoming_data(self.ac_in_buffer[:index]) |
| self.ac_in_buffer = self.ac_in_buffer[index+terminator_len:] |
| # This does the Right Thing if the terminator |
| # is changed here. |
| self.found_terminator() |
| else: |
| # check for a prefix of the terminator |
| index = find_prefix_at_end(self.ac_in_buffer, terminator) |
| if index: |
| if index != lb: |
| # we found a prefix, collect up to the prefix |
| self.collect_incoming_data(self.ac_in_buffer[:-index]) |
| self.ac_in_buffer = self.ac_in_buffer[-index:] |
| break |
| else: |
| # no prefix, collect it all |
| self.collect_incoming_data(self.ac_in_buffer) |
| self.ac_in_buffer = b'' |
| |
| def handle_write(self): |
| self.initiate_send() |
| |
| def handle_close(self): |
| self.close() |
| |
| def push(self, data): |
| if not isinstance(data, (bytes, bytearray, memoryview)): |
| raise TypeError('data argument must be byte-ish (%r)', |
| type(data)) |
| sabs = self.ac_out_buffer_size |
| if len(data) > sabs: |
| for i in range(0, len(data), sabs): |
| self.producer_fifo.append(data[i:i+sabs]) |
| else: |
| self.producer_fifo.append(data) |
| self.initiate_send() |
| |
| def push_with_producer(self, producer): |
| self.producer_fifo.append(producer) |
| self.initiate_send() |
| |
| def readable(self): |
| "predicate for inclusion in the readable for select()" |
| # cannot use the old predicate, it violates the claim of the |
| # set_terminator method. |
| |
| # return (len(self.ac_in_buffer) <= self.ac_in_buffer_size) |
| return 1 |
| |
| def writable(self): |
| "predicate for inclusion in the writable for select()" |
| return self.producer_fifo or (not self.connected) |
| |
| def close_when_done(self): |
| "automatically close this channel once the outgoing queue is empty" |
| self.producer_fifo.append(None) |
| |
| def initiate_send(self): |
| while self.producer_fifo and self.connected: |
| first = self.producer_fifo[0] |
| # handle empty string/buffer or None entry |
| if not first: |
| del self.producer_fifo[0] |
| if first is None: |
| self.handle_close() |
| return |
| |
| # handle classic producer behavior |
| obs = self.ac_out_buffer_size |
| try: |
| data = first[:obs] |
| except TypeError: |
| data = first.more() |
| if data: |
| self.producer_fifo.appendleft(data) |
| else: |
| del self.producer_fifo[0] |
| continue |
| |
| if isinstance(data, str) and self.use_encoding: |
| data = bytes(data, self.encoding) |
| |
| # send the data |
| try: |
| num_sent = self.send(data) |
| except OSError: |
| self.handle_error() |
| return |
| |
| if num_sent: |
| if num_sent < len(data) or obs < len(first): |
| self.producer_fifo[0] = first[num_sent:] |
| else: |
| del self.producer_fifo[0] |
| # we tried to send some actual data |
| return |
| |
| def discard_buffers(self): |
| # Emergencies only! |
| self.ac_in_buffer = b'' |
| del self.incoming[:] |
| self.producer_fifo.clear() |
| |
| |
| class simple_producer: |
| |
| def __init__(self, data, buffer_size=512): |
| self.data = data |
| self.buffer_size = buffer_size |
| |
| def more(self): |
| if len(self.data) > self.buffer_size: |
| result = self.data[:self.buffer_size] |
| self.data = self.data[self.buffer_size:] |
| return result |
| else: |
| result = self.data |
| self.data = b'' |
| return result |
| |
| |
| # Given 'haystack', see if any prefix of 'needle' is at its end. This |
| # assumes an exact match has already been checked. Return the number of |
| # characters matched. |
| # for example: |
| # f_p_a_e("qwerty\r", "\r\n") => 1 |
| # f_p_a_e("qwertydkjf", "\r\n") => 0 |
| # f_p_a_e("qwerty\r\n", "\r\n") => <undefined> |
| |
| # this could maybe be made faster with a computed regex? |
| # [answer: no; circa Python-2.0, Jan 2001] |
| # new python: 28961/s |
| # old python: 18307/s |
| # re: 12820/s |
| # regex: 14035/s |
| |
| def find_prefix_at_end(haystack, needle): |
| l = len(needle) - 1 |
| while l and not haystack.endswith(needle[:l]): |
| l -= 1 |
| return l |
