Function receive

3rdparty/libprocess/src/http.cpp:1937–1992

Future<Nothing> receive(
    network::Socket socket,
    std::function<Future<Response>(const Request&)>&& f,
    Queue<Option<Item>> pipeline)
{
  // Get the peer address to augment any requests we receive.
  Try<network::Address> address = socket.peer();

  if (address.isError()) {
    return Failure("Failed to get peer address: " + address.error());
  }

  const size_t size = io::BUFFERED_READ_SIZE;
  char* data = new char[size];

  StreamingRequestDecoder* decoder = new StreamingRequestDecoder();

  return loop(
      [=]() {
        return socket.recv(data, size);
      },
      [=](size_t length) mutable -> Future<ControlFlow<Nothing>> {
        if (length == 0) {
          return Break();
        }

        // Decode as much of the data as possible into HTTP requests.
        const deque<Request*> requests = decoder->decode(data, length);

        // NOTE: it's possible the decoder has failed but some
        // requests might be available, i.e., `requests.empty()` is
        // not true, so we wait to return a `Failure` until when there
        // are no requests.

        if (decoder->failed() && requests.empty()) {
          return Failure("Decoder error while receiving");
        }

        foreach (Request* request, requests) {
          request->client = address.get();
          // TODO(benh): To support HTTP pipelining we invoke `f`
          // regardless of whether the previous response has been
          // completed. This can make handling of requests more
          // difficult so we could consider supporting disabling HTTP
          // pipelining via some sort of "options" initially passed
          // in.
          pipeline.put(Item{request, f(*request)});
        }

        return Continue(); // Keep looping!
      })
    .onAny([=]() {
      delete decoder;
      delete[] data;
    });
}