Like os.pipe() but with overlapped support and using handles not fds.
(*, duplex=False, overlapped=(True, True), bufsize=BUFSIZE)
| 30 | |
| 31 | |
| 32 | def pipe(*, duplex=False, overlapped=(True, True), bufsize=BUFSIZE): |
| 33 | """Like os.pipe() but with overlapped support and using handles not fds.""" |
| 34 | address = tempfile.mktemp( |
| 35 | prefix=r'\\.\pipe\python-pipe-{:d}-{:d}-'.format( |
| 36 | os.getpid(), next(_mmap_counter))) |
| 37 | |
| 38 | if duplex: |
| 39 | openmode = _winapi.PIPE_ACCESS_DUPLEX |
| 40 | access = _winapi.GENERIC_READ | _winapi.GENERIC_WRITE |
| 41 | obsize, ibsize = bufsize, bufsize |
| 42 | else: |
| 43 | openmode = _winapi.PIPE_ACCESS_INBOUND |
| 44 | access = _winapi.GENERIC_WRITE |
| 45 | obsize, ibsize = 0, bufsize |
| 46 | |
| 47 | openmode |= _winapi.FILE_FLAG_FIRST_PIPE_INSTANCE |
| 48 | |
| 49 | if overlapped[0]: |
| 50 | openmode |= _winapi.FILE_FLAG_OVERLAPPED |
| 51 | |
| 52 | if overlapped[1]: |
| 53 | flags_and_attribs = _winapi.FILE_FLAG_OVERLAPPED |
| 54 | else: |
| 55 | flags_and_attribs = 0 |
| 56 | |
| 57 | h1 = h2 = None |
| 58 | try: |
| 59 | h1 = _winapi.CreateNamedPipe( |
| 60 | address, openmode, _winapi.PIPE_WAIT, |
| 61 | 1, obsize, ibsize, _winapi.NMPWAIT_WAIT_FOREVER, _winapi.NULL) |
| 62 | |
| 63 | h2 = _winapi.CreateFile( |
| 64 | address, access, 0, _winapi.NULL, _winapi.OPEN_EXISTING, |
| 65 | flags_and_attribs, _winapi.NULL) |
| 66 | |
| 67 | ov = _winapi.ConnectNamedPipe(h1, overlapped=True) |
| 68 | ov.GetOverlappedResult(True) |
| 69 | return h1, h2 |
| 70 | except: |
| 71 | if h1 is not None: |
| 72 | _winapi.CloseHandle(h1) |
| 73 | if h2 is not None: |
| 74 | _winapi.CloseHandle(h2) |
| 75 | raise |
| 76 | |
| 77 | |
| 78 | # Wrapper for a pipe handle |
no test coverage detected