gsh/buffered_dispatcher.py

   1 # This program is free software; you can redistribute it and/or modify
   2 # it under the terms of the GNU General Public License as published by
   3 # the Free Software Foundation; either version 2 of the License, or
   4 # (at your option) any later version.
   5 #
   6 # This program is distributed in the hope that it will be useful,
   7 # but WITHOUT ANY WARRANTY; without even the implied warranty of
   8 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
   9 # GNU Library General Public License for more details.
  10 #
  11 # You should have received a copy of the GNU General Public License
  12 # along with this program; if not, write to the Free Software
  13 # Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
  14 #
  15 # See the COPYING file for license information.
  16 #
  17 # Copyright (c) 2006, 2007 Guillaume Chazarain <guichaz@yahoo.fr>
  18
  19 import asyncore
  20 import errno
  21
  22 class buffered_dispatcher(asyncore.file_dispatcher):
  23     """A dispatcher with a write buffer to allow asynchronous writers, and a
  24     read buffer to permit line oriented manipulations"""
  25
  26     # 1 MiB should be enough for everybody
  27     MAX_BUFFER_SIZE = 1 * 1024 * 1024
  28
  29     # We always try to read PIECE_SIZE bytes at a time
  30     PIECE_SIZE = 4096
  31
  32     def __init__(self, fd):
  33         asyncore.file_dispatcher.__init__(self, fd)
  34         self.fd = fd
  35         self.read_buffer = ''
  36         self.write_buffer = ''
  37
  38     def handle_error(self):
  39         """Handle the Ctrl-C or print the exception and its stack trace.
  40         Returns True if it was an actual error"""
  41         try:
  42             raise
  43         except KeyboardInterrupt:
  44             # The main loop will launch the control shell
  45             raise
  46         except OSError:
  47             # I/O error, let the parent take action
  48             return True
  49
  50     def handle_expt(self):
  51         # Emulate the select with poll as in: asyncore.loop(use_poll=True)
  52         self.handle_read()
  53
  54     def handle_read(self):
  55         """Some data can be read"""
  56         new_data = ''
  57         buffer_length = len(self.read_buffer)
  58         while buffer_length < buffered_dispatcher.MAX_BUFFER_SIZE:
  59             try:
  60                 piece = self.recv(buffered_dispatcher.PIECE_SIZE)
  61             except OSError, e:
  62                 if e.errno == errno.EAGAIN:
  63                     # End of the available data
  64                     piece = None
  65                 else:
  66                     raise
  67             if not piece:
  68                 break
  69             new_data += piece
  70             piece_len = len(piece)
  71             buffer_length += piece_len
  72             if piece_len < buffered_dispatcher.PIECE_SIZE:
  73                 # No need to try another read, this one was the last
  74                 break
  75         self.read_buffer += new_data
  76         return new_data
  77
  78     def readable(self):
  79         """No need to ask data if our buffer is full"""
  80         return len(self.read_buffer) < buffered_dispatcher.MAX_BUFFER_SIZE
  81
  82     def writable(self):
  83         """Do we have something to write?"""
  84         return self.write_buffer != ''
  85
  86     def handle_write(self):
  87         """Let's write as much as we can"""
  88         num_sent = self.send(self.write_buffer)
  89         self.write_buffer = self.write_buffer[num_sent:]
  90
  91     def dispatch_write(self, buf):
  92         """Augment the buffer with stuff to write when possible"""
  93         self.write_buffer += buf
  94         if len(self.write_buffer) > buffered_dispatcher.MAX_BUFFER_SIZE:
  95             raise Exception, 'Buffer too big (%d)' % (len(self.write_buffer))