Add asynchat.py and asyncore.py from CPython v3.12.0

Author: NakanoMiku39

Lib/test/support/asynchat.py

@@ -0,0 +1,314 @@
+# TODO: This module was deprecated and removed from CPython 3.12
+# Now it is a test-only helper. Any attempts to rewrite exising tests that
+# are using this module and remove it completely are appreciated!
+# See: https://github.com/python/cpython/issues/72719
+
+# -*- Mode: Python; tab-width: 4 -*-
+#       Id: asynchat.py,v 2.26 2000/09/07 22:29:26 rushing Exp
+#       Author: Sam Rushing <[email protected]>
+
+# ======================================================================
+# 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.
+"""
+
+from collections import deque
+
+from test.support import asyncore
+
+
+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:
+            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