You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@mesos.apache.org by an...@apache.org on 2016/11/22 03:27:21 UTC
[01/13] mesos git commit: Disabled tests relying on filtering HTTP
events.
Repository: mesos
Updated Branches:
refs/heads/master 69efd9a73 -> 0ca11623e
Disabled tests relying on filtering HTTP events.
Some tests that rely on filtering HTTP events based on type won't
work now since the request body is not yet known when \`visit()\`
is invoked. These would be enabled later as part of a separate JIRA
issue.
Review: https://reviews.apache.org/r/53491/
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/0ca11623
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/0ca11623
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/0ca11623
Branch: refs/heads/master
Commit: 0ca11623e6deb98dc05ec90b0339960c251cd64e
Parents: 5cd134f
Author: Anand Mazumdar <an...@apache.org>
Authored: Mon Nov 21 18:08:44 2016 -0800
Committer: Anand Mazumdar <an...@apache.org>
Committed: Mon Nov 21 19:27:08 2016 -0800
----------------------------------------------------------------------
src/tests/master_tests.cpp | 2 +-
src/tests/slave_recovery_tests.cpp | 10 +++++-----
2 files changed, 6 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/0ca11623/src/tests/master_tests.cpp
----------------------------------------------------------------------
diff --git a/src/tests/master_tests.cpp b/src/tests/master_tests.cpp
index c8cd892..dfedbbd 100644
--- a/src/tests/master_tests.cpp
+++ b/src/tests/master_tests.cpp
@@ -5382,7 +5382,7 @@ TEST_F(MasterTest, RejectFrameworkWithInvalidFailoverTimeout)
// This test verifies that we recover resources when an orphaned task reaches
// a terminal state.
-TEST_F(MasterTest, RecoverResourcesOrphanedTask)
+TEST_F(MasterTest, DISABLED_RecoverResourcesOrphanedTask)
{
Try<Owned<cluster::Master>> master = StartMaster();
ASSERT_SOME(master);
http://git-wip-us.apache.org/repos/asf/mesos/blob/0ca11623/src/tests/slave_recovery_tests.cpp
----------------------------------------------------------------------
diff --git a/src/tests/slave_recovery_tests.cpp b/src/tests/slave_recovery_tests.cpp
index 99c2739..3b7d8cf 100644
--- a/src/tests/slave_recovery_tests.cpp
+++ b/src/tests/slave_recovery_tests.cpp
@@ -429,7 +429,7 @@ TYPED_TEST(SlaveRecoveryTest, RecoverStatusUpdateManager)
// The slave is stopped before the first update for a task is received from the
// HTTP based command executor. When it comes back up with recovery=reconnect,
// make sure the executor subscribes and the slave properly sends the update.
-TYPED_TEST(SlaveRecoveryTest, ReconnectHTTPExecutor)
+TYPED_TEST(SlaveRecoveryTest, DISABLED_ReconnectHTTPExecutor)
{
Try<Owned<cluster::Master>> master = this->StartMaster();
ASSERT_SOME(master);
@@ -525,7 +525,7 @@ TYPED_TEST(SlaveRecoveryTest, ReconnectHTTPExecutor)
//
// TODO(anand): Remove the `ROOT_CGROUPS` prefix once the posix isolator
// is nested aware.
-TYPED_TEST(SlaveRecoveryTest, ROOT_CGROUPS_ReconnectDefaultExecutor)
+TYPED_TEST(SlaveRecoveryTest, DISABLED_ROOT_CGROUPS_ReconnectDefaultExecutor)
{
Try<Owned<cluster::Master>> master = this->StartMaster();
ASSERT_SOME(master);
@@ -811,7 +811,7 @@ TYPED_TEST(SlaveRecoveryTest, ReconnectExecutor)
// The slave is stopped before the HTTP based command executor is
// registered. When it comes back up with recovery=reconnect, make
// sure the executor is killed and the task is transitioned to LOST.
-TYPED_TEST(SlaveRecoveryTest, RecoverUnregisteredHTTPExecutor)
+TYPED_TEST(SlaveRecoveryTest, DISABLED_RecoverUnregisteredHTTPExecutor)
{
Try<Owned<cluster::Master>> master = this->StartMaster();
ASSERT_SOME(master);
@@ -1610,7 +1610,7 @@ TYPED_TEST(SlaveRecoveryTest, RecoverCompletedExecutor)
// The slave is stopped before a terminal update is received from the HTTP
// based command executor. The slave is then restarted in recovery=cleanup mode.
// It kills the executor, and terminates. Master should then send TASK_LOST.
-TYPED_TEST(SlaveRecoveryTest, CleanupHTTPExecutor)
+TYPED_TEST(SlaveRecoveryTest, DISABLED_CleanupHTTPExecutor)
{
Try<Owned<cluster::Master>> master = this->StartMaster();
ASSERT_SOME(master);
@@ -2016,7 +2016,7 @@ TYPED_TEST(SlaveRecoveryTest, NonCheckpointingFramework)
// command executor that has been running before the slave restarted.
// This test ensures that a restarted slave is able to communicate
// with all components (scheduler, master, executor).
-TYPED_TEST(SlaveRecoveryTest, KillTaskWithHTTPExecutor)
+TYPED_TEST(SlaveRecoveryTest, DISABLED_KillTaskWithHTTPExecutor)
{
Try<Owned<cluster::Master>> master = this->StartMaster();
ASSERT_SOME(master);
[08/13] mesos git commit: Introduced a `readAll()` helper on
`http::Pipe::Reader`.
Posted by an...@apache.org.
Introduced a `readAll()` helper on `http::Pipe::Reader`.
The helper reads from the pipe till EOF. This is used later to
read BODY requests from the streaming request decoder.
Review: https://reviews.apache.org/r/53485/
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/e8e3fe59
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/e8e3fe59
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/e8e3fe59
Branch: refs/heads/master
Commit: e8e3fe596f242767fc10ccb95cbdcd36c49a89a5
Parents: 5152728
Author: Anand Mazumdar <an...@apache.org>
Authored: Mon Nov 21 18:07:29 2016 -0800
Committer: Anand Mazumdar <an...@apache.org>
Committed: Mon Nov 21 19:27:08 2016 -0800
----------------------------------------------------------------------
3rdparty/libprocess/include/process/http.hpp | 11 +++++++
3rdparty/libprocess/src/http.cpp | 25 ++++++++++++++
3rdparty/libprocess/src/tests/http_tests.cpp | 40 +++++++++++++++++++++++
3 files changed, 76 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/e8e3fe59/3rdparty/libprocess/include/process/http.hpp
----------------------------------------------------------------------
diff --git a/3rdparty/libprocess/include/process/http.hpp b/3rdparty/libprocess/include/process/http.hpp
index ed853d8..a684e09 100644
--- a/3rdparty/libprocess/include/process/http.hpp
+++ b/3rdparty/libprocess/include/process/http.hpp
@@ -307,6 +307,12 @@ public:
// is closed.
Future<std::string> read();
+ // Performs a series of asynchronous reads, until EOF is reached.
+ // Returns the concatenated result of the reads.
+ // Returns Failure if the writer failed, or the read-end
+ // is closed.
+ Future<std::string> readAll();
+
// Closing the read-end of the pipe before the write-end closes
// or fails will notify the writer that the reader is no longer
// interested. Returns false if the read-end was already closed.
@@ -327,6 +333,11 @@ public:
explicit Reader(const std::shared_ptr<Data>& _data) : data(_data) {}
+ // Continuation for `readAll()`.
+ static Future<std::string> _readAll(
+ Pipe::Reader reader,
+ const std::shared_ptr<std::string>& buffer);
+
std::shared_ptr<Data> data;
};
http://git-wip-us.apache.org/repos/asf/mesos/blob/e8e3fe59/3rdparty/libprocess/src/http.cpp
----------------------------------------------------------------------
diff --git a/3rdparty/libprocess/src/http.cpp b/3rdparty/libprocess/src/http.cpp
index b5ddac9..ca7c948 100644
--- a/3rdparty/libprocess/src/http.cpp
+++ b/3rdparty/libprocess/src/http.cpp
@@ -418,6 +418,31 @@ Future<string> Pipe::Reader::read()
}
+Future<string> Pipe::Reader::readAll()
+{
+ std::shared_ptr<string> buffer(new string());
+
+ return _readAll(*this, buffer);
+}
+
+
+Future<string> Pipe::Reader::_readAll(
+ Pipe::Reader reader,
+ const std::shared_ptr<string>& buffer)
+{
+ return reader.read()
+ .then([reader, buffer](const string& read) -> Future<string> {
+ if (read.empty()) { // EOF.
+ return std::move(*buffer);
+ }
+
+ buffer->append(read);
+
+ return _readAll(reader, buffer);
+ });
+}
+
+
bool Pipe::Reader::close()
{
bool closed = false;
http://git-wip-us.apache.org/repos/asf/mesos/blob/e8e3fe59/3rdparty/libprocess/src/tests/http_tests.cpp
----------------------------------------------------------------------
diff --git a/3rdparty/libprocess/src/tests/http_tests.cpp b/3rdparty/libprocess/src/tests/http_tests.cpp
index 9147679..10d3fb0 100644
--- a/3rdparty/libprocess/src/tests/http_tests.cpp
+++ b/3rdparty/libprocess/src/tests/http_tests.cpp
@@ -395,6 +395,46 @@ TEST(HTTPTest, PipeFailure)
}
+TEST(HTTPTest, PipeReadAll)
+{
+ {
+ http::Pipe pipe;
+ http::Pipe::Reader reader = pipe.reader();
+ http::Pipe::Writer writer = pipe.writer();
+
+ Future<string> readAll = reader.readAll();
+ EXPECT_TRUE(readAll.isPending());
+
+ // Close the writer after writing some data. This should result in
+ // a successful `readAll()` operation.
+ EXPECT_TRUE(writer.write("hello"));
+ EXPECT_TRUE(writer.write("world"));
+
+ EXPECT_TRUE(writer.close());
+
+ AWAIT_EXPECT_EQ("helloworld", readAll);
+ }
+
+ {
+ http::Pipe pipe;
+ http::Pipe::Reader reader = pipe.reader();
+ http::Pipe::Writer writer = pipe.writer();
+
+ Future<string> readAll = reader.readAll();
+ EXPECT_TRUE(readAll.isPending());
+
+ // Fail the writer after writing some data. This should result in
+ // a failed `readAll()` operation.
+ EXPECT_TRUE(writer.write("hello"));
+ EXPECT_TRUE(writer.write("world"));
+
+ EXPECT_TRUE(writer.fail("disconnected!"));
+
+ AWAIT_EXPECT_FAILED(readAll);
+ }
+}
+
+
TEST(HTTPTest, PipeReaderCloses)
{
http::Pipe pipe;
[06/13] mesos git commit: Introduced a reader member to `Request` to
support request streaming.
Posted by an...@apache.org.
Introduced a reader member to `Request` to support request streaming.
These new members are needed for supporting request streaming i.e.,
the caller can use the writer to stream chunks to the server if
the request body is not known in advance.
Review: https://reviews.apache.org/r/53483/
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/f6286048
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/f6286048
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/f6286048
Branch: refs/heads/master
Commit: f6286048a5f897ff6859b38a24c3d64aa3b54d01
Parents: 1003f3d
Author: Anand Mazumdar <an...@apache.org>
Authored: Mon Nov 21 18:07:20 2016 -0800
Committer: Anand Mazumdar <an...@apache.org>
Committed: Mon Nov 21 19:27:08 2016 -0800
----------------------------------------------------------------------
3rdparty/libprocess/include/process/http.hpp | 24 +++++++++++++++++------
1 file changed, 18 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/f6286048/3rdparty/libprocess/include/process/http.hpp
----------------------------------------------------------------------
diff --git a/3rdparty/libprocess/include/process/http.hpp b/3rdparty/libprocess/include/process/http.hpp
index 51bfc7e..ed853d8 100644
--- a/3rdparty/libprocess/include/process/http.hpp
+++ b/3rdparty/libprocess/include/process/http.hpp
@@ -414,7 +414,7 @@ private:
struct Request
{
Request()
- : keepAlive(false) {}
+ : keepAlive(false), type(BODY) {}
std::string method;
@@ -426,11 +426,6 @@ struct Request
Headers headers;
- // TODO(bmahler): Add a 'query' field which contains both
- // the URL query and the parsed form data from the body.
-
- std::string body;
-
// TODO(bmahler): Ensure this is consistent with the 'Connection'
// header; perhaps make this a function that checks the header.
//
@@ -447,6 +442,23 @@ struct Request
// Note that this may correspond to a proxy or load balancer address.
network::Address client;
+ // Clients can choose to provide the entire body at once
+ // via BODY or can choose to stream the body over to the
+ // server via PIPE.
+ //
+ // Default: BODY.
+ enum
+ {
+ BODY,
+ PIPE
+ } type;
+
+ // TODO(bmahler): Add a 'query' field which contains both
+ // the URL query and the parsed form data from the body.
+
+ std::string body;
+ Option<Pipe::Reader> reader;
+
/**
* Returns whether the encoding is considered acceptable in the
* response. See RFC 2616 section 14.3 for details.
[07/13] mesos git commit: Introduced `RouteOptions` to support
streaming requests.
Posted by an...@apache.org.
Introduced `RouteOptions` to support streaming requests.
This allows routes to specify configuration options. Currently, it
only has one member `streaming` i.e, if the route supports request
streaming. This also enables us to add more options in the future
without polluting overloads.
Review: https://reviews.apache.org/r/53484/
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/5152728e
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/5152728e
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/5152728e
Branch: refs/heads/master
Commit: 5152728e3eeac8d6fac52545d0ebc5df6f2e42cb
Parents: f628604
Author: Anand Mazumdar <an...@apache.org>
Authored: Mon Nov 21 18:07:25 2016 -0800
Committer: Anand Mazumdar <an...@apache.org>
Committed: Mon Nov 21 19:27:08 2016 -0800
----------------------------------------------------------------------
3rdparty/libprocess/include/process/process.hpp | 28 +++++++++++++++-----
3rdparty/libprocess/src/process.cpp | 8 ++++--
2 files changed, 28 insertions(+), 8 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/5152728e/3rdparty/libprocess/include/process/process.hpp
----------------------------------------------------------------------
diff --git a/3rdparty/libprocess/include/process/process.hpp b/3rdparty/libprocess/include/process/process.hpp
index f389d3d..d2ce4fe 100644
--- a/3rdparty/libprocess/include/process/process.hpp
+++ b/3rdparty/libprocess/include/process/process.hpp
@@ -253,6 +253,17 @@ protected:
typedef lambda::function<Future<http::Response>(const http::Request&)>
HttpRequestHandler;
+ // Options to control the behavior of a route.
+ struct RouteOptions
+ {
+ RouteOptions()
+ : requestStreaming(false) {}
+
+ // Set to true if the endpoint supports request streaming.
+ // Default: false.
+ bool requestStreaming;
+ };
+
/**
* Sets up a handler for HTTP requests with the specified name.
*
@@ -262,7 +273,8 @@ protected:
void route(
const std::string& name,
const Option<std::string>& help,
- const HttpRequestHandler& handler);
+ const HttpRequestHandler& handler,
+ const RouteOptions& options = RouteOptions());
/**
* @copydoc process::ProcessBase::route
@@ -271,14 +283,15 @@ protected:
void route(
const std::string& name,
const Option<std::string>& help,
- Future<http::Response> (T::*method)(const http::Request&))
+ Future<http::Response> (T::*method)(const http::Request&),
+ const RouteOptions& options = RouteOptions())
{
// Note that we use dynamic_cast here so a process can use
// multiple inheritance if it sees so fit (e.g., to implement
// multiple callback interfaces).
HttpRequestHandler handler =
lambda::bind(method, dynamic_cast<T*>(this), lambda::_1);
- route(name, help, handler);
+ route(name, help, handler, options);
}
/**
@@ -305,7 +318,8 @@ protected:
const std::string& name,
const std::string& realm,
const Option<std::string>& help,
- const AuthenticatedHttpRequestHandler& handler);
+ const AuthenticatedHttpRequestHandler& handler,
+ const RouteOptions& options = RouteOptions());
/**
* @copydoc process::ProcessBase::route
@@ -317,14 +331,15 @@ protected:
const Option<std::string>& help,
Future<http::Response> (T::*method)(
const http::Request&,
- const Option<std::string>&))
+ const Option<std::string>&),
+ const RouteOptions& options = RouteOptions())
{
// Note that we use dynamic_cast here so a process can use
// multiple inheritance if it sees so fit (e.g., to implement
// multiple callback interfaces).
AuthenticatedHttpRequestHandler handler =
lambda::bind(method, dynamic_cast<T*>(this), lambda::_1, lambda::_2);
- route(name, realm, help, handler);
+ route(name, realm, help, handler, options);
}
/**
@@ -423,6 +438,7 @@ private:
Option<std::string> realm;
Option<AuthenticatedHttpRequestHandler> authenticatedHandler;
+ RouteOptions options;
};
// Handlers for messages and HTTP requests.
http://git-wip-us.apache.org/repos/asf/mesos/blob/5152728e/3rdparty/libprocess/src/process.cpp
----------------------------------------------------------------------
diff --git a/3rdparty/libprocess/src/process.cpp b/3rdparty/libprocess/src/process.cpp
index 6dc10df..86628a6 100644
--- a/3rdparty/libprocess/src/process.cpp
+++ b/3rdparty/libprocess/src/process.cpp
@@ -3786,13 +3786,15 @@ UPID ProcessBase::link(const UPID& to, const RemoteConnection remote)
void ProcessBase::route(
const string& name,
const Option<string>& help_,
- const HttpRequestHandler& handler)
+ const HttpRequestHandler& handler,
+ const RouteOptions& options)
{
// Routes must start with '/'.
CHECK(name.find('/') == 0);
HttpEndpoint endpoint;
endpoint.handler = handler;
+ endpoint.options = options;
handlers.http[name.substr(1)] = endpoint;
@@ -3804,7 +3806,8 @@ void ProcessBase::route(
const string& name,
const string& realm,
const Option<string>& help_,
- const AuthenticatedHttpRequestHandler& handler)
+ const AuthenticatedHttpRequestHandler& handler,
+ const RouteOptions& options)
{
// Routes must start with '/'.
CHECK(name.find('/') == 0);
@@ -3812,6 +3815,7 @@ void ProcessBase::route(
HttpEndpoint endpoint;
endpoint.realm = realm;
endpoint.authenticatedHandler = handler;
+ endpoint.options = options;
handlers.http[name.substr(1)] = endpoint;
[10/13] mesos git commit: Introduced a streaming request decoder in
libprocess.
Posted by an...@apache.org.
Introduced a streaming request decoder in libprocess.
This would become the default de facto decoder used by libprocess
and replace the existing `DataDecoder`.
Review: https://reviews.apache.org/r/53486/
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/32e203ea
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/32e203ea
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/32e203ea
Branch: refs/heads/master
Commit: 32e203ea194e3531ea8c5ded4d538bacb7cc2781
Parents: e8e3fe5
Author: Anand Mazumdar <an...@apache.org>
Authored: Mon Nov 21 18:07:41 2016 -0800
Committer: Anand Mazumdar <an...@apache.org>
Committed: Mon Nov 21 19:27:08 2016 -0800
----------------------------------------------------------------------
3rdparty/libprocess/src/decoder.hpp | 303 +++++++++++++++++++++++++++++++
1 file changed, 303 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/32e203ea/3rdparty/libprocess/src/decoder.hpp
----------------------------------------------------------------------
diff --git a/3rdparty/libprocess/src/decoder.hpp b/3rdparty/libprocess/src/decoder.hpp
index 0a60128..88ef1d1 100644
--- a/3rdparty/libprocess/src/decoder.hpp
+++ b/3rdparty/libprocess/src/decoder.hpp
@@ -740,6 +740,309 @@ private:
std::deque<http::Response*> responses;
};
+
+// Provides a request decoder that returns 'PIPE' requests once
+// the request headers are received, but before the body data
+// is received. Callers are expected to read the body from the
+// Pipe::Reader in the request.
+class StreamingRequestDecoder
+{
+public:
+ explicit StreamingRequestDecoder()
+ : failure(false), header(HEADER_FIELD), request(nullptr)
+ {
+ settings.on_message_begin =
+ &StreamingRequestDecoder::on_message_begin;
+ settings.on_url =
+ &StreamingRequestDecoder::on_url;
+ settings.on_header_field =
+ &StreamingRequestDecoder::on_header_field;
+ settings.on_header_value =
+ &StreamingRequestDecoder::on_header_value;
+ settings.on_headers_complete =
+ &StreamingRequestDecoder::on_headers_complete;
+ settings.on_body =
+ &StreamingRequestDecoder::on_body;
+ settings.on_message_complete =
+ &StreamingRequestDecoder::on_message_complete;
+ settings.on_chunk_complete =
+ &StreamingRequestDecoder::on_chunk_complete;
+ settings.on_chunk_header =
+ &StreamingRequestDecoder::on_chunk_header;
+
+ http_parser_init(&parser, HTTP_REQUEST);
+
+ parser.data = this;
+ }
+
+ ~StreamingRequestDecoder()
+ {
+ delete request;
+
+ if (writer.isSome()) {
+ writer->fail("Decoder is being deleted");
+ }
+
+ foreach (http::Request* request, requests) {
+ delete request;
+ }
+ }
+
+ std::deque<http::Request*> decode(const char* data, size_t length)
+ {
+ size_t parsed = http_parser_execute(&parser, &settings, data, length);
+ if (parsed != length) {
+ // TODO(bmahler): joyent/http-parser exposes error reasons.
+ failure = true;
+
+ // If we're still writing the body, fail the writer!
+ if (writer.isSome()) {
+ http::Pipe::Writer writer_ = writer.get(); // Remove const.
+ writer_.fail("failed to decode body");
+ writer = None();
+ }
+ }
+
+ if (!requests.empty()) {
+ std::deque<http::Request*> result = requests;
+ requests.clear();
+ return result;
+ }
+
+ return std::deque<http::Request*>();
+ }
+
+ bool failed() const
+ {
+ return failure;
+ }
+
+private:
+ static int on_message_begin(http_parser* p)
+ {
+ StreamingRequestDecoder* decoder = (StreamingRequestDecoder*) p->data;
+
+ CHECK(!decoder->failure);
+
+ decoder->header = HEADER_FIELD;
+ decoder->field.clear();
+ decoder->value.clear();
+ decoder->query.clear();
+ decoder->url.clear();
+
+ CHECK(decoder->request == nullptr);
+ CHECK_NONE(decoder->writer);
+
+ decoder->request = new http::Request();
+ decoder->request->type = http::Request::PIPE;
+ decoder->writer = None();
+ decoder->decompressor.reset();
+
+ return 0;
+ }
+
+ static int on_chunk_complete(http_parser* p)
+ {
+ return 0;
+ }
+
+ static int on_chunk_header(http_parser* p)
+ {
+ return 0;
+ }
+
+ static int on_url(http_parser* p, const char* data, size_t length)
+ {
+ StreamingRequestDecoder* decoder = (StreamingRequestDecoder*) p->data;
+
+ CHECK_NOTNULL(decoder->request);
+
+ // The current http_parser library (version 2.6.2 and below)
+ // does not support incremental parsing of URLs. To compensate
+ // we incrementally collect the data and parse it in
+ // `on_header_complete`.
+ decoder->url.append(data, length);
+
+ return 0;
+ }
+
+ static int on_header_field(http_parser* p, const char* data, size_t length)
+ {
+ StreamingRequestDecoder* decoder = (StreamingRequestDecoder*) p->data;
+
+ CHECK_NOTNULL(decoder->request);
+
+ if (decoder->header != HEADER_FIELD) {
+ decoder->request->headers[decoder->field] = decoder->value;
+ decoder->field.clear();
+ decoder->value.clear();
+ }
+
+ decoder->field.append(data, length);
+ decoder->header = HEADER_FIELD;
+
+ return 0;
+ }
+
+ static int on_header_value(http_parser* p, const char* data, size_t length)
+ {
+ StreamingRequestDecoder* decoder = (StreamingRequestDecoder*) p->data;
+
+ CHECK_NOTNULL(decoder->request);
+
+ decoder->value.append(data, length);
+ decoder->header = HEADER_VALUE;
+ return 0;
+ }
+
+ static int on_headers_complete(http_parser* p)
+ {
+ StreamingRequestDecoder* decoder = (StreamingRequestDecoder*) p->data;
+
+ CHECK_NOTNULL(decoder->request);
+
+ // Add final header.
+ decoder->request->headers[decoder->field] = decoder->value;
+ decoder->field.clear();
+ decoder->value.clear();
+
+ decoder->request->method =
+ http_method_str((http_method) decoder->parser.method);
+
+ decoder->request->keepAlive = http_should_keep_alive(&decoder->parser) != 0;
+
+ // Parse the URL. This data was incrementally built up during calls
+ // to `on_url`.
+ http_parser_url url;
+ http_parser_url_init(&url);
+ int parse_url =
+ http_parser_parse_url(decoder->url.data(), decoder->url.size(), 0, &url);
+
+ if (parse_url != 0) {
+ return parse_url;
+ }
+
+ if (url.field_set & (1 << UF_PATH)) {
+ decoder->request->url.path = std::string(
+ decoder->url.data() + url.field_data[UF_PATH].off,
+ url.field_data[UF_PATH].len);
+ }
+
+ if (url.field_set & (1 << UF_FRAGMENT)) {
+ decoder->request->url.fragment = std::string(
+ decoder->url.data() + url.field_data[UF_FRAGMENT].off,
+ url.field_data[UF_FRAGMENT].len);
+ }
+
+ if (url.field_set & (1 << UF_QUERY)) {
+ decoder->query = std::string(
+ decoder->url.data() + url.field_data[UF_QUERY].off,
+ url.field_data[UF_QUERY].len);
+ }
+
+ // Parse the query key/values.
+ Try<hashmap<std::string, std::string>> decoded =
+ http::query::decode(decoder->query);
+
+ if (decoded.isError()) {
+ return 1;
+ }
+
+ decoder->request->url.query = std::move(decoded.get());
+
+ Option<std::string> encoding =
+ decoder->request->headers.get("Content-Encoding");
+
+ if (encoding.isSome() && encoding.get() == "gzip") {
+ decoder->decompressor =
+ Owned<gzip::Decompressor>(new gzip::Decompressor());
+ }
+
+ CHECK_NONE(decoder->writer);
+
+ http::Pipe pipe;
+ decoder->writer = pipe.writer();
+ decoder->request->reader = pipe.reader();
+
+ // Send the request to the caller, but keep a Pipe::Writer for
+ // streaming the body content into the request.
+ decoder->requests.push_back(decoder->request);
+ decoder->request = nullptr;
+
+ return 0;
+ }
+
+ static int on_body(http_parser* p, const char* data, size_t length)
+ {
+ StreamingRequestDecoder* decoder = (StreamingRequestDecoder*) p->data;
+
+ CHECK_SOME(decoder->writer);
+
+ http::Pipe::Writer writer = decoder->writer.get(); // Remove const.
+
+ std::string body;
+ if (decoder->decompressor.get() != nullptr) {
+ Try<std::string> decompressed =
+ decoder->decompressor->decompress(std::string(data, length));
+
+ if (decompressed.isError()) {
+ return 1;
+ }
+
+ body = std::move(decompressed.get());
+ } else {
+ body = std::string(data, length);
+ }
+
+ writer.write(std::move(body));
+
+ return 0;
+ }
+
+ static int on_message_complete(http_parser* p)
+ {
+ StreamingRequestDecoder* decoder = (StreamingRequestDecoder*) p->data;
+
+ CHECK_SOME(decoder->writer);
+
+ http::Pipe::Writer writer = decoder->writer.get(); // Remove const.
+
+ if (decoder->decompressor.get() != nullptr &&
+ !decoder->decompressor->finished()) {
+ writer.fail("Failed to decompress body");
+ return 1;
+ }
+
+ writer.close();
+
+ decoder->writer = None();
+
+ return 0;
+ }
+
+ bool failure;
+
+ http_parser parser;
+ http_parser_settings settings;
+
+ enum
+ {
+ HEADER_FIELD,
+ HEADER_VALUE
+ } header;
+
+ std::string field;
+ std::string value;
+ std::string query;
+ std::string url;
+
+ http::Request* request;
+ Option<http::Pipe::Writer> writer;
+ Owned<gzip::Decompressor> decompressor;
+
+ std::deque<http::Request*> requests;
+};
+
} // namespace process {
#endif // __DECODER_HPP__
Re: [11/13] mesos git commit: Wired the libprocess code to use the
streaming decoder.
Posted by Benjamin Mahler <bm...@apache.org>.
The continuation needs to go through the process manager once again to
obtain a reference to the process that we'd like to deliver the message to.
One thing we could have done to avoid capturing 'this' is to grab the
ProcessReference early and pass it through to the continuation.
Unfortunately the assumption is that ProcessReferences are used quickly
after being obtained, since the termination code spins on this:
https://github.com/apache/mesos/blob/c89f571c7c81048ca83dc7098e35e847281ff672/3rdparty/libprocess/src/process.cpp#L3117-L3122
Another approach is to extend 'this' explicitly, benh has proposed
something like a "defer to 'this'" to me recently which would make
deferring to this safe, but it requires that the class extend from
enable_shared_from_this and will make us use shared pointers.
However, this newly introduced code which captures 'this' tries to fit in
with the existing pattern of libprocess finalization, where we need to
ensure that no further accesses to the process manager will occur before we
delete it (we don't leak the global object AFAIK):
https://github.com/apache/mesos/blob/c89f571c7c81048ca83dc7098e35e847281ff672/3rdparty/libprocess/src/process.cpp#L1189-L1252
Here we assume that finalization will shutdown all of the sockets before
deleting the process manager, which is the same assumption that is made in
the socket recv loops that access the process manager:
https://github.com/apache/mesos/blob/c89f571c7c81048ca83dc7098e35e847281ff672/3rdparty/libprocess/src/process.cpp#L802
We could have written it without a 'this' capture and instead use the
global 'process_manager', and then it becomes logically equivalent to the
approach in the recv loops:
parse(*request)
.onAny([socket, request](const Future<Message*>& message) {
...
process_manager->deliver(...);
// instead of
// deliver(...);
...
}
Note that there are additional accesses to process_manager as well that
need to be thought through in terms of finalization for any solution to
finalization. If you have some ideas to make the finalization more robust
that would be great!
On Tue, Nov 22, 2016 at 2:02 AM, Benjamin Bannier <
benjamin.bannier@mesosphere.io> wrote:
> Hi,
>
> just came across this with our `mesos-this-capture` clang-tidy check:
>
> > + // It is guaranteed that the continuation would run before the next
> > + // request arrives. Also, it's fine to pass the `this` pointer to
> the
> > + // continuation as this would get executed synchronously (if still
> pending)
> > + // from `SocketManager::finalize()` due to it closing all active
> sockets
> > + // during libprocess finalization.
> > + parse(*request)
> > + .onAny([this, socket, request](const Future<Message*>& future) {
>
> Even though there is a comment hinting that capturing `this` here should be
> safe, I am not sure this is a maintainable solution, e.g., still working
> if we
> begin to manage the lifetime of `process_manager` instead of simple
> leaking a
> global object. Above code would continue to compile in that case, but
> become racy.
>
> Is there some actor we could dispatch to to make this safe, or do we need
> a new abstraction?
>
>
> Cheers,
>
> Benjamin
>
>
Re: [11/13] mesos git commit: Wired the libprocess code to use the
streaming decoder.
Posted by Benjamin Bannier <be...@mesosphere.io>.
Hi,
just came across this with our `mesos-this-capture` clang-tidy check:
> + // It is guaranteed that the continuation would run before the next
> + // request arrives. Also, it's fine to pass the `this` pointer to the
> + // continuation as this would get executed synchronously (if still pending)
> + // from `SocketManager::finalize()` due to it closing all active sockets
> + // during libprocess finalization.
> + parse(*request)
> + .onAny([this, socket, request](const Future<Message*>& future) {
Even though there is a comment hinting that capturing `this` here should be
safe, I am not sure this is a maintainable solution, e.g., still working if we
begin to manage the lifetime of `process_manager` instead of simple leaking a
global object. Above code would continue to compile in that case, but become racy.
Is there some actor we could dispatch to to make this safe, or do we need a new abstraction?
Cheers,
Benjamin
[11/13] mesos git commit: Wired the libprocess code to use the
streaming decoder.
Posted by an...@apache.org.
Wired the libprocess code to use the streaming decoder.
Old libprocess style messages and routes not supporting request
streaming read the body from the piped reader. Otherwise, the
request is forwarded to the handler when the route supports
streaming.
Review: https://reviews.apache.org/r/53487/
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/d30039de
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/d30039de
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/d30039de
Branch: refs/heads/master
Commit: d30039ded0a434cdf9583e0a12b73e1b3661380e
Parents: c95e86c
Author: Anand Mazumdar <an...@apache.org>
Authored: Mon Nov 21 18:08:01 2016 -0800
Committer: Anand Mazumdar <an...@apache.org>
Committed: Mon Nov 21 19:27:08 2016 -0800
----------------------------------------------------------------------
3rdparty/libprocess/include/process/process.hpp | 6 +
3rdparty/libprocess/src/process.cpp | 365 ++++++++++---------
2 files changed, 205 insertions(+), 166 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/d30039de/3rdparty/libprocess/include/process/process.hpp
----------------------------------------------------------------------
diff --git a/3rdparty/libprocess/include/process/process.hpp b/3rdparty/libprocess/include/process/process.hpp
index d2ce4fe..de23f0c 100644
--- a/3rdparty/libprocess/include/process/process.hpp
+++ b/3rdparty/libprocess/include/process/process.hpp
@@ -459,6 +459,12 @@ private:
std::map<std::string, std::string> types;
};
+ // Continuation for `visit(const HttpEvent&)`.
+ Future<http::Response> _visit(
+ const HttpEndpoint& endpoint,
+ const std::string& name,
+ const Owned<http::Request>& request);
+
// Static assets(s) to provide.
std::map<std::string, Asset> assets;
http://git-wip-us.apache.org/repos/asf/mesos/blob/d30039de/3rdparty/libprocess/src/process.cpp
----------------------------------------------------------------------
diff --git a/3rdparty/libprocess/src/process.cpp b/3rdparty/libprocess/src/process.cpp
index 86628a6..d0c6eed 100644
--- a/3rdparty/libprocess/src/process.cpp
+++ b/3rdparty/libprocess/src/process.cpp
@@ -667,7 +667,26 @@ static bool libprocess(Request* request)
}
-static Message* parse(Request* request)
+// Returns a 'BODY' request once the body of the provided
+// 'PIPE' request can be read completely.
+static Future<Owned<Request>> convert(Owned<Request>&& pipeRequest)
+{
+ CHECK_EQ(Request::PIPE, pipeRequest->type);
+ CHECK_SOME(pipeRequest->reader);
+ CHECK(pipeRequest->body.empty());
+
+ return pipeRequest->reader->readAll()
+ .then([pipeRequest](const string& body) {
+ pipeRequest->type = Request::BODY;
+ pipeRequest->body = body;
+ pipeRequest->reader = None(); // Remove the reader.
+
+ return pipeRequest;;
+ });
+}
+
+
+static Future<Message*> parse(const Request& request)
{
// TODO(benh): Do better error handling (to deal with a malformed
// libprocess message, malicious or otherwise).
@@ -675,11 +694,11 @@ static Message* parse(Request* request)
// First try and determine 'from'.
Option<UPID> from = None();
- if (request->headers.contains("Libprocess-From")) {
- from = UPID(strings::trim(request->headers["Libprocess-From"]));
+ if (request.headers.contains("Libprocess-From")) {
+ from = UPID(strings::trim(request.headers.at("Libprocess-From")));
} else {
// Try and get 'from' from the User-Agent.
- const string& agent = request->headers["User-Agent"];
+ const string& agent = request.headers.at("User-Agent");
const string identifier = "libprocess/";
size_t index = agent.find(identifier);
if (index != string::npos) {
@@ -688,37 +707,42 @@ static Message* parse(Request* request)
}
if (from.isNone()) {
- return nullptr;
+ return Failure("Failed to determine sender from request headers");
}
// Now determine 'to'.
- size_t index = request->url.path.find('/', 1);
+ size_t index = request.url.path.find('/', 1);
index = index != string::npos ? index - 1 : string::npos;
// Decode possible percent-encoded 'to'.
- Try<string> decode = http::decode(request->url.path.substr(1, index));
+ Try<string> decode = http::decode(request.url.path.substr(1, index));
if (decode.isError()) {
- VLOG(2) << "Failed to decode URL path: " << decode.get();
- return nullptr;
+ return Failure("Failed to decode URL path: " + decode.get());
}
const UPID to(decode.get(), __address__);
// And now determine 'name'.
- index = index != string::npos ? index + 2: request->url.path.size();
- const string name = request->url.path.substr(index);
+ index = index != string::npos ? index + 2: request.url.path.size();
+ const string name = request.url.path.substr(index);
VLOG(2) << "Parsed message name '" << name
<< "' for " << to << " from " << from.get();
- Message* message = new Message();
- message->name = name;
- message->from = from.get();
- message->to = to;
- message->body = request->body;
+ CHECK_SOME(request.reader);
+ http::Pipe::Reader reader = request.reader.get(); // Remove const.
- return message;
+ return reader.readAll()
+ .then([from, name, to](const string& body) {
+ Message* message = new Message();
+ message->name = name;
+ message->from = from.get();
+ message->to = to;
+ message->body = body;
+
+ return message;
+ });
}
@@ -729,7 +753,7 @@ void decode_recv(
char* data,
size_t size,
Socket socket,
- DataDecoder* decoder)
+ StreamingRequestDecoder* decoder)
{
if (length.isDiscarded() || length.isFailed()) {
if (length.isFailed()) {
@@ -775,7 +799,7 @@ void decode_recv(
foreach (Request* request, requests) {
request->client = address.get();
- process_manager->handle(decoder->socket(), request);
+ process_manager->handle(socket, request);
}
}
@@ -835,7 +859,7 @@ void on_accept(const Future<Socket>& socket)
const size_t size = 80 * 1024;
char* data = new char[size];
- DataDecoder* decoder = new DataDecoder(socket.get());
+ StreamingRequestDecoder* decoder = new StreamingRequestDecoder();
socket.get().recv(data, size)
.onAny(lambda::bind(
@@ -2683,42 +2707,57 @@ void ProcessManager::handle(
// Check if this is a libprocess request (i.e., 'User-Agent:
// libprocess/id@ip:port') and if so, parse as a message.
if (libprocess(request)) {
- Message* message = parse(request);
- if (message != nullptr) {
- // TODO(benh): Use the sender PID when delivering in order to
- // capture happens-before timing relationships for testing.
- bool accepted = deliver(message->to, new MessageEvent(message));
-
- // Get the HttpProxy pid for this socket.
- PID<HttpProxy> proxy = socket_manager->proxy(socket);
-
- // Only send back an HTTP response if this isn't from libprocess
- // (which we determine by looking at the User-Agent). This is
- // necessary because older versions of libprocess would try and
- // recv the data and parse it as an HTTP request which would
- // fail thus causing the socket to get closed (but now
- // libprocess will ignore responses, see ignore_data).
- Option<string> agent = request->headers.get("User-Agent");
- if (agent.getOrElse("").find("libprocess/") == string::npos) {
- if (accepted) {
- VLOG(2) << "Accepted libprocess message to " << request->url.path;
- dispatch(proxy, &HttpProxy::enqueue, Accepted(), *request);
- } else {
- VLOG(1) << "Failed to handle libprocess message to "
- << request->url.path << ": not found";
- dispatch(proxy, &HttpProxy::enqueue, NotFound(), *request);
+ // It is guaranteed that the continuation would run before the next
+ // request arrives. Also, it's fine to pass the `this` pointer to the
+ // continuation as this would get executed synchronously (if still pending)
+ // from `SocketManager::finalize()` due to it closing all active sockets
+ // during libprocess finalization.
+ parse(*request)
+ .onAny([this, socket, request](const Future<Message*>& future) {
+ // Get the HttpProxy pid for this socket.
+ PID<HttpProxy> proxy = socket_manager->proxy(socket);
+
+ if (!future.isReady()) {
+ Response response = InternalServerError(
+ future.isFailed() ? future.failure() : "discarded future");
+
+ dispatch(proxy, &HttpProxy::enqueue, response, *request);
+
+ VLOG(1) << "Returning '" << response.status << "' for '"
+ << request->url.path << "': " << response.body;
+
+ delete request;
+ return;
}
- }
- delete request;
- return;
- }
+ Message* message = CHECK_NOTNULL(future.get());
+
+ // TODO(benh): Use the sender PID when delivering in order to
+ // capture happens-before timing relationships for testing.
+ bool accepted = deliver(message->to, new MessageEvent(message));
+
+ // Only send back an HTTP response if this isn't from libprocess
+ // (which we determine by looking at the User-Agent). This is
+ // necessary because older versions of libprocess would try and
+ // recv the data and parse it as an HTTP request which would
+ // fail thus causing the socket to get closed (but now
+ // libprocess will ignore responses, see ignore_data).
+ Option<string> agent = request->headers.get("User-Agent");
+ if (agent.getOrElse("").find("libprocess/") == string::npos) {
+ if (accepted) {
+ VLOG(2) << "Accepted libprocess message to " << request->url.path;
+ dispatch(proxy, &HttpProxy::enqueue, Accepted(), *request);
+ } else {
+ VLOG(1) << "Failed to handle libprocess message to "
+ << request->url.path << ": not found";
+ dispatch(proxy, &HttpProxy::enqueue, NotFound(), *request);
+ }
+ }
- VLOG(1) << "Failed to handle libprocess message: "
- << request->method << " " << request->url.path
- << " (User-Agent: " << request->headers["User-Agent"] << ")";
+ delete request;
+ return;
+ });
- delete request;
return;
}
@@ -3576,10 +3615,12 @@ void ProcessBase::visit(const HttpEvent& event)
handlers.httpSequence.reset(new Sequence("__auth_handlers__"));
}
- CHECK(event.request->url.path.find('/') == 0); // See ProcessManager::handle.
+ const string& path = event.request->url.path;
+
+ CHECK(path.find('/') == 0); // See ProcessManager::handle.
// Split the path by '/'.
- vector<string> tokens = strings::tokenize(event.request->url.path, "/");
+ vector<string> tokens = strings::tokenize(path, "/");
CHECK(!tokens.empty());
const string id = http::decode(tokens[0]).get();
@@ -3603,126 +3644,41 @@ void ProcessBase::visit(const HttpEvent& event)
continue;
}
- HttpEndpoint endpoint = handlers.http[name];
- Future<Option<AuthenticationResult>> authentication = None();
-
- if (endpoint.realm.isSome()) {
- authentication = authenticator_manager->authenticate(
- *event.request, endpoint.realm.get());
- }
-
- // Sequence the authentication future to ensure the handlers
- // are invoked in the same order that requests arrive.
- authentication = handlers.httpSequence->add<Option<AuthenticationResult>>(
- [authentication]() { return authentication; });
-
- Request request = *event.request;
- Promise<Response>* response = new Promise<Response>();
- event.response->associate(response->future());
-
- authentication
- .onAny(defer(self(), [this, endpoint, request, response, name, id](
- const Future<Option<AuthenticationResult>>& authentication) {
- if (!authentication.isReady()) {
- response->set(
- authentication.isFailed()
- ? ServiceUnavailable(authentication.failure())
- : ServiceUnavailable());
-
- VLOG(1) << "Returning '" << response->future()->status << "'"
- << " for '" << request.url.path << "'"
- << " (authentication failed: "
- << (authentication.isFailed()
- ? authentication.failure()
- : "discarded") << ")";
-
- delete response;
- return;
- }
-
- Option<string> principal = None();
-
- // If authentication failed, we do not continue with authorization.
- if (authentication->isSome()) {
- if (authentication.get()->unauthorized.isSome()) {
- // Request was not authenticated, challenged issued.
- response->set(authentication.get()->unauthorized.get());
-
- delete response;
- return;
- } else if (authentication.get()->forbidden.isSome()) {
- // Request was not authenticated, no challenge issued.
- response->set(authentication.get()->forbidden.get());
-
- delete response;
- return;
- }
-
- principal = authentication.get()->principal;
- }
+ const HttpEndpoint& endpoint = handlers.http[name];
- // The result of a call to an authorization callback.
- Future<bool> authorization;
+ Owned<Request> request(new Request(*event.request));
+ Future<Response> response;
- // Look for an authorization callback installed for this endpoint path.
- // If none is found, use a trivial one.
- const string callback_path = path::join("/" + id, name);
- if (authorization_callbacks != nullptr &&
- authorization_callbacks->count(callback_path) > 0) {
- authorization = authorization_callbacks->at(callback_path)(
- request, principal);
+ if (!endpoint.options.requestStreaming) {
+ // Consume the request body on behalf of the endpoint.
+ response = convert(std::move(request))
+ .then(defer(self(), [this, endpoint, name](
+ const Owned<Request>& request) {
+ return _visit(endpoint, name, request);
+ }));
+ } else {
+ response = _visit(endpoint, name, request);
+ }
- // Sequence the authorization future to ensure the handlers
- // are invoked in the same order that requests arrive.
- authorization = handlers.httpSequence->add<bool>(
- [authorization]() { return authorization; });
- } else {
- authorization = handlers.httpSequence->add<bool>(
- []() { return true; });
+ response
+ .onAny([path](const Future<Response>& response) {
+ if (!response.isReady()) {
+ VLOG(1) << "Failed to process request for '" << path << "': "
+ << (response.isFailed() ? response.failure() : "discarded");
}
+ });
- // Install a callback on the authorization result.
- authorization
- .onAny(defer(self(), [endpoint, request, response, principal](
- const Future<bool>& authorization) {
- if (!authorization.isReady()) {
- response->set(
- authorization.isFailed()
- ? ServiceUnavailable(authorization.failure())
- : ServiceUnavailable());
-
- VLOG(1) << "Returning '" << response->future()->status << "'"
- << " for '" << request.url.path << "'"
- << " (authorization failed: "
- << (authorization.isFailed()
- ? authorization.failure()
- : "discarded") << ")";
-
- delete response;
- return;
- }
-
- if (authorization.get() == true) {
- // Authorization succeeded, so forward request to the handler.
- if (endpoint.realm.isNone()) {
- response->associate(endpoint.handler.get()(request));
- } else {
- response->associate(endpoint.authenticatedHandler.get()(
- request, principal));
- }
- } else {
- // Authorization failed, so return a `Forbidden` response.
- response->set(Forbidden());
- }
-
- delete response;
- return;
- }));
- }));
-
+ event.response->associate(response);
return;
}
+ // Ensure the body is consumed so that no backpressure is applied
+ // to the socket (ignore the content since we do not care about it).
+ //
+ // TODO(anand): Is this an error?
+ CHECK_SOME(event.request->reader);
+ event.request->reader->readAll();
+
// If no HTTP handler is found look in assets.
name = tokens.size() > 1 ? tokens[1] : "";
@@ -3759,6 +3715,83 @@ void ProcessBase::visit(const HttpEvent& event)
}
+Future<Response> ProcessBase::_visit(
+ const HttpEndpoint& endpoint,
+ const string& name,
+ const Owned<Request>& request)
+{
+ Future<Option<AuthenticationResult>> authentication = None();
+
+ if (endpoint.realm.isSome()) {
+ authentication = authenticator_manager->authenticate(
+ *request, endpoint.realm.get());
+ }
+
+ // Sequence the authentication future to ensure the handlers
+ // are invoked in the same order that requests arrive.
+ authentication = handlers.httpSequence->add<Option<AuthenticationResult>>(
+ [authentication]() { return authentication; });
+
+ return authentication
+ .then(defer(self(), [this, endpoint, request, name](
+ const Option<AuthenticationResult>& authentication)
+ -> Future<Response> {
+ Option<string> principal = None();
+
+ // If authentication failed, we do not continue with authorization.
+ if (authentication.isSome()) {
+ if (authentication->unauthorized.isSome()) {
+ // Request was not authenticated, challenged issued.
+ return authentication->unauthorized.get();
+ } else if (authentication->forbidden.isSome()) {
+ // Request was not authenticated, no challenge issued.
+ return authentication->forbidden.get();
+ }
+
+ principal = authentication->principal;
+ }
+
+ // The result of a call to an authorization callback.
+ Future<bool> authorization;
+
+ // Look for an authorization callback installed for this endpoint path.
+ // If none is found, use a trivial one.
+ const string callback_path = path::join("/" + pid.id, name);
+ if (authorization_callbacks != nullptr &&
+ authorization_callbacks->count(callback_path) > 0) {
+ authorization = authorization_callbacks->at(callback_path)(
+ *request, principal);
+
+ // Sequence the authorization future to ensure the handlers
+ // are invoked in the same order that requests arrive.
+ authorization = handlers.httpSequence->add<bool>(
+ [authorization]() { return authorization; });
+ } else {
+ authorization = handlers.httpSequence->add<bool>(
+ []() { return true; });
+ }
+
+ // Install a callback on the authorization result.
+ return authorization
+ .then(defer(self(), [endpoint, request, principal](
+ bool authorization) -> Future<Response> {
+ if (authorization) {
+ // Authorization succeeded, so forward request to the handler.
+ if (endpoint.realm.isNone()) {
+ return endpoint.handler.get()(*request);
+ }
+
+ return endpoint.authenticatedHandler.get()(*request, principal);
+ }
+
+ // Authorization failed, so return a `Forbidden` response.
+ return Forbidden();
+ }
+ ));
+ }));
+}
+
+
void ProcessBase::visit(const ExitedEvent& event)
{
exited(event.pid);
[13/13] mesos git commit: Removed `convert()` continuations in favor
of using `readAll()`.
Posted by an...@apache.org.
Removed `convert()` continuations in favor of using `readAll()`.
Review: https://reviews.apache.org/r/53488/
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/d06d7456
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/d06d7456
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/d06d7456
Branch: refs/heads/master
Commit: d06d74562740767f0750e10907a327c5b45fef4c
Parents: d30039d
Author: Anand Mazumdar <an...@apache.org>
Authored: Mon Nov 21 18:08:12 2016 -0800
Committer: Anand Mazumdar <an...@apache.org>
Committed: Mon Nov 21 19:27:08 2016 -0800
----------------------------------------------------------------------
3rdparty/libprocess/src/http.cpp | 49 ++++++-----------------------------
1 file changed, 8 insertions(+), 41 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/d06d7456/3rdparty/libprocess/src/http.cpp
----------------------------------------------------------------------
diff --git a/3rdparty/libprocess/src/http.cpp b/3rdparty/libprocess/src/http.cpp
index ca7c948..0c2dfd3 100644
--- a/3rdparty/libprocess/src/http.cpp
+++ b/3rdparty/libprocess/src/http.cpp
@@ -937,56 +937,23 @@ string encode(const Request& request)
}
-// Forward declarations.
-Future<string> _convert(
- Pipe::Reader reader,
- const std::shared_ptr<string>& buffer,
- const string& read);
-Response __convert(
- const Response& pipeResponse,
- const string& body);
-
-
// Returns a 'BODY' response once the body of the provided
// 'PIPE' response can be read completely.
Future<Response> convert(const Response& pipeResponse)
{
- std::shared_ptr<string> buffer(new string());
-
CHECK_EQ(Response::PIPE, pipeResponse.type);
CHECK_SOME(pipeResponse.reader);
Pipe::Reader reader = pipeResponse.reader.get();
- return reader.read()
- .then(lambda::bind(&_convert, reader, buffer, lambda::_1))
- .then(lambda::bind(&__convert, pipeResponse, lambda::_1));
-}
-
-
-Future<string> _convert(
- Pipe::Reader reader,
- const std::shared_ptr<string>& buffer,
- const string& read)
-{
- if (read.empty()) { // EOF.
- return *buffer;
- }
-
- buffer->append(read);
-
- return reader.read()
- .then(lambda::bind(&_convert, reader, buffer, lambda::_1));
-}
-
-
-Response __convert(const Response& pipeResponse, const string& body)
-{
- Response bodyResponse = pipeResponse;
- bodyResponse.type = Response::BODY;
- bodyResponse.body = body;
- bodyResponse.reader = None(); // Remove the reader.
- return bodyResponse;
+ return reader.readAll()
+ .then([pipeResponse](const string& body) {
+ Response bodyResponse = pipeResponse;
+ bodyResponse.type = Response::BODY;
+ bodyResponse.body = body;
+ bodyResponse.reader = None(); // Remove the reader.
+ return bodyResponse;
+ });
}
[02/13] mesos git commit: Added support for request streaming to the
connection abstraction.
Posted by an...@apache.org.
Added support for request streaming to the connection abstraction.
This required modifications to the `encode()` method to return
a `Pipe::Reader` instead of the request body. The `send()` then
reads from this pipe to send the request via the socket.
Review: https://reviews.apache.org/r/53489/
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/a24cb498
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/a24cb498
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/a24cb498
Branch: refs/heads/master
Commit: a24cb4985c2333e2d15eeb8f971242f1754f81ab
Parents: d06d745
Author: Anand Mazumdar <an...@apache.org>
Authored: Mon Nov 21 18:08:18 2016 -0800
Committer: Anand Mazumdar <an...@apache.org>
Committed: Mon Nov 21 19:27:08 2016 -0800
----------------------------------------------------------------------
3rdparty/libprocess/src/http.cpp | 104 +++++++++++++++++++++++++++++++---
1 file changed, 97 insertions(+), 7 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/a24cb498/3rdparty/libprocess/src/http.cpp
----------------------------------------------------------------------
diff --git a/3rdparty/libprocess/src/http.cpp b/3rdparty/libprocess/src/http.cpp
index 0c2dfd3..3f16f29 100644
--- a/3rdparty/libprocess/src/http.cpp
+++ b/3rdparty/libprocess/src/http.cpp
@@ -866,7 +866,14 @@ ostream& operator<<(ostream& stream, const URL& url)
namespace internal {
-string encode(const Request& request)
+void _encode(Pipe::Reader reader, Pipe::Writer writer); // Forward declaration.
+
+
+// Encodes the request by writing into a pipe, the caller can
+// read the encoded data from the returned read end of the pipe.
+// A pipe is used since the request body can be a pipe and must
+// be read asynchronously.
+Pipe::Reader encode(const Request& request)
{
// TODO(bmahler): Replace this with a RequestEncoder.
std::ostringstream out;
@@ -915,8 +922,18 @@ string encode(const Request& request)
headers["Connection"] = "close";
}
- // Make sure the Content-Length is set correctly.
- headers["Content-Length"] = stringify(request.body.length());
+ if (request.type == Request::PIPE) {
+ CHECK(!headers.contains("Content-Length"));
+
+ // Make sure the Transfer-Encoding header is set correctly
+ // for PIPE requests.
+ headers["Transfer-Encoding"] = "chunked";
+ } else {
+ CHECK_EQ(Request::BODY, request.type);
+
+ // Make sure the Content-Length header is set correctly.
+ headers["Content-Length"] = stringify(request.body.length());
+ }
// TODO(bmahler): Use a 'Request' and a 'RequestEncoder' here!
// Currently this does not handle 'gzip' content encoding,
@@ -931,9 +948,54 @@ string encode(const Request& request)
}
out << "\r\n";
- out << request.body;
- return out.str();
+ Pipe pipe;
+ Pipe::Reader reader = pipe.reader();
+ Pipe::Writer writer = pipe.writer();
+
+ // Write the head of the request.
+ writer.write(out.str());
+
+ switch (request.type) {
+ case Request::BODY:
+ writer.write(request.body);
+ writer.close();
+ break;
+ case Request::PIPE:
+ CHECK_SOME(request.reader);
+ CHECK(request.body.empty());
+ _encode(request.reader.get(), writer);
+ break;
+ }
+
+ return reader;
+}
+
+
+void _encode(Pipe::Reader reader, Pipe::Writer writer)
+{
+ reader.read()
+ .onAny([reader, writer](const Future<string>& chunk) mutable {
+ if (!chunk.isReady()) {
+ writer.fail(chunk.isFailed() ? chunk.failure() : "discarded");
+ return;
+ }
+
+ if (chunk->empty()) {
+ // EOF case.
+ writer.write("0\r\n\r\n");
+ writer.close();
+ return;
+ }
+
+ std::ostringstream out;
+ out << std::hex << chunk->size() << "\r\n";
+ out << chunk.get();
+ out << "\r\n";
+
+ writer.write(out.str());
+ _encode(reader, writer);
+ });
}
@@ -976,6 +1038,21 @@ public:
return Failure("Cannot pipeline after 'Connection: close'");
}
+ if (request.type == Request::PIPE) {
+ if (request.reader.isNone()) {
+ return Failure("Request reader must be set for PIPE request");
+ }
+
+ if (!request.body.empty()) {
+ return Failure("Request body must be empty for PIPE request");
+ }
+
+ Option<string> contentLength = request.headers.get("Content-Length");
+ if (request.headers.contains("Content-Length")) {
+ return Failure("'Content-Length' cannot be set for PIPE request");
+ }
+ }
+
if (!request.keepAlive) {
close = true;
}
@@ -985,8 +1062,8 @@ public:
Socket socket_ = socket;
sendChain = sendChain
- .then([socket_, request]() mutable {
- return socket_.send(encode(request));
+ .then([socket_, request]() {
+ return _send(socket_, encode(request));
});
// If we can't write to the socket, disconnect.
@@ -1045,6 +1122,19 @@ protected:
}
private:
+ static Future<Nothing> _send(Socket socket, Pipe::Reader reader)
+ {
+ return reader.read()
+ .then([socket, reader](const string& data) mutable -> Future<Nothing> {
+ if (data.empty()) {
+ return Nothing(); // EOF.
+ }
+
+ return socket.send(data)
+ .then(lambda::bind(_send, socket, reader));
+ });
+ }
+
void read()
{
socket.recv()
[09/13] mesos git commit: Parameterized existing decoder tests on the
type of decoder.
Posted by an...@apache.org.
Parameterized existing decoder tests on the type of decoder.
This allows us to not duplicate tests for the streaming request
decoder.
Review: https://reviews.apache.org/r/53511/
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/c95e86c3
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/c95e86c3
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/c95e86c3
Branch: refs/heads/master
Commit: c95e86c37dd82130016abf3a240ebfd869dfed2c
Parents: dceacc5
Author: Anand Mazumdar <an...@apache.org>
Authored: Mon Nov 21 18:07:54 2016 -0800
Committer: Anand Mazumdar <an...@apache.org>
Committed: Mon Nov 21 19:27:08 2016 -0800
----------------------------------------------------------------------
3rdparty/libprocess/src/tests/decoder_tests.cpp | 46 ++++++++++++++------
1 file changed, 33 insertions(+), 13 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/c95e86c3/3rdparty/libprocess/src/tests/decoder_tests.cpp
----------------------------------------------------------------------
diff --git a/3rdparty/libprocess/src/tests/decoder_tests.cpp b/3rdparty/libprocess/src/tests/decoder_tests.cpp
index d4194a6..5f84d84 100644
--- a/3rdparty/libprocess/src/tests/decoder_tests.cpp
+++ b/3rdparty/libprocess/src/tests/decoder_tests.cpp
@@ -15,6 +15,7 @@
#include <deque>
#include <string>
+#include <process/gtest.hpp>
#include <process/owned.hpp>
#include <process/socket.hpp>
@@ -28,18 +29,33 @@ using process::DataDecoder;
using process::Future;
using process::Owned;
using process::ResponseDecoder;
+using process::StreamingRequestDecoder;
using process::StreamingResponseDecoder;
+using process::http::Request;
+
using process::network::Socket;
using std::deque;
using std::string;
-TEST(DecoderTest, Request)
+// TODO(anand): Parameterize the response decoder tests.
+
+template <typename T>
+class RequestDecoderTest : public ::testing::Test {};
+
+
+typedef ::testing::Types<DataDecoder, StreamingRequestDecoder>
+ RequestDecoderTypes;
+
+
+// The request decoder tests are parameterized by the type of request decoder.
+TYPED_TEST_CASE(RequestDecoderTest, RequestDecoderTypes);
+
+
+TYPED_TEST(RequestDecoderTest, Request)
{
- Try<Socket> socket = Socket::create();
- ASSERT_SOME(socket);
- DataDecoder decoder = DataDecoder(socket.get());
+ TypeParam decoder;
const string data =
"GET /path/file.json?key1=value1&key2=value2#fragment HTTP/1.1\r\n"
@@ -61,7 +77,15 @@ TEST(DecoderTest, Request)
EXPECT_SOME_EQ("value1", request->url.query.get("key1"));
EXPECT_SOME_EQ("value2", request->url.query.get("key2"));
- EXPECT_TRUE(request->body.empty());
+ Future<string> body = [&request]() -> Future<string> {
+ if (request->type == Request::BODY) {
+ return request->body;
+ }
+
+ return request->reader->readAll();
+ }();
+
+ AWAIT_EXPECT_EQ(string(""), body);
EXPECT_FALSE(request->keepAlive);
EXPECT_EQ(3u, request->headers.size());
@@ -71,11 +95,9 @@ TEST(DecoderTest, Request)
}
-TEST(DecoderTest, RequestHeaderContinuation)
+TYPED_TEST(RequestDecoderTest, HeaderContinuation)
{
- Try<Socket> socket = Socket::create();
- ASSERT_SOME(socket);
- DataDecoder decoder = DataDecoder(socket.get());
+ TypeParam decoder;
const string data =
"GET /path/file.json HTTP/1.1\r\n"
@@ -95,11 +117,9 @@ TEST(DecoderTest, RequestHeaderContinuation)
}
-TEST(DecoderTest, RequestHeaderCaseInsensitive)
+TYPED_TEST(RequestDecoderTest, HeaderCaseInsensitive)
{
- Try<Socket> socket = Socket::create();
- ASSERT_SOME(socket);
- DataDecoder decoder = DataDecoder(socket.get());
+ TypeParam decoder;
const string data =
"GET /path/file.json HTTP/1.1\r\n"
[05/13] mesos git commit: Initialized the POD type in the `Request`
struct.
Posted by an...@apache.org.
Initialized the POD type in the `Request` struct.
Previously, the `keepAlive` member was not initialized correctly,
the behavior is undefined if POD types are not correctly
initialized.
Review: https://reviews.apache.org/r/53482/
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/1003f3d2
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/1003f3d2
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/1003f3d2
Branch: refs/heads/master
Commit: 1003f3d208f6e06e8bf485e395190f9bd4e5fe24
Parents: 6a4de74
Author: Anand Mazumdar <an...@apache.org>
Authored: Mon Nov 21 18:07:11 2016 -0800
Committer: Anand Mazumdar <an...@apache.org>
Committed: Mon Nov 21 19:27:08 2016 -0800
----------------------------------------------------------------------
3rdparty/libprocess/include/process/http.hpp | 11 +++++++++++
1 file changed, 11 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/1003f3d2/3rdparty/libprocess/include/process/http.hpp
----------------------------------------------------------------------
diff --git a/3rdparty/libprocess/include/process/http.hpp b/3rdparty/libprocess/include/process/http.hpp
index 5f7c0f6..51bfc7e 100644
--- a/3rdparty/libprocess/include/process/http.hpp
+++ b/3rdparty/libprocess/include/process/http.hpp
@@ -413,6 +413,9 @@ private:
struct Request
{
+ Request()
+ : keepAlive(false) {}
+
std::string method;
// TODO(benh): Add major/minor version.
@@ -430,6 +433,14 @@ struct Request
// TODO(bmahler): Ensure this is consistent with the 'Connection'
// header; perhaps make this a function that checks the header.
+ //
+ // TODO(anand): Ideally, this could default to 'true' since
+ // persistent connections are the default since HTTP 1.1.
+ // Perhaps, we need to go from `keepAlive` to `closeConnection`
+ // to reflect the header more accurately, and to have an
+ // intuitive default of false.
+ //
+ // Default: false.
bool keepAlive;
// For server requests, this contains the address of the client.
[03/13] mesos git commit: Removed extraneous socket argument from
`DataDecoder` constructor.
Posted by an...@apache.org.
Removed extraneous socket argument from `DataDecoder` constructor.
This argument is not used anywhere in the code. This makes it
consistent with the streaming request decoder.
Review: https://reviews.apache.org/r/53510/
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/dceacc50
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/dceacc50
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/dceacc50
Branch: refs/heads/master
Commit: dceacc50ce8577b1d0fd5cde0e55dadef1907fdf
Parents: 32e203e
Author: Anand Mazumdar <an...@apache.org>
Authored: Mon Nov 21 18:07:47 2016 -0800
Committer: Anand Mazumdar <an...@apache.org>
Committed: Mon Nov 21 19:27:08 2016 -0800
----------------------------------------------------------------------
3rdparty/libprocess/src/decoder.hpp | 12 ++----------
1 file changed, 2 insertions(+), 10 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/dceacc50/3rdparty/libprocess/src/decoder.hpp
----------------------------------------------------------------------
diff --git a/3rdparty/libprocess/src/decoder.hpp b/3rdparty/libprocess/src/decoder.hpp
index 88ef1d1..13db2ee 100644
--- a/3rdparty/libprocess/src/decoder.hpp
+++ b/3rdparty/libprocess/src/decoder.hpp
@@ -23,7 +23,6 @@
#include <vector>
#include <process/http.hpp>
-#include <process/socket.hpp>
#include <stout/foreach.hpp>
#include <stout/gzip.hpp>
@@ -43,8 +42,8 @@ namespace process {
class DataDecoder
{
public:
- explicit DataDecoder(const network::Socket& _s)
- : s(_s), failure(false), request(nullptr)
+ DataDecoder()
+ : failure(false), request(nullptr)
{
settings.on_message_begin = &DataDecoder::on_message_begin;
settings.on_url = &DataDecoder::on_url;
@@ -93,11 +92,6 @@ public:
return failure;
}
- network::Socket socket() const
- {
- return s;
- }
-
private:
static int on_message_begin(http_parser* p)
{
@@ -262,8 +256,6 @@ private:
return 0;
}
- const network::Socket s; // The socket this decoder is associated with.
-
bool failure;
http_parser parser;
[04/13] mesos git commit: Moved around the `Request` struct in the
file.
Posted by an...@apache.org.
Moved around the `Request` struct in the file.
This serves two purposes:
- This brings it near the `Response` struct thereby making it easy
to switch between them for readability.
- In the future, `Request` would have a `Pipe::Reader` member
for supporting request streaming. Otherwise, we would need to
add another forward declaration and forward declaring nested
structs is not quite possible.
Review: https://reviews.apache.org/r/53481/
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/6a4de740
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/6a4de740
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/6a4de740
Branch: refs/heads/master
Commit: 6a4de74007604862e3c24b022ab565c8a9c6cdc9
Parents: 69efd9a
Author: Anand Mazumdar <an...@apache.org>
Authored: Mon Nov 21 18:07:06 2016 -0800
Committer: Anand Mazumdar <an...@apache.org>
Committed: Mon Nov 21 19:27:08 2016 -0800
----------------------------------------------------------------------
3rdparty/libprocess/include/process/http.hpp | 78 +++++++++++------------
1 file changed, 39 insertions(+), 39 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/6a4de740/3rdparty/libprocess/include/process/http.hpp
----------------------------------------------------------------------
diff --git a/3rdparty/libprocess/include/process/http.hpp b/3rdparty/libprocess/include/process/http.hpp
index 9722c62..5f7c0f6 100644
--- a/3rdparty/libprocess/include/process/http.hpp
+++ b/3rdparty/libprocess/include/process/http.hpp
@@ -257,45 +257,6 @@ typedef hashmap<std::string,
CaseInsensitiveEqual> Headers;
-struct Request
-{
- std::string method;
-
- // TODO(benh): Add major/minor version.
-
- // For client requests, the URL should be a URI.
- // For server requests, the URL may be a URI or a relative reference.
- URL url;
-
- Headers headers;
-
- // TODO(bmahler): Add a 'query' field which contains both
- // the URL query and the parsed form data from the body.
-
- std::string body;
-
- // TODO(bmahler): Ensure this is consistent with the 'Connection'
- // header; perhaps make this a function that checks the header.
- bool keepAlive;
-
- // For server requests, this contains the address of the client.
- // Note that this may correspond to a proxy or load balancer address.
- network::Address client;
-
- /**
- * Returns whether the encoding is considered acceptable in the
- * response. See RFC 2616 section 14.3 for details.
- */
- bool acceptsEncoding(const std::string& encoding) const;
-
- /**
- * Returns whether the media type is considered acceptable in the
- * response. See RFC 2616, section 14.1 for the details.
- */
- bool acceptsMediaType(const std::string& mediaType) const;
-};
-
-
// Represents an asynchronous in-memory unbuffered Pipe, currently
// used for streaming HTTP responses via chunked encoding. Note that
// being an in-memory pipe means that this cannot be used across OS
@@ -450,6 +411,45 @@ private:
};
+struct Request
+{
+ std::string method;
+
+ // TODO(benh): Add major/minor version.
+
+ // For client requests, the URL should be a URI.
+ // For server requests, the URL may be a URI or a relative reference.
+ URL url;
+
+ Headers headers;
+
+ // TODO(bmahler): Add a 'query' field which contains both
+ // the URL query and the parsed form data from the body.
+
+ std::string body;
+
+ // TODO(bmahler): Ensure this is consistent with the 'Connection'
+ // header; perhaps make this a function that checks the header.
+ bool keepAlive;
+
+ // For server requests, this contains the address of the client.
+ // Note that this may correspond to a proxy or load balancer address.
+ network::Address client;
+
+ /**
+ * Returns whether the encoding is considered acceptable in the
+ * response. See RFC 2616 section 14.3 for details.
+ */
+ bool acceptsEncoding(const std::string& encoding) const;
+
+ /**
+ * Returns whether the media type is considered acceptable in the
+ * response. See RFC 2616, section 14.1 for the details.
+ */
+ bool acceptsMediaType(const std::string& mediaType) const;
+};
+
+
struct Response
{
Response()
[12/13] mesos git commit: Added a test for request streaming and GZIP
compression.
Posted by an...@apache.org.
Added a test for request streaming and GZIP compression.
These tests validate that clients can stream requests and send
compressed GZIP requests using the connection abstraction. This
also test the implementation of the streaming decoder indirectly.
Review: https://reviews.apache.org/r/53490/
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/5cd134fe
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/5cd134fe
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/5cd134fe
Branch: refs/heads/master
Commit: 5cd134fedf69900523f3088de32daae81f216437
Parents: a24cb49
Author: Anand Mazumdar <an...@apache.org>
Authored: Mon Nov 21 18:08:30 2016 -0800
Committer: Anand Mazumdar <an...@apache.org>
Committed: Mon Nov 21 19:27:08 2016 -0800
----------------------------------------------------------------------
3rdparty/libprocess/src/tests/http_tests.cpp | 118 ++++++++++++++++++++++
1 file changed, 118 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/5cd134fe/3rdparty/libprocess/src/tests/http_tests.cpp
----------------------------------------------------------------------
diff --git a/3rdparty/libprocess/src/tests/http_tests.cpp b/3rdparty/libprocess/src/tests/http_tests.cpp
index 10d3fb0..d41929a 100644
--- a/3rdparty/libprocess/src/tests/http_tests.cpp
+++ b/3rdparty/libprocess/src/tests/http_tests.cpp
@@ -21,6 +21,7 @@
#include <netinet/tcp.h>
#endif // __WINDOWS__
+#include <algorithm>
#include <string>
#include <vector>
@@ -86,6 +87,7 @@ public:
MOCK_METHOD1(requestDelete, Future<http::Response>(const http::Request&));
MOCK_METHOD1(a, Future<http::Response>(const http::Request&));
MOCK_METHOD1(abc, Future<http::Response>(const http::Request&));
+ MOCK_METHOD1(requestStreaming, Future<http::Response>(const http::Request&));
MOCK_METHOD2(
authenticated,
@@ -103,6 +105,12 @@ protected:
route("/a", None(), &HttpProcess::a);
route("/a/b/c", None(), &HttpProcess::abc);
route("/authenticated", "realm", None(), &HttpProcess::authenticated);
+
+ // Route accepting a streaming request.
+ RouteOptions options;
+ options.requestStreaming = true;
+
+ route("/requeststreaming", None(), &HttpProcess::requestStreaming, options);
}
};
@@ -832,6 +840,51 @@ TEST(HTTPTest, Request)
}
+// This test verifies that the server can correctly receive the
+// uncompressed data from the request.
+TEST(HTTPConnectionTest, GzipRequestBody)
+{
+ Http http;
+
+ http::URL url = http::URL(
+ "http",
+ http.process->self().address.ip,
+ http.process->self().address.port,
+ http.process->self().id + "/body");
+
+ Future<http::Connection> connect = http::connect(url);
+ AWAIT_READY(connect);
+
+ http::Connection connection = connect.get();
+
+ Promise<http::Response> promise;
+ Future<http::Request> expected;
+
+ EXPECT_CALL(*http.process, body(_))
+ .WillOnce(DoAll(FutureArg<0>(&expected), Return(promise.future())));
+
+ string uncompressed = "Hello World";
+
+ http::Request request;
+ request.method = "POST";
+ request.url = url;
+ request.body = gzip::compress(uncompressed).get();
+ request.keepAlive = true;
+
+ request.headers["Content-Encoding"] = "gzip";
+ request.headers["Content-Length"] = request.body.length();
+
+ Future<http::Response> response = connection.send(request);
+
+ AWAIT_READY(expected);
+ EXPECT_EQ(uncompressed, expected->body);
+
+ // Disconnect.
+ AWAIT_READY(connection.disconnect());
+ AWAIT_READY(connection.disconnected());
+}
+
+
TEST(HTTPConnectionTest, Serial)
{
Http http;
@@ -1175,6 +1228,71 @@ TEST(HTTPConnectionTest, Equality)
}
+// This test verifies that we can stream the request body using the
+// connection abstraction to a streaming enabled route.
+TEST(HTTPConnectionTest, RequestStreaming)
+{
+ Http http;
+
+ http::URL url = http::URL(
+ "http",
+ http.process->self().address.ip,
+ http.process->self().address.port,
+ http.process->self().id + "/requeststreaming");
+
+ Future<http::Connection> connect = http::connect(url);
+ AWAIT_READY(connect);
+
+ http::Connection connection = connect.get();
+
+ Promise<http::Response> promise;
+ Future<http::Request> expected;
+
+ EXPECT_CALL(*http.process, requestStreaming(_))
+ .WillOnce(DoAll(FutureArg<0>(&expected), Return(promise.future())));
+
+ http::Pipe pipe;
+
+ http::Request request;
+ request.method = "POST";
+ request.url = url;
+ request.type = http::Request::PIPE;
+ request.reader = pipe.reader();
+ request.keepAlive = true;
+
+ Future<http::Response> response = connection.send(request);
+
+ AWAIT_READY(expected);
+ ASSERT_EQ(http::Request::PIPE, expected->type);
+ ASSERT_SOME(expected->reader);
+
+ http::Pipe::Reader reader = expected->reader.get();
+
+ // Start streaming the request body.
+ pipe.writer().write("Hello");
+
+ string read;
+ while (read != "Hello") {
+ Future<string> future = reader.read();
+ AWAIT_READY(future);
+
+ read.append(future.get());
+ ASSERT_TRUE(std::equal(read.begin(), read.end(), "Hello"));
+ }
+
+ pipe.writer().close();
+
+ promise.set(http::OK("1"));
+
+ AWAIT_EXPECT_RESPONSE_STATUS_EQ(http::OK().status, response);
+ EXPECT_EQ("1", response->body);
+
+ // Disconnect.
+ AWAIT_READY(connection.disconnect());
+ AWAIT_READY(connection.disconnected());
+}
+
+
// TODO(hausdorff): This test seems to create inconsistent (though not
// incorrect) results across platforms. Fix and enable the test on Windows. In
// particular, the encoding in the 3rd example puts the first variable into the