Create a namedpipe pipe ``addr`` :returns type: :class:`PipeConnection`
(cls, addr, security_descriptor=None)
| 38 | |
| 39 | @classmethod |
| 40 | def create(cls, addr, security_descriptor=None): |
| 41 | """Create a namedpipe pipe ``addr`` |
| 42 | |
| 43 | :returns type: :class:`PipeConnection` |
| 44 | """ |
| 45 | addr = full_pipe_address(addr) |
| 46 | |
| 47 | security_attributes = None |
| 48 | if security_descriptor is not None: |
| 49 | if isinstance(security_descriptor, str): |
| 50 | security_descriptor = windows.security.SecurityDescriptor.from_string(security_descriptor) |
| 51 | security_attributes = gdef.SECURITY_ATTRIBUTES() |
| 52 | security_attributes.nLength = ctypes.sizeof(security_attributes) |
| 53 | security_attributes.lpSecurityDescriptor = security_descriptor # Accept as arg ? |
| 54 | security_attributes.bInheritHandle = True # Accept as arg ? |
| 55 | |
| 56 | |
| 57 | pipehandle = winproxy.CreateNamedPipeA( |
| 58 | addr, gdef.PIPE_ACCESS_DUPLEX, |
| 59 | gdef.PIPE_TYPE_MESSAGE | gdef.PIPE_READMODE_MESSAGE | |
| 60 | gdef.PIPE_WAIT, |
| 61 | gdef.PIPE_UNLIMITED_INSTANCES, cls.BUFFER_SIZE, cls.BUFFER_SIZE, |
| 62 | gdef.NMPWAIT_WAIT_FOREVER, security_attributes |
| 63 | ) |
| 64 | return cls.from_handle(pipehandle, name=addr, server=True) |
| 65 | |
| 66 | @classmethod |
| 67 | def connect(cls, addr): |
nothing calls this directly
no test coverage detected