1
# -*- test-case-name: twisted.conch.test.test_manhole -*-
2
# Copyright (c) 2001-2007 Twisted Matrix Laboratories.
3
# See LICENSE for details.
6
Asynchronous local terminal input handling
11
import os, tty, sys, termios
13
from twisted.internet import reactor, stdio, protocol, defer
14
from twisted.python import failure, reflect, log
16
from twisted.conch.insults.insults import ServerProtocol
17
from twisted.conch.manhole import ColoredManhole
19
class UnexpectedOutputError(Exception):
22
class TerminalProcessProtocol(protocol.ProcessProtocol):
23
def __init__(self, proto):
25
self.onConnection = defer.Deferred()
27
def connectionMade(self):
28
self.proto.makeConnection(self)
29
self.onConnection.callback(None)
30
self.onConnection = None
32
def write(self, bytes):
33
self.transport.write(bytes)
35
def outReceived(self, bytes):
36
self.proto.dataReceived(bytes)
38
def errReceived(self, bytes):
39
self.transport.loseConnection()
40
if self.proto is not None:
41
self.proto.connectionLost(failure.Failure(UnexpectedOutputError(bytes)))
44
def childConnectionLost(self, childFD):
45
if self.proto is not None:
46
self.proto.childConnectionLost(childFD)
48
def processEnded(self, reason):
49
if self.proto is not None:
50
self.proto.connectionLost(reason)
55
class ConsoleManhole(ColoredManhole):
57
A manhole protocol specifically for use with L{stdio.StandardIO}.
59
def connectionLost(self, reason):
61
When the connection is lost, there is nothing more to do. Stop the
62
reactor so that the process can exit.
68
def runWithProtocol(klass):
69
fd = sys.__stdin__.fileno()
70
oldSettings = termios.tcgetattr(fd)
73
p = ServerProtocol(klass)
77
termios.tcsetattr(fd, termios.TCSANOW, oldSettings)
78
os.write(fd, "\r\x1bc\r")
83
log.startLogging(file('child.log', 'w'))
88
klass = reflect.namedClass(argv[0])
90
klass = ConsoleManhole
91
runWithProtocol(klass)
94
if __name__ == '__main__':