parameters: (self)
docstring: Streams all input data (e.g. STDIN) from the client to the agent.
docstring_summary: Streams all input data (e.g. STDIN) from the client to the agent.
docstring_tokens: Streams all input data ( e . g . STDIN ) from the client to the agent .
function: def _attach_container_input(self):
          """
          Streams all input data (e.g. STDIN) from the client to the agent.
          """
  
          def _initial_input_streamer():
              """
              Generator function yielding the initial ATTACH_CONTAINER_INPUT
              message for streaming. We have a separate generator for this so
              that we can attempt the connection once before committing to a
              persistent connection where we stream the rest of the input.
  
              :returns: A RecordIO encoded message
              """
  
              message = {
                  'type': 'ATTACH_CONTAINER_INPUT',
                  'attach_container_input': {
                      'type': 'CONTAINER_ID',
                      'container_id': self.container_id}}
  
              yield self.encoder.encode(message)
  
          def _input_streamer():
              """
              Generator function yielding ATTACH_CONTAINER_INPUT messages for
              streaming. It yields the _intitial_input_streamer() message,
              followed by messages from the input_queue on each subsequent call.
  
              :returns: A RecordIO encoded message
              """
              yield next(_initial_input_streamer(), None)
  
              while True:
                  record = self.input_queue.get()
                  if not record:
                      if self.exit_sequence_detected:
                          sys.stdout.write("\r\n")
                          sys.stdout.flush()
                          self.exit_event.set()
                      break
                  yield record
  
          req_extra_args = {
              'verify': self.config.agent_ssl_verify(),
              'additional_headers': {
                  'Content-Type': 'application/recordio',
                  'Message-Content-Type': 'application/json',
                  'Accept': 'application/json',
                  'Connection': 'close',
                  'Transfer-Encoding': 'chunked'
              }
          }
  
          # Ensure we don't try to attach our input to a container that isn't
          # fully up and running by waiting until the
          # `_process_output_stream` function signals us that it's ready.
          self.attach_input_event.wait()
  
          # Send an intial "Test" message to ensure that we are able to
          # establish a connection with the agent. If we aren't we will throw
          # an exception and break out of this thread. However, in cases where
          # we receive a 500 response from the agent, we actually want to
          # continue without throwing an exception. A 500 error indicates that
          # we can't connect to the container because it has already finished
          # running. In that case we continue running to allow the output queue
          # to be flushed.
          resource = mesos.http.Resource(self.agent_url)
          try:
              resource.request(
                  mesos.http.METHOD_POST,
                  data=_initial_input_streamer(),
                  retry=False,
                  auth=self.config.authentication_header(),
                  **req_extra_args)
          except MesosHTTPException as e:
              if not e.response.status_code == 500:
                  raise e
  
          # If we succeeded with that connection, unblock process_output_stream()
          # from sending output data to the output thread.
          self.print_output_event.set()
  
          # Begin streaming the input.
          resource = mesos.http.Resource(self.agent_url)
          resource.request(
              mesos.http.METHOD_POST,
              data=_input_streamer(),
              retry=False,
              timeout=None,
              auth=self.config.authentication_header(),
              **req_extra_args)
function_tokens: def _attach_container_input ( self ) : def _initial_input_streamer ( ) : """
              Generator function yielding the initial ATTACH_CONTAINER_INPUT
              message for streaming. We have a separate generator for this so
              that we can attempt the connection once before committing to a
              persistent connection where we stream the rest of the input.
  
              :returns: A RecordIO encoded message
              """ message = { 'type' : 'ATTACH_CONTAINER_INPUT' , 'attach_container_input' : { 'type' : 'CONTAINER_ID' , 'container_id' : self . container_id } } yield self . encoder . encode ( message ) def _input_streamer ( ) : """
              Generator function yielding ATTACH_CONTAINER_INPUT messages for
              streaming. It yields the _intitial_input_streamer() message,
              followed by messages from the input_queue on each subsequent call.
  
              :returns: A RecordIO encoded message
              """ yield next ( _initial_input_streamer ( ) , None ) while True : record = self . input_queue . get ( ) if not record : if self . exit_sequence_detected : sys . stdout . write ( "\r\n" ) sys . stdout . flush ( ) self . exit_event . set ( ) break yield record req_extra_args = { 'verify' : self . config . agent_ssl_verify ( ) , 'additional_headers' : { 'Content-Type' : 'application/recordio' , 'Message-Content-Type' : 'application/json' , 'Accept' : 'application/json' , 'Connection' : 'close' , 'Transfer-Encoding' : 'chunked' } } # Ensure we don't try to attach our input to a container that isn't # fully up and running by waiting until the # `_process_output_stream` function signals us that it's ready. self . attach_input_event . wait ( ) # Send an intial "Test" message to ensure that we are able to # establish a connection with the agent. If we aren't we will throw # an exception and break out of this thread. However, in cases where # we receive a 500 response from the agent, we actually want to # continue without throwing an exception. A 500 error indicates that # we can't connect to the container because it has already finished # running. In that case we continue running to allow the output queue # to be flushed. resource = mesos . http . Resource ( self . agent_url ) try : resource . request ( mesos . http . METHOD_POST , data = _initial_input_streamer ( ) , retry = False , auth = self . config . authentication_header ( ) , * * req_extra_args ) except MesosHTTPException as e : if not e . response . status_code == 500 : raise e # If we succeeded with that connection, unblock process_output_stream() # from sending output data to the output thread. self . print_output_event . set ( ) # Begin streaming the input. resource = mesos . http . Resource ( self . agent_url ) resource . request ( mesos . http . METHOD_POST , data = _input_streamer ( ) , retry = False , timeout = None , auth = self . config . authentication_header ( ) , * * req_extra_args )