Reads from the output_queue and writes the data to the appropriate STDOUT or STDERR.
(self)
| 769 | self.input_queue.put(None) |
| 770 | |
| 771 | def _output_thread(self): |
| 772 | """ |
| 773 | Reads from the output_queue and writes the data |
| 774 | to the appropriate STDOUT or STDERR. |
| 775 | """ |
| 776 | |
| 777 | while True: |
| 778 | # Get a message from the output queue and decode it. |
| 779 | # Then write the data to the appropriate stdout or stderr. |
| 780 | output = self.output_queue.get() |
| 781 | if not output.get('data'): |
| 782 | raise CLIException("Error no 'data' field in output message") |
| 783 | |
| 784 | data = output['data'] |
| 785 | data = base64.b64decode(data.encode('utf-8')) |
| 786 | |
| 787 | if output.get('type') and output['type'] == 'STDOUT': |
| 788 | sys.stdout.buffer.write(data) |
| 789 | sys.stdout.flush() |
| 790 | elif output.get('type') and output['type'] == 'STDERR': |
| 791 | sys.stderr.buffer.write(data) |
| 792 | sys.stderr.flush() |
| 793 | else: |
| 794 | raise CLIException("Unsupported data type in output stream") |
| 795 | |
| 796 | self.output_queue.task_done() |
| 797 | |
| 798 | def _heartbeat_thread(self): |
| 799 | """ |
nothing calls this directly
no test coverage detected