python-eventlet/eventlet/greenio/py3.py

   1 import _pyio as _original_pyio
   2 import errno
   3 import os as _original_os
   4 import socket as _original_socket
   5 from io import (
   6     BufferedRandom as _OriginalBufferedRandom,
   7     BufferedReader as _OriginalBufferedReader,
   8     BufferedWriter as _OriginalBufferedWriter,
   9     DEFAULT_BUFFER_SIZE,
  10     TextIOWrapper as _OriginalTextIOWrapper,
  11     IOBase as _OriginalIOBase,
  12 )
  13 from types import FunctionType
  14
  15 from eventlet.greenio.base import (
  16     _operation_on_closed_file,
  17     greenpipe_doc,
  18     set_nonblocking,
  19     SOCKET_BLOCKING,
  20 )
  21 from eventlet.hubs import notify_close, notify_opened, IOClosed, trampoline
  22 from eventlet.support import get_errno, six
  23
  24 __all__ = ['_fileobject', 'GreenPipe']
  25
  26 # TODO get rid of this, it only seems like the original _fileobject
  27 _fileobject = _original_socket.SocketIO
  28
  29 # Large part of the following code is copied from the original
  30 # eventlet.greenio module
  31
  32
  33 class GreenFileIO(_OriginalIOBase):
  34     def __init__(self, name, mode='r', closefd=True, opener=None):
  35         if isinstance(name, int):
  36             fileno = name
  37             self._name = "<fd:%d>" % fileno
  38         else:
  39             assert isinstance(name, six.string_types)
  40             with open(name, mode) as fd:
  41                 self._name = fd.name
  42                 fileno = _original_os.dup(fd.fileno())
  43
  44         notify_opened(fileno)
  45         self._fileno = fileno
  46         self._mode = mode
  47         self._closed = False
  48         set_nonblocking(self)
  49         self._seekable = None
  50
  51     @property
  52     def closed(self):
  53         return self._closed
  54
  55     def seekable(self):
  56         if self._seekable is None:
  57             try:
  58                 _original_os.lseek(self._fileno, 0, _original_os.SEEK_CUR)
  59             except IOError as e:
  60                 if get_errno(e) == errno.ESPIPE:
  61                     self._seekable = False
  62                 else:
  63                     raise
  64             else:
  65                 self._seekable = True
  66
  67         return self._seekable
  68
  69     def readable(self):
  70         return 'r' in self._mode or '+' in self._mode
  71
  72     def writable(self):
  73         return 'w' in self._mode or '+' in self._mode
  74
  75     def fileno(self):
  76         return self._fileno
  77
  78     def read(self, buflen):
  79         while True:
  80             try:
  81                 return _original_os.read(self._fileno, buflen)
  82             except OSError as e:
  83                 if get_errno(e) not in SOCKET_BLOCKING:
  84                     raise IOError(*e.args)
  85             self._trampoline(self, read=True)
  86
  87     def readinto(self, b):
  88         up_to = len(b)
  89         data = self.read(up_to)
  90         bytes_read = len(data)
  91         b[:bytes_read] = data
  92         return bytes_read
  93
  94     def isatty(self):
  95         try:
  96             return _original_os.isatty(self.fileno())
  97         except OSError as e:
  98             raise IOError(*e.args)
  99
 100     def _trampoline(self, fd, read=False, write=False, timeout=None, timeout_exc=None):
 101         if self._closed:
 102             # Don't trampoline if we're already closed.
 103             raise IOClosed()
 104         try:
 105             return trampoline(fd, read=read, write=write, timeout=timeout,
 106                               timeout_exc=timeout_exc,
 107                               mark_as_closed=self._mark_as_closed)
 108         except IOClosed:
 109             # Our fileno has been obsoleted. Defang ourselves to
 110             # prevent spurious closes.
 111             self._mark_as_closed()
 112             raise
 113
 114     def _mark_as_closed(self):
 115         """ Mark this socket as being closed """
 116         self._closed = True
 117
 118     def write(self, data):
 119         while True:
 120             try:
 121                 return _original_os.write(self._fileno, data)
 122             except OSError as e:
 123                 if get_errno(e) not in SOCKET_BLOCKING:
 124                     raise IOError(*e.args)
 125                 else:
 126                     trampoline(self, write=True)
 127
 128     def close(self):
 129         if not self._closed:
 130             self._closed = True
 131             _original_os.close(self._fileno)
 132         notify_close(self._fileno)
 133         for method in [
 134                 'fileno', 'flush', 'isatty', 'next', 'read', 'readinto',
 135                 'readline', 'readlines', 'seek', 'tell', 'truncate',
 136                 'write', 'xreadlines', '__iter__', '__next__', 'writelines']:
 137             setattr(self, method, _operation_on_closed_file)
 138
 139     def truncate(self, size=-1):
 140         if size == -1:
 141             size = self.tell()
 142         try:
 143             rv = _original_os.ftruncate(self._fileno, size)
 144         except OSError as e:
 145             raise IOError(*e.args)
 146         else:
 147             self.seek(size)  # move position&clear buffer
 148             return rv
 149
 150     def seek(self, offset, whence=_original_os.SEEK_SET):
 151         try:
 152             return _original_os.lseek(self._fileno, offset, whence)
 153         except OSError as e:
 154             raise IOError(*e.args)
 155
 156     def __enter__(self):
 157         return self
 158
 159     def __exit__(self, *args):
 160         self.close()
 161
 162
 163 _open_environment = dict(globals())
 164 _open_environment.update(dict(
 165     BufferedRandom=_OriginalBufferedRandom,
 166     BufferedWriter=_OriginalBufferedWriter,
 167     BufferedReader=_OriginalBufferedReader,
 168     TextIOWrapper=_OriginalTextIOWrapper,
 169     FileIO=GreenFileIO,
 170     os=_original_os,
 171 ))
 172
 173 _open = FunctionType(
 174     six.get_function_code(_original_pyio.open),
 175     _open_environment,
 176 )
 177
 178
 179 def GreenPipe(name, mode="r", buffering=-1, encoding=None, errors=None,
 180               newline=None, closefd=True, opener=None):
 181     try:
 182         fileno = name.fileno()
 183     except AttributeError:
 184         pass
 185     else:
 186         fileno = _original_os.dup(fileno)
 187         name.close()
 188         name = fileno
 189
 190     return _open(name, mode, buffering, encoding, errors, newline, closefd, opener)
 191
 192 GreenPipe.__doc__ = greenpipe_doc