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 import fcntl
  22 import os
  23
  24 from gsh.console import console_output
  25
  26 class buffered_dispatcher(asyncore.file_dispatcher):
  27     """A dispatcher with a write buffer to allow asynchronous writers, and a
  28     read buffer to permit line oriented manipulations"""
  29
  30     # 1 MiB should be enough for everybody
  31     MAX_BUFFER_SIZE = 1 * 1024 * 1024
  32
  33     def __init__(self, fd):
  34         asyncore.file_dispatcher.__init__(self, fd)
  35         self.fd = fd
  36         self.read_buffer = ''
  37         self.write_buffer = ''
  38         self.allow_write = True
  39
  40     def handle_read(self):
  41         """Some data can be read"""
  42         new_data = ''
  43         buffer_length = len(self.read_buffer)
  44         try:
  45             while buffer_length < buffered_dispatcher.MAX_BUFFER_SIZE:
  46                 try:
  47                     piece = self.recv(4096)
  48                 except OSError, e:
  49                     if e.errno == errno.EAGAIN:
  50                         # End of the available data
  51                         break
  52                     elif e.errno == errno.EIO and new_data:
  53                         # Hopefully we could read an error message before the
  54                         # actual termination
  55                         break
  56                     else:
  57                         raise
  58                 new_data += piece
  59                 buffer_length += len(piece)
  60         finally:
  61             new_data = new_data.replace('\r', '\n')
  62             self.read_buffer += new_data
  63         return new_data
  64
  65     def readable(self):
  66         """No need to ask data if our buffer is already full"""
  67         return len(self.read_buffer) < buffered_dispatcher.MAX_BUFFER_SIZE
  68
  69     def writable(self):
  70         """Do we have something to write?"""
  71         return self.write_buffer != ''
  72
  73     def dispatch_write(self, buf):
  74         """Augment the buffer with stuff to write when possible"""
  75         assert self.allow_write
  76         self.write_buffer += buf
  77         if len(self.write_buffer) > buffered_dispatcher.MAX_BUFFER_SIZE:
  78             console_output('Buffer too big (%d) for %s\n' %
  79                                             (len(self.write_buffer), str(self)))
  80             raise asyncore.ExitNow(1)
  81
  82     def drain_and_block_writing(self):
  83         # set the fd to blocking mode
  84         self.allow_write = False
  85         flags = fcntl.fcntl(self.fd, fcntl.F_GETFL, 0)
  86         flags = flags & ~os.O_NONBLOCK
  87         fcntl.fcntl(self.fd, fcntl.F_SETFL, flags)
  88         if self.writable():
  89             self.handle_write()
  90
  91     def allow_writing(self):
  92         # set the fd to non-blocking mode
  93         flags = fcntl.fcntl(self.fd, fcntl.F_GETFL, 0)
  94         flags = flags | os.O_NONBLOCK
  95         fcntl.fcntl(self.fd, fcntl.F_SETFL, flags)
  96         self.allow_write = True