Streams all input data (e.g. STDIN) from the client to the agent.
(self)
| 611 | self.exit_event.set() |
| 612 | |
| 613 | def _attach_container_input(self): |
| 614 | """ |
| 615 | Streams all input data (e.g. STDIN) from the client to the agent. |
| 616 | """ |
| 617 | |
| 618 | def _initial_input_streamer(): |
| 619 | """ |
| 620 | Generator function yielding the initial ATTACH_CONTAINER_INPUT |
| 621 | message for streaming. We have a separate generator for this so |
| 622 | that we can attempt the connection once before committing to a |
| 623 | persistent connection where we stream the rest of the input. |
| 624 | |
| 625 | :returns: A RecordIO encoded message |
| 626 | """ |
| 627 | |
| 628 | message = { |
| 629 | 'type': 'ATTACH_CONTAINER_INPUT', |
| 630 | 'attach_container_input': { |
| 631 | 'type': 'CONTAINER_ID', |
| 632 | 'container_id': self.container_id}} |
| 633 | |
| 634 | yield self.encoder.encode(message) |
| 635 | |
| 636 | def _input_streamer(): |
| 637 | """ |
| 638 | Generator function yielding ATTACH_CONTAINER_INPUT messages for |
| 639 | streaming. It yields the _intitial_input_streamer() message, |
| 640 | followed by messages from the input_queue on each subsequent call. |
| 641 | |
| 642 | :returns: A RecordIO encoded message |
| 643 | """ |
| 644 | yield next(_initial_input_streamer(), None) |
| 645 | |
| 646 | while True: |
| 647 | record = self.input_queue.get() |
| 648 | if not record: |
| 649 | if self.exit_sequence_detected: |
| 650 | sys.stdout.write("\r\n") |
| 651 | sys.stdout.flush() |
| 652 | self.exit_event.set() |
| 653 | break |
| 654 | yield record |
| 655 | |
| 656 | req_extra_args = { |
| 657 | 'additional_headers': { |
| 658 | 'Content-Type': 'application/recordio', |
| 659 | 'Message-Content-Type': 'application/json', |
| 660 | 'Accept': 'application/json', |
| 661 | 'Connection': 'close', |
| 662 | 'Transfer-Encoding': 'chunked' |
| 663 | } |
| 664 | } |
| 665 | |
| 666 | # Ensure we don't try to attach our input to a container that isn't |
| 667 | # fully up and running by waiting until the |
| 668 | # `_process_output_stream` function signals us that it's ready. |
| 669 | self.attach_input_event.wait() |
| 670 |