1 # -*- test-case-name: twisted.conch.test.test_manhole -*-
2 # Copyright (c) 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__':