1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
|
# -*- test-case-name: twisted.test.test_stdio -*-
"""
Windows-specific implementation of the L{twisted.internet.stdio} interface.
"""
import msvcrt
import os
from zope.interface import implementer
import win32api
from twisted.internet import _pollingfile, main
from twisted.internet.interfaces import (
IAddress,
IConsumer,
IHalfCloseableProtocol,
IPushProducer,
ITransport,
)
from twisted.logger import Logger
from twisted.python.failure import Failure
_log = Logger()
@implementer(IAddress)
class Win32PipeAddress:
pass
@implementer(ITransport, IConsumer, IPushProducer)
class StandardIO(_pollingfile._PollingTimer):
disconnecting = False
disconnected = False
def __init__(self, proto, reactor=None):
"""
Start talking to standard IO with the given protocol.
Also, put it stdin/stdout/stderr into binary mode.
"""
if reactor is None:
from twisted.internet import reactor
for stdfd in range(0, 1, 2):
msvcrt.setmode(stdfd, os.O_BINARY)
_pollingfile._PollingTimer.__init__(self, reactor)
self.proto = proto
hstdin = win32api.GetStdHandle(win32api.STD_INPUT_HANDLE)
hstdout = win32api.GetStdHandle(win32api.STD_OUTPUT_HANDLE)
self.stdin = _pollingfile._PollableReadPipe(
hstdin, self.dataReceived, self.readConnectionLost
)
self.stdout = _pollingfile._PollableWritePipe(hstdout, self.writeConnectionLost)
self._addPollableResource(self.stdin)
self._addPollableResource(self.stdout)
self.proto.makeConnection(self)
def dataReceived(self, data):
self.proto.dataReceived(data)
def readConnectionLost(self):
with _log.failuresHandled("read connection lost") as op:
if IHalfCloseableProtocol.providedBy(self.proto):
self.proto.readConnectionLost()
self.checkConnLost()
if not op.succeeded and not self.disconnecting:
self.loseConnection()
def writeConnectionLost(self):
with _log.failuresHandled("write connection lost") as op:
if IHalfCloseableProtocol.providedBy(self.proto):
self.proto.writeConnectionLost()
self.checkConnLost()
if not op.succeeded and not self.disconnecting:
self.loseConnection()
connsLost = 0
def checkConnLost(self):
self.connsLost += 1
if self.connsLost >= 2:
self.disconnecting = True
self.disconnected = True
self.proto.connectionLost(Failure(main.CONNECTION_DONE))
# ITransport
def write(self, data):
self.stdout.write(data)
def writeSequence(self, seq):
self.stdout.write(b"".join(seq))
def loseConnection(self):
self.disconnecting = True
self.stdin.close()
self.stdout.close()
def getPeer(self):
return Win32PipeAddress()
def getHost(self):
return Win32PipeAddress()
# IConsumer
def registerProducer(self, producer, streaming):
return self.stdout.registerProducer(producer, streaming)
def unregisterProducer(self):
return self.stdout.unregisterProducer()
# def write() above
# IProducer
def stopProducing(self):
self.stdin.stopProducing()
# IPushProducer
def pauseProducing(self):
self.stdin.pauseProducing()
def resumeProducing(self):
self.stdin.resumeProducing()
|
