MCPcopy Index your code
hub / github.com/RustPython/RustPython / __init__

Method __init__

Lib/asyncio/windows_utils.py:130–173  ·  view source on GitHub ↗
(self, args, stdin=None, stdout=None, stderr=None, **kwds)

Source from the content-addressed store, hash-verified

128 The stdin, stdout, stderr are None or instances of PipeHandle.
129 """
130 def __init__(self, args, stdin=None, stdout=None, stderr=None, **kwds):
131 assert not kwds.get('universal_newlines')
132 assert kwds.get('bufsize', 0) == 0
133 stdin_rfd = stdout_wfd = stderr_wfd = None
134 stdin_wh = stdout_rh = stderr_rh = None
135 if stdin == PIPE:
136 stdin_rh, stdin_wh = pipe(overlapped=(False, True), duplex=True)
137 stdin_rfd = msvcrt.open_osfhandle(stdin_rh, os.O_RDONLY)
138 else:
139 stdin_rfd = stdin
140 if stdout == PIPE:
141 stdout_rh, stdout_wh = pipe(overlapped=(True, False))
142 stdout_wfd = msvcrt.open_osfhandle(stdout_wh, 0)
143 else:
144 stdout_wfd = stdout
145 if stderr == PIPE:
146 stderr_rh, stderr_wh = pipe(overlapped=(True, False))
147 stderr_wfd = msvcrt.open_osfhandle(stderr_wh, 0)
148 elif stderr == STDOUT:
149 stderr_wfd = stdout_wfd
150 else:
151 stderr_wfd = stderr
152 try:
153 super().__init__(args, stdin=stdin_rfd, stdout=stdout_wfd,
154 stderr=stderr_wfd, **kwds)
155 except:
156 for h in (stdin_wh, stdout_rh, stderr_rh):
157 if h is not None:
158 _winapi.CloseHandle(h)
159 raise
160 else:
161 if stdin_wh is not None:
162 self.stdin = PipeHandle(stdin_wh)
163 if stdout_rh is not None:
164 self.stdout = PipeHandle(stdout_rh)
165 if stderr_rh is not None:
166 self.stderr = PipeHandle(stderr_rh)
167 finally:
168 if stdin == PIPE:
169 os.close(stdin_rfd)
170 if stdout == PIPE:
171 os.close(stdout_wfd)
172 if stderr == PIPE:
173 os.close(stderr_wfd)

Callers

nothing calls this directly

Calls 5

superClass · 0.85
PipeHandleClass · 0.85
pipeFunction · 0.70
getMethod · 0.45
closeMethod · 0.45

Tested by

no test coverage detected