todo this module was deprecated and removed from cpython 3 12 now it is a testonly helper any attempts to rewrite exising tests that are using this module and remove it completely are appreciated see https github compythoncpythonissues72719 mode python tabwidth 4 id asynchat py v 2 26 20000907 22 29 26 rushing exp sam rushing rushingnightmare com 1996 by sam rushing 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 notice appear in all copies and that both that 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 from collections import deque from test support import asyncore class asyncchatasyncore dispatcher these are overridable defaults 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 for string terminator matching we use a list here rather than io bytesio for a few reasons del lst is faster than bio truncate0 lst is faster than bio truncate0 we toss the use of the simple producer and replace it with a pure deque which the original fifo was a wrapping of set the input delimiter can be a fixed string of any length an integer or none grab some more data from the socket throw it to the collector method check for the terminator if found transition to the next state continue to search for self terminator in self acinbuffer while calling self collectincomingdata the while loop is necessary because we might read several dataterminator combos with a single recv4096 no terminator collect it all numeric terminator 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 we found the terminator don t bother reporting the empty string source of subtle bugs this does the right thing if the terminator is changed here check for a prefix of the terminator we found a prefix collect up to the prefix no prefix collect it all cannot use the old predicate it violates the claim of the setterminator method return lenself acinbuffer self acinbuffersize handle empty stringbuffer or none entry handle classic producer behavior send the data we tried to send some actual data emergencies only 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 fpaeqwertyr rn 1 fpaeqwertydkjf rn 0 fpaeqwertyrn rn undefined this could maybe be made faster with a computed regex answer no circa python2 0 jan 2001 new python 28961s old python 18307s re 12820s regex 14035s 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 sam rushing rushing nightmare com 1996 by sam rushing 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 notice appear in all copies and that both that 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 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 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 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 for string terminator matching 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 we toss the use of the simple producer and replace it with a pure deque which the original fifo was a wrapping of set the input delimiter can be a fixed string of any length an integer or none grab some more data from the socket throw it to the collector method check for the terminator if found transition to the next state 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 no terminator collect it all numeric terminator 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 we found the terminator don t bother reporting the empty string source of subtle bugs this does the right thing if the terminator is changed here check for a prefix of the terminator we found a prefix collect up to the prefix no prefix collect it all 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 handle empty string buffer or none entry handle classic producer behavior send the data we tried to send some actual data emergencies only 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
r

from collections import deque

from test.support import asyncore


class async_chat(asyncore.dispatcher):
    

    

    ac_in_buffer_size = 65536
    ac_out_buffer_size = 65536

    
    

    use_encoding = 0
    encoding = 'latin-1'

    def __init__(self, sock=None, map=None):
        
        self.ac_in_buffer = b''

        
        
        
        self.incoming = []

        
        
        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):
        
        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

    
    
    
    

    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

        
        
        
        

        while self.ac_in_buffer:
            lb = len(self.ac_in_buffer)
            terminator = self.get_terminator()
            if not terminator:
                
                self.collect_incoming_data(self.ac_in_buffer)
                self.ac_in_buffer = b''
            elif isinstance(terminator, int):
                
                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:
                
                
                
                
                
                
                
                terminator_len = len(terminator)
                index = self.ac_in_buffer.find(terminator)
                if index != -1:
                    
                    if index > 0:
                        
                        
                        self.collect_incoming_data(self.ac_in_buffer[:index])
                    self.ac_in_buffer = self.ac_in_buffer[index+terminator_len:]
                    
                    
                    self.found_terminator()
                else:
                    
                    index = find_prefix_at_end(self.ac_in_buffer, terminator)
                    if index:
                        if index != lb:
                            
                            self.collect_incoming_data(self.ac_in_buffer[:-index])
                            self.ac_in_buffer = self.ac_in_buffer[-index:]
                        break
                    else:
                        
                        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()"
        
        

        
        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]
            
            if not first:
                del self.producer_fifo[0]
                if first is None:
                    self.handle_close()
                    return

            
            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)

            
            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]
            
            return

    def discard_buffers(self):
        
        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

















def find_prefix_at_end(haystack, needle):
    l = len(needle) - 1
    while l and not haystack.endswith(needle[:l]):
        l -= 1
    return l