(self, *pipes)
| 56 | print("###### %s" % pn) |
| 57 | |
| 58 | def __init__(self, *pipes): |
| 59 | # type: (*Pipe) -> None |
| 60 | ObjectPipe.__init__(self, "PipeEngine") |
| 61 | self.active_pipes = set() # type: Set[Pipe] |
| 62 | self.active_sources = set() # type: Set[Union[Source, PipeEngine]] |
| 63 | self.active_drains = set() # type: Set[Pipe] |
| 64 | self.active_sinks = set() # type: Set[Pipe] |
| 65 | self._add_pipes(*pipes) |
| 66 | self.thread_lock = Lock() |
| 67 | self.command_lock = Lock() |
| 68 | self.thread = None # type: Optional[Thread] |
| 69 | |
| 70 | def __getattr__(self, attr): |
| 71 | # type: (str) -> Callable[..., Pipe] |
nothing calls this directly
no test coverage detected