def _make_eof_intr():
"""Set constants _EOF and _INTR.
This avoids doing potentially costly operations on module load.
"""
global _EOF, _INTR
if (_EOF is not None) and (_INTR is not None):
return
# inherit EOF and INTR definitions from controlling process.
try:
from termios import VEOF, VINTR
try:
fd = sys.__stdin__.fileno()
except ValueError:
# ValueError: I/O operation on closed file
fd = sys.__stdout__.fileno()
intr = ord(termios.tcgetattr(fd)[6][VINTR])
eof = ord(termios.tcgetattr(fd)[6][VEOF])
except (ImportError, OSError, IOError, ValueError, termios.error):
# unless the controlling process is also not a terminal,
# such as cron(1), or when stdin and stdout are both closed.
# Fall-back to using CEOF and CINTR. There
try:
from termios import CEOF, CINTR
(intr, eof) = (CINTR, CEOF)
except ImportError:
# ^C, ^D
(intr, eof) = (3, 4)
_INTR = _byte(intr)
_EOF = _byte(eof)
python类VEOF的实例源码
def _make_eof_intr():
"""Set constants _EOF and _INTR.
This avoids doing potentially costly operations on module load.
"""
global _EOF, _INTR
if (_EOF is not None) and (_INTR is not None):
return
# inherit EOF and INTR definitions from controlling process.
try:
from termios import VEOF, VINTR
try:
fd = sys.__stdin__.fileno()
except ValueError:
# ValueError: I/O operation on closed file
fd = sys.__stdout__.fileno()
intr = ord(termios.tcgetattr(fd)[6][VINTR])
eof = ord(termios.tcgetattr(fd)[6][VEOF])
except (ImportError, OSError, IOError, ValueError, termios.error):
# unless the controlling process is also not a terminal,
# such as cron(1), or when stdin and stdout are both closed.
# Fall-back to using CEOF and CINTR. There
try:
from termios import CEOF, CINTR
(intr, eof) = (CINTR, CEOF)
except ImportError:
# ^C, ^D
(intr, eof) = (3, 4)
_INTR = _byte(intr)
_EOF = _byte(eof)
def sendeof(self):
"""This sends an EOF to the child. This sends a character which causes
the pending parent output buffer to be sent to the waiting child
program without waiting for end-of-line. If it is the first character
of the line, the read() in the user program returns 0, which signifies
end-of-file. This means to work as expected a sendeof() has to be
called at the beginning of a line. This method does not send a newline.
It is the responsibility of the caller to ensure the eof is sent at the
beginning of a line. """
### Hmmm... how do I send an EOF?
###C if ((m = write(pty, *buf, p - *buf)) < 0)
###C return (errno == EWOULDBLOCK) ? n : -1;
#fd = sys.stdin.fileno()
#old = termios.tcgetattr(fd) # remember current state
#attr = termios.tcgetattr(fd)
#attr[3] = attr[3] | termios.ICANON # ICANON must be set to recognize EOF
#try: # use try/finally to ensure state gets restored
# termios.tcsetattr(fd, termios.TCSADRAIN, attr)
# if hasattr(termios, 'CEOF'):
# os.write (self.child_fd, '%c' % termios.CEOF)
# else:
# # Silly platform does not define CEOF so assume CTRL-D
# os.write (self.child_fd, '%c' % 4)
#finally: # restore state
# termios.tcsetattr(fd, termios.TCSADRAIN, old)
if hasattr(termios, 'VEOF'):
char = termios.tcgetattr(self.child_fd)[6][termios.VEOF]
else:
# platform does not define VEOF so assume CTRL-D
char = chr(4)
self.send(char)
def _make_eof_intr():
"""Set constants _EOF and _INTR.
This avoids doing potentially costly operations on module load.
"""
global _EOF, _INTR
if (_EOF is not None) and (_INTR is not None):
return
# inherit EOF and INTR definitions from controlling process.
try:
from termios import VEOF, VINTR
try:
fd = sys.__stdin__.fileno()
except ValueError:
# ValueError: I/O operation on closed file
fd = sys.__stdout__.fileno()
intr = ord(termios.tcgetattr(fd)[6][VINTR])
eof = ord(termios.tcgetattr(fd)[6][VEOF])
except (ImportError, OSError, IOError, ValueError, termios.error):
# unless the controlling process is also not a terminal,
# such as cron(1), or when stdin and stdout are both closed.
# Fall-back to using CEOF and CINTR. There
try:
from termios import CEOF, CINTR
(intr, eof) = (CINTR, CEOF)
except ImportError:
# ^C, ^D
(intr, eof) = (3, 4)
_INTR = _byte(intr)
_EOF = _byte(eof)
def _make_eof_intr():
"""Set constants _EOF and _INTR.
This avoids doing potentially costly operations on module load.
"""
global _EOF, _INTR
if (_EOF is not None) and (_INTR is not None):
return
# inherit EOF and INTR definitions from controlling process.
try:
from termios import VEOF, VINTR
try:
fd = sys.__stdin__.fileno()
except ValueError:
# ValueError: I/O operation on closed file
fd = sys.__stdout__.fileno()
intr = ord(termios.tcgetattr(fd)[6][VINTR])
eof = ord(termios.tcgetattr(fd)[6][VEOF])
except (ImportError, OSError, IOError, ValueError, termios.error):
# unless the controlling process is also not a terminal,
# such as cron(1), or when stdin and stdout are both closed.
# Fall-back to using CEOF and CINTR. There
try:
from termios import CEOF, CINTR
(intr, eof) = (CINTR, CEOF)
except ImportError:
# ^C, ^D
(intr, eof) = (3, 4)
_INTR = _byte(intr)
_EOF = _byte(eof)
def _make_eof_intr():
"""Set constants _EOF and _INTR.
This avoids doing potentially costly operations on module load.
"""
global _EOF, _INTR
if (_EOF is not None) and (_INTR is not None):
return
# inherit EOF and INTR definitions from controlling process.
try:
from termios import VEOF, VINTR
try:
fd = sys.__stdin__.fileno()
except ValueError:
# ValueError: I/O operation on closed file
fd = sys.__stdout__.fileno()
intr = ord(termios.tcgetattr(fd)[6][VINTR])
eof = ord(termios.tcgetattr(fd)[6][VEOF])
except (ImportError, OSError, IOError, ValueError, termios.error):
# unless the controlling process is also not a terminal,
# such as cron(1), or when stdin and stdout are both closed.
# Fall-back to using CEOF and CINTR. There
try:
from termios import CEOF, CINTR
(intr, eof) = (CINTR, CEOF)
except ImportError:
# ^C, ^D
(intr, eof) = (3, 4)
_INTR = _byte(intr)
_EOF = _byte(eof)
def sendeof(self):
"""This sends an EOF to the child. This sends a character which causes
the pending parent output buffer to be sent to the waiting child
program without waiting for end-of-line. If it is the first character
of the line, the read() in the user program returns 0, which signifies
end-of-file. This means to work as expected a sendeof() has to be
called at the beginning of a line. This method does not send a newline.
It is the responsibility of the caller to ensure the eof is sent at the
beginning of a line. """
# Hmmm... how do I send an EOF?
# C if ((m = write(pty, *buf, p - *buf)) < 0)
# C return (errno == EWOULDBLOCK) ? n : -1;
#fd = sys.stdin.fileno()
# old = termios.tcgetattr(fd) # remember current state
#attr = termios.tcgetattr(fd)
# attr[3] = attr[3] | termios.ICANON # ICANON must be set to recognize EOF
# try: # use try/finally to ensure state gets restored
# termios.tcsetattr(fd, termios.TCSADRAIN, attr)
# if hasattr(termios, 'CEOF'):
# os.write (self.child_fd, '%c' % termios.CEOF)
# else:
# # Silly platform does not define CEOF so assume CTRL-D
# os.write (self.child_fd, '%c' % 4)
# finally: # restore state
# termios.tcsetattr(fd, termios.TCSADRAIN, old)
if hasattr(termios, 'VEOF'):
char = termios.tcgetattr(self.child_fd)[6][termios.VEOF]
else:
# platform does not define VEOF so assume CTRL-D
char = chr(4)
self.send(char)
def write(self):
""" attempt to get a chunk of data to write to our child process's
stdin, then write it. the return value answers the questions "are we
done writing forever?" """
# get_chunk may sometimes return bytes, and sometimes return strings
# because of the nature of the different types of STDIN objects we
# support
try:
chunk = self.get_chunk()
if chunk is None:
raise DoneReadingForever
except DoneReadingForever:
self.log.debug("done reading")
if self.tty_in:
# EOF time
try:
char = termios.tcgetattr(self.stream)[6][termios.VEOF]
except:
char = chr(4).encode()
# normally, one EOF should be enough to signal to an program
# that is read()ing, to return 0 and be on your way. however,
# some programs are misbehaved, like python3.1 and python3.2.
# they don't stop reading sometimes after read() returns 0.
# this can be demonstrated with the following program:
#
# import sys
# sys.stdout.write(sys.stdin.read())
#
# then type 'a' followed by ctrl-d 3 times. in python
# 2.6,2.7,3.3,3.4,3.5,3.6, it only takes 2 ctrl-d to terminate.
# however, in python 3.1 and 3.2, it takes all 3.
#
# so here we send an extra EOF along, just in case. i don't
# believe it can hurt anything
os.write(self.stream, char)
os.write(self.stream, char)
return True
except NotYetReadyToRead:
self.log.debug("received no data")
return False
# if we're not bytes, make us bytes
if IS_PY3 and hasattr(chunk, "encode"):
chunk = chunk.encode(self.encoding)
for proc_chunk in self.stream_bufferer.process(chunk):
self.log.debug("got chunk size %d: %r", len(proc_chunk),
proc_chunk[:30])
self.log.debug("writing chunk to process")
try:
os.write(self.stream, proc_chunk)
except OSError:
self.log.debug("OSError writing stdin chunk")
return True
