# -*- 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
# 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
# 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
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
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_out_buffer_size
= 4096
def __init__ (self
, conn
=None):
self
.producer_fifo
= fifo()
asyncore
.dispatcher
.__init
__ (self
, conn
)
def collect_incoming_data(self
, data
):
raise NotImplementedError, "must be implemented in subclass"
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"
def get_terminator (self
):
# grab some more data from the socket,
# throw it to the collector method,
# check for the terminator,
# if found, transition to the next state.
data
= self
.recv (self
.ac_in_buffer_size
)
except socket
.error
, why
:
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(1024).
lb
= len(self
.ac_in_buffer
)
terminator
= self
.get_terminator()
if terminator
is None or terminator
== '':
# no terminator, collect it all
self
.collect_incoming_data (self
.ac_in_buffer
)
elif isinstance(terminator
, int):
self
.collect_incoming_data (self
.ac_in_buffer
)
self
.terminator
= self
.terminator
- lb
self
.collect_incoming_data (self
.ac_in_buffer
[:n
])
self
.ac_in_buffer
= self
.ac_in_buffer
[n
:]
# 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:
terminator_len
= len(terminator
)
index
= self
.ac_in_buffer
.find(terminator
)
# we found the terminator
# 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.
# check for a prefix of the terminator
index
= find_prefix_at_end (self
.ac_in_buffer
, terminator
)
# 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
:]
# no prefix, collect it all
self
.collect_incoming_data (self
.ac_in_buffer
)
self
.producer_fifo
.push (simple_producer (data
))
def push_with_producer (self
, producer
):
self
.producer_fifo
.push (producer
)
"predicate for inclusion in the readable for select()"
return (len(self
.ac_in_buffer
) <= self
.ac_in_buffer_size
)
"predicate for inclusion in the writable for select()"
# return len(self.ac_out_buffer) or len(self.producer_fifo) or (not self.connected)
# this is about twice as fast, though not as clear.
(self
.ac_out_buffer
== '') and
self
.producer_fifo
.is_empty() and
def close_when_done (self
):
"automatically close this channel once the outgoing queue is empty"
self
.producer_fifo
.push (None)
# refill the outgoing buffer by calling the more() method
# of the first producer in the queue
def refill_buffer (self
):
if len(self
.producer_fifo
):
p
= self
.producer_fifo
.first()
# a 'None' in the producer fifo is a sentinel,
# telling us to close the channel.
if not self
.ac_out_buffer
:
self
.ac_out_buffer
= self
.ac_out_buffer
+ p
self
.ac_out_buffer
= self
.ac_out_buffer
+ data
def initiate_send (self
):
obs
= self
.ac_out_buffer_size
# try to refill the buffer
if (len (self
.ac_out_buffer
) < obs
):
if self
.ac_out_buffer
and self
.connected
:
num_sent
= self
.send (self
.ac_out_buffer
[:obs
])
self
.ac_out_buffer
= self
.ac_out_buffer
[num_sent
:]
except socket
.error
, why
:
def discard_buffers (self
):
while self
.producer_fifo
:
def __init__ (self
, data
, buffer_size
=512):
self
.buffer_size
= buffer_size
if len (self
.data
) > self
.buffer_size
:
result
= self
.data
[:self
.buffer_size
]
self
.data
= self
.data
[self
.buffer_size
:]
def __init__ (self
, list=None):
return (1, self
.list.popleft())
# 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
# 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]
def find_prefix_at_end (haystack
, needle
):
while l
and not haystack
.endswith(needle
[:l
]):