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, 2008 Guillaume Chazarain <guichaz@gmail.com>
  18
  19 import asyncore
  20 import errno
  21
  22 from gsh.console import console_output
  23
  24 class buffered_dispatcher(asyncore.file_dispatcher):
  25     """A dispatcher with a write buffer to allow asynchronous writers, and a
  26     read buffer to permit line oriented manipulations"""
  27
  28     # 1 MiB should be enough for everybody
  29     MAX_BUFFER_SIZE = 1 * 1024 * 1024
  30
  31     def __init__(self, fd):
  32         asyncore.file_dispatcher.__init__(self, fd)
  33         self.fd = fd
  34         self.read_buffer = ''
  35         self.write_buffer = ''
  36
  37     def handle_error(self):
  38         """Handle the Ctrl-C or print the exception and its stack trace.
  39         Returns True if it was an actual error"""
  40         try:
  41             raise
  42         except OSError:
  43             # I/O error, let the parent take action
  44             return True
  45
  46     def handle_expt(self):
  47         # Emulate the select with poll as in: asyncore.loop(use_poll=True)
  48         self.handle_read()
  49
  50     def handle_read(self):
  51         """Some data can be read"""
  52         new_data = ''
  53         buffer_length = len(self.read_buffer)
  54         try:
  55             while buffer_length < buffered_dispatcher.MAX_BUFFER_SIZE:
  56                 try:
  57                     piece = self.recv(4096)
  58                 except OSError, e:
  59                     if e.errno == errno.EAGAIN:
  60                         # End of the available data
  61                         break
  62                     elif e.errno == errno.EIO and new_data:
  63                         # Hopefully we could read an error message before the
  64                         # actual termination
  65                         break
  66                     else:
  67                         raise
  68                 new_data += piece
  69                 buffer_length += len(piece)
  70         finally:
  71             new_data = new_data.replace('\r', '\n')
  72             self.read_buffer += new_data
  73         return new_data
  74
  75     def readable(self):
  76         """No need to ask data if our buffer is already full"""
  77         return len(self.read_buffer) < buffered_dispatcher.MAX_BUFFER_SIZE
  78
  79     def writable(self):
  80         """Do we have something to write?"""
  81         return self.write_buffer != ''
  82
  83     def dispatch_write(self, buf):
  84         """Augment the buffer with stuff to write when possible"""
  85         self.write_buffer += buf
  86         if len(self.write_buffer) > buffered_dispatcher.MAX_BUFFER_SIZE:
  87             console_output('Buffer too big (%d) for %s\n' %
  88                                             (len(self.write_buffer), str(self)))
  89             raise asyncore.ExitNow(1)
  90