Gecko [ oric.aiou.edu.pk ]

Name	Size	Permission
__pycache__	[ DIR ]	drwxr-xr-x
iocpreactor	[ DIR ]	drwxr-xr-x
test	[ DIR ]	drwxr-xr-x
__init__.py	521 B	-rw-r--r--
_baseprocess.py	1.87 KB	-rw-r--r--
_dumbwin32proc.py	12.8 KB	-rw-r--r--
_glibbase.py	12.51 KB	-rw-r--r--
_idna.py	1.36 KB	-rw-r--r--
_newtls.py	9.11 KB	-rw-r--r--
_pollingfile.py	8.77 KB	-rw-r--r--
_posixserialport.py	1.95 KB	-rw-r--r--
_posixstdio.py	4.58 KB	-rw-r--r--
_producer_helpers.py	3.7 KB	-rw-r--r--
_resolver.py	8.33 KB	-rw-r--r--
_signals.py	2.65 KB	-rw-r--r--
_sslverify.py	70.09 KB	-rw-r--r--
_threadedselect.py	11.49 KB	-rw-r--r--
_win32serialport.py	4.63 KB	-rw-r--r--
_win32stdio.py	3.13 KB	-rw-r--r--
abstract.py	18.82 KB	-rw-r--r--
address.py	5.12 KB	-rw-r--r--
asyncioreactor.py	10.31 KB	-rw-r--r--
base.py	43.22 KB	-rw-r--r--
cfreactor.py	17.09 KB	-rw-r--r--
default.py	1.9 KB	-rw-r--r--
defer.py	69.52 KB	-rw-r--r--
endpoints.py	75.4 KB	-rw-r--r--
epollreactor.py	8.29 KB	-rw-r--r--
error.py	12.35 KB	-rw-r--r--
fdesc.py	3.15 KB	-rw-r--r--
gireactor.py	5.98 KB	-rw-r--r--
glib2reactor.py	1.09 KB	-rw-r--r--
gtk2reactor.py	3.53 KB	-rw-r--r--
gtk3reactor.py	2.2 KB	-rw-r--r--
inotify.py	14.35 KB	-rw-r--r--
interfaces.py	94.25 KB	-rw-r--r--
kqreactor.py	10.05 KB	-rw-r--r--
main.py	1.03 KB	-rw-r--r--
pollreactor.py	5.88 KB	-rw-r--r--
posixbase.py	25.74 KB	-rw-r--r--
process.py	37.98 KB	-rw-r--r--
protocol.py	26.5 KB	-rw-r--r--
pyuisupport.py	817 B	-rw-r--r--
reactor.py	1.82 KB	-rw-r--r--
selectreactor.py	6.07 KB	-rw-r--r--
serialport.py	2.26 KB	-rw-r--r--
ssl.py	8.25 KB	-rw-r--r--
stdio.py	1.02 KB	-rw-r--r--
task.py	30.39 KB	-rw-r--r--
tcp.py	53.62 KB	-rw-r--r--
threads.py	3.86 KB	-rw-r--r--
tksupport.py	2 KB	-rw-r--r--
udp.py	18.13 KB	-rw-r--r--
unix.py	21.45 KB	-rw-r--r--
utils.py	7.69 KB	-rw-r--r--
win32eventreactor.py	14.84 KB	-rw-r--r--
wxreactor.py	5.14 KB	-rw-r--r--
wxsupport.py	1.33 KB	-rw-r--r--

Code Editor : _win32stdio.py

# -*- test-case-name: twisted.test.test_stdio -*-

"""
Windows-specific implementation of the L{twisted.internet.stdio} interface.
"""

from __future__ import absolute_import, division

import win32api
import os
import msvcrt

from zope.interface import implementer

from twisted.internet.interfaces import (IHalfCloseableProtocol, ITransport,
                                         IConsumer, IPushProducer, IAddress)

from twisted.internet import _pollingfile, main
from twisted.python.failure import Failure

@implementer(IAddress)
class Win32PipeAddress(object):
    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):
        if IHalfCloseableProtocol.providedBy(self.proto):
            self.proto.readConnectionLost()
        self.checkConnLost()

def writeConnectionLost(self):
        if IHalfCloseableProtocol.providedBy(self.proto):
            self.proto.writeConnectionLost()
        self.checkConnLost()

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()

${this.title}

Code Editor : _win32stdio.py