Method _attachContainerInput

src/slave/http.cpp:3870–3949

Future<Response> Http::_attachContainerInput(
    const mesos::agent::Call& call,
    Owned<Reader<mesos::agent::Call>>&& decoder,
    const RequestMediaTypes& mediaTypes) const
{
  const ContainerID& containerId = call.attach_container_input().container_id();

  Pipe pipe;
  Pipe::Reader reader = pipe.reader();
  Pipe::Writer writer = pipe.writer();

  CHECK_SOME(mediaTypes.messageContent);
  auto encoder = [mediaTypes](const mesos::agent::Call& call) {
    string record = serialize(mediaTypes.messageContent.get(), call);
    return ::recordio::encode(record);
  };

  // Write the first record. We had extracted it from the `decoder`
  // in the `api()` handler to identify the call type earlier.
  pipe.writer().write(encoder(call));

  // We create this here since C++11 does not support move capture of `reader`.
  Future<Nothing> transform = recordio::transform<mesos::agent::Call>(
      std::move(decoder), encoder, writer);

  return slave->containerizer->attach(containerId)
    .then(defer(slave->self(), [=](
        Connection connection) mutable -> Future<Response> {
      Request request;
      request.method = "POST";
      request.type = Request::PIPE;
      request.reader = reader;
      request.headers = {{"Content-Type", stringify(mediaTypes.content)},
                         {MESSAGE_CONTENT_TYPE,
                             stringify(mediaTypes.messageContent.get())},
                         {"Accept", stringify(mediaTypes.accept)}};

      // See comments in `attachContainerOutput()` for the reasoning
      // behind these values.
      request.url.domain = "";
      request.url.path = "/";

      transform
        .onAny([writer](
            const Future<Nothing>& future) mutable {
          CHECK(!future.isDiscarded());

          if (future.isFailed()) {
            writer.fail(future.failure());
            return;
          }

          writer.close();
         });

      // This is a non Keep-Alive request which means the connection
      // will be closed when the response is received. Since the
      // 'Connection' is reference-counted, we must maintain a copy