(self, name: str, timeout: float | None = None)
| 242 | BUFFER_SIZE: Final = 2**16 |
| 243 | |
| 244 | def __init__(self, name: str, timeout: float | None = None) -> None: |
| 245 | if sys.platform == "win32": |
| 246 | name = r"\\.\pipe\{}-{}.pipe".format(name, urlsafe_b64encode(os.urandom(6)).decode()) |
| 247 | else: |
| 248 | name = f"{name}.sock" |
| 249 | super().__init__(name, timeout) |
| 250 | if sys.platform == "win32": |
| 251 | self.connection = _winapi.CreateNamedPipe( |
| 252 | self.name, |
| 253 | _winapi.PIPE_ACCESS_DUPLEX |
| 254 | | _winapi.FILE_FLAG_FIRST_PIPE_INSTANCE |
| 255 | | _winapi.FILE_FLAG_OVERLAPPED, |
| 256 | _winapi.PIPE_READMODE_MESSAGE |
| 257 | | _winapi.PIPE_TYPE_MESSAGE |
| 258 | | _winapi.PIPE_WAIT |
| 259 | | 0x8, # PIPE_REJECT_REMOTE_CLIENTS |
| 260 | 1, # one instance |
| 261 | self.BUFFER_SIZE, |
| 262 | self.BUFFER_SIZE, |
| 263 | _winapi.NMPWAIT_WAIT_FOREVER, |
| 264 | 0, # Use default security descriptor |
| 265 | ) |
| 266 | if self.connection == -1: # INVALID_HANDLE_VALUE |
| 267 | err = _winapi.GetLastError() |
| 268 | raise IPCException(f"Invalid handle to pipe: {err}") |
| 269 | else: |
| 270 | self.sock_directory = tempfile.mkdtemp() |
| 271 | sockfile = os.path.join(self.sock_directory, self.name) |
| 272 | self.sock = socket.socket(socket.AF_UNIX) |
| 273 | self.sock.bind(sockfile) |
| 274 | self.sock.listen(1) |
| 275 | if timeout is not None: |
| 276 | self.sock.settimeout(timeout) |
| 277 | |
| 278 | def __enter__(self) -> IPCServer: |
| 279 | if sys.platform == "win32": |
nothing calls this directly
no test coverage detected