| 46 | |
| 47 | |
| 48 | class PipeWatcher(object): |
| 49 | def __init__(self, pipe, sink, queue=None, drop=True, suppressed=False): |
| 50 | """Watch a pipe, and buffer its output if drop is False.""" |
| 51 | self._pipe = pipe |
| 52 | self._sink = sink |
| 53 | self._drop = drop |
| 54 | self._suppressed = suppressed |
| 55 | self._filters = [] |
| 56 | |
| 57 | if queue is None: |
| 58 | self._lines = Queue() |
| 59 | else: |
| 60 | self._lines = queue |
| 61 | |
| 62 | def read_and_poll(self): |
| 63 | for line in self._pipe: |
| 64 | if self._filter(line) and not self._suppressed: |
| 65 | try: |
| 66 | self._sink.write(line) |
| 67 | self._sink.flush() |
| 68 | if not self._drop: |
| 69 | self._lines.put(line) |
| 70 | except: # noqa: E722, pylint: disable=bare-except |
| 71 | pass |
| 72 | |
| 73 | self._polling_thread = threading.Thread(target=read_and_poll, args=(self,)) |
| 74 | self._polling_thread.daemon = True |
| 75 | self._polling_thread.start() |
| 76 | |
| 77 | def poll(self, block=True, timeout=None): |
| 78 | return self._lines.get(block=block, timeout=timeout) |
| 79 | |
| 80 | def poll_all(self): |
| 81 | while True: |
| 82 | try: |
| 83 | yield self._lines.get(block=False) |
| 84 | except Empty: |
| 85 | break |
| 86 | |
| 87 | def drop(self, drop=True): |
| 88 | self._drop = drop |
| 89 | |
| 90 | def suppress(self, suppressed=True): |
| 91 | self._suppressed = suppressed |
| 92 | |
| 93 | def add_filter(self, func): |
| 94 | if not (func in self._filters): |
| 95 | self._filters.append(func) |
| 96 | |
| 97 | def _filter(self, line): |
| 98 | for func in self._filters: |
| 99 | if not func(line): # assume callable - will raise if not |
| 100 | return False |
| 101 | return True |
| 102 | |
| 103 | |
| 104 | class PipeMerger(object): |
no outgoing calls
no test coverage detected