Function receive

3rdparty/libprocess/src/process.cpp:801–856

namespace internal {

void receive(Socket socket)
{
  StreamingRequestDecoder* decoder = new StreamingRequestDecoder();

  const size_t size = 80 * 1024;
  char* data = new char[size];

  Future<Nothing> recv_loop = process::loop(
      None(),
      [=] {
        return socket.recv(data, size);
      },
      [=](size_t length) -> Future<ControlFlow<Nothing>> {
        if (length == 0) {
          return Break(); // EOF.
        }

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

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

        if (!requests.empty()) {
          // Get the peer address to augment the requests.
          Try<Address> address = socket.peer();

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

          foreach (Request* request, requests) {
            request->client = address.get();
            process_manager->handle(socket, request);
          }
        }

        return Continue();
      });

  recv_loop.onAny([=](const Future<Nothing> f) {
    if (f.isFailed()) {
      Try<Address> peer = socket.peer();

      LOG(WARNING)
        << "Failed to recv on socket " << socket.get() << " to peer '"
        << (peer.isSome() ? stringify(peer.get()) : "unknown")
        << "': " << f.failure();
    }

    socket_manager->close(socket);
    delete[] data;
    delete decoder;
  });
}

} // namespace internal {