Repository: mesos Updated Branches: refs/heads/master 9b14ba2e4 -> a0d201b71
Re-ordered decoder callbacks to be in execution order. Review: https://reviews.apache.org/r/32343 Project: http://git-wip-us.apache.org/repos/asf/mesos/repo Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/595e0615 Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/595e0615 Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/595e0615 Branch: refs/heads/master Commit: 595e0615e8529637eb4a6328c7e09729efe932ec Parents: 2622b01 Author: Benjamin Mahler <[email protected]> Authored: Fri Mar 20 13:58:41 2015 -0700 Committer: Benjamin Mahler <[email protected]> Committed: Tue Mar 24 16:47:18 2015 -0700 ---------------------------------------------------------------------- 3rdparty/libprocess/src/decoder.hpp | 260 ++++++++++++++++--------------- 1 file changed, 131 insertions(+), 129 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/mesos/blob/595e0615/3rdparty/libprocess/src/decoder.hpp ---------------------------------------------------------------------- diff --git a/3rdparty/libprocess/src/decoder.hpp b/3rdparty/libprocess/src/decoder.hpp index 25b0228..4b3c53f 100644 --- a/3rdparty/libprocess/src/decoder.hpp +++ b/3rdparty/libprocess/src/decoder.hpp @@ -29,12 +29,6 @@ public: : s(_s), failure(false), request(NULL) { settings.on_message_begin = &DataDecoder::on_message_begin; - settings.on_header_field = &DataDecoder::on_header_field; - settings.on_header_value = &DataDecoder::on_header_value; - settings.on_url = &DataDecoder::on_url; - settings.on_body = &DataDecoder::on_body; - settings.on_headers_complete = &DataDecoder::on_headers_complete; - settings.on_message_complete = &DataDecoder::on_message_complete; #if !(HTTP_PARSER_VERSION_MAJOR >= 2) settings.on_path = &DataDecoder::on_path; @@ -42,6 +36,13 @@ public: settings.on_query_string = &DataDecoder::on_query_string; #endif + settings.on_url = &DataDecoder::on_url; + settings.on_header_field = &DataDecoder::on_header_field; + settings.on_header_value = &DataDecoder::on_header_value; + settings.on_headers_complete = &DataDecoder::on_headers_complete; + settings.on_body = &DataDecoder::on_body; + settings.on_message_complete = &DataDecoder::on_message_complete; + http_parser_init(&parser, HTTP_REQUEST); parser.data = this; @@ -100,82 +101,31 @@ private: return 0; } - static int on_headers_complete(http_parser* p) - { - DataDecoder* decoder = (DataDecoder*) p->data; - - // 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); - - return 0; - } - - static int on_message_complete(http_parser* p) +#if !(HTTP_PARSER_VERSION_MAJOR >= 2) + static int on_path(http_parser* p, const char* data, size_t length) { DataDecoder* decoder = (DataDecoder*) p->data; -// std::cout << "http::Request:" << std::endl; -// std::cout << " method: " << decoder->request->method << std::endl; -// std::cout << " path: " << decoder->request->path << std::endl; - - // Parse the query key/values. - Try<hashmap<std::string, std::string>> decoded = - http::query::decode(decoder->query); - - if (decoded.isError()) { - return 1; - } - - decoder->request->query = decoded.get(); - - Option<std::string> encoding = - decoder->request->headers.get("Content-Encoding"); - if (encoding.isSome() && encoding.get() == "gzip") { - Try<std::string> decompressed = gzip::decompress(decoder->request->body); - if (decompressed.isError()) { - return 1; - } - decoder->request->body = decompressed.get(); - decoder->request->headers["Content-Length"] = - decoder->request->body.length(); - } - - decoder->requests.push_back(decoder->request); - decoder->request = NULL; + assert(decoder->request != NULL); + decoder->request->path.append(data, length); return 0; } - static int on_header_field(http_parser* p, const char* data, size_t length) + static int on_query_string(http_parser* p, const char* data, size_t length) { DataDecoder* decoder = (DataDecoder*) p->data; assert(decoder->request != NULL); - - 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; - + decoder->query.append(data, length); return 0; } - static int on_header_value(http_parser* p, const char* data, size_t length) + static int on_fragment(http_parser* p, const char* data, size_t length) { DataDecoder* decoder = (DataDecoder*) p->data; assert(decoder->request != NULL); - decoder->value.append(data, length); - decoder->header = HEADER_VALUE; + decoder->request->fragment.append(data, length); return 0; } +#endif // !(HTTP_PARSER_VERSION_MAJOR >= 2) static int on_url(http_parser* p, const char* data, size_t length) { @@ -213,31 +163,48 @@ private: return result; } -#if !(HTTP_PARSER_VERSION_MAJOR >= 2) - static int on_path(http_parser* p, const char* data, size_t length) + static int on_header_field(http_parser* p, const char* data, size_t length) { DataDecoder* decoder = (DataDecoder*) p->data; assert(decoder->request != NULL); - decoder->request->path.append(data, length); + + 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_query_string(http_parser* p, const char* data, size_t length) + static int on_header_value(http_parser* p, const char* data, size_t length) { DataDecoder* decoder = (DataDecoder*) p->data; assert(decoder->request != NULL); - decoder->query.append(data, length); + decoder->value.append(data, length); + decoder->header = HEADER_VALUE; return 0; } - static int on_fragment(http_parser* p, const char* data, size_t length) + static int on_headers_complete(http_parser* p) { DataDecoder* decoder = (DataDecoder*) p->data; - assert(decoder->request != NULL); - decoder->request->fragment.append(data, length); + + // 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); + return 0; } -#endif // !(HTTP_PARSER_VERSION_MAJOR >= 2) static int on_body(http_parser* p, const char* data, size_t length) { @@ -247,6 +214,40 @@ private: return 0; } + static int on_message_complete(http_parser* p) + { + DataDecoder* decoder = (DataDecoder*) p->data; +// std::cout << "http::Request:" << std::endl; +// std::cout << " method: " << decoder->request->method << std::endl; +// std::cout << " path: " << decoder->request->path << std::endl; + + // Parse the query key/values. + Try<hashmap<std::string, std::string>> decoded = + http::query::decode(decoder->query); + + if (decoded.isError()) { + return 1; + } + + decoder->request->query = decoded.get(); + + Option<std::string> encoding = + decoder->request->headers.get("Content-Encoding"); + if (encoding.isSome() && encoding.get() == "gzip") { + Try<std::string> decompressed = gzip::decompress(decoder->request->body); + if (decompressed.isError()) { + return 1; + } + decoder->request->body = decompressed.get(); + decoder->request->headers["Content-Length"] = + decoder->request->body.length(); + } + + decoder->requests.push_back(decoder->request); + decoder->request = NULL; + return 0; + } + const network::Socket s; // The socket this decoder is associated with. bool failure; @@ -276,12 +277,6 @@ public: : failure(false), header(HEADER_FIELD), response(NULL) { settings.on_message_begin = &ResponseDecoder::on_message_begin; - settings.on_header_field = &ResponseDecoder::on_header_field; - settings.on_header_value = &ResponseDecoder::on_header_value; - settings.on_url = &ResponseDecoder::on_url; - settings.on_body = &ResponseDecoder::on_body; - settings.on_headers_complete = &ResponseDecoder::on_headers_complete; - settings.on_message_complete = &ResponseDecoder::on_message_complete; #if !(HTTP_PARSER_VERSION_MAJOR >=2) settings.on_path = &ResponseDecoder::on_path; @@ -289,6 +284,13 @@ public: settings.on_query_string = &ResponseDecoder::on_query_string; #endif + settings.on_url = &ResponseDecoder::on_url; + settings.on_header_field = &ResponseDecoder::on_header_field; + settings.on_header_value = &ResponseDecoder::on_header_value; + settings.on_headers_complete = &ResponseDecoder::on_headers_complete; + settings.on_body = &ResponseDecoder::on_body; + settings.on_message_complete = &ResponseDecoder::on_message_complete; + http_parser_init(&parser, HTTP_RESPONSE); parser.data = this; @@ -339,46 +341,25 @@ private: return 0; } - static int on_headers_complete(http_parser* p) +#if !(HTTP_PARSER_VERSION_MAJOR >= 2) + static int on_path(http_parser* p, const char* data, size_t length) { - ResponseDecoder* decoder = (ResponseDecoder*) p->data; - - // Add final header. - decoder->response->headers[decoder->field] = decoder->value; - decoder->field.clear(); - decoder->value.clear(); - return 0; } - static int on_message_complete(http_parser* p) + static int on_query_string(http_parser* p, const char* data, size_t length) { - ResponseDecoder* decoder = (ResponseDecoder*) p->data; - - // Get the response status string. - if (http::statuses.contains(decoder->parser.status_code)) { - decoder->response->status = http::statuses[decoder->parser.status_code]; - } else { - decoder->failure = true; - return 1; - } + return 0; + } - // We can only provide the gzip encoding. - Option<std::string> encoding = - decoder->response->headers.get("Content-Encoding"); - if (encoding.isSome() && encoding.get() == "gzip") { - Try<std::string> decompressed = gzip::decompress(decoder->response->body); - if (decompressed.isError()) { - decoder->failure = true; - return 1; - } - decoder->response->body = decompressed.get(); - decoder->response->headers["Content-Length"] = - decoder->response->body.length(); - } + static int on_fragment(http_parser* p, const char* data, size_t length) + { + return 0; + } +#endif // !(HTTP_PARSER_VERSION_MAJOR >= 2) - decoder->responses.push_back(decoder->response); - decoder->response = NULL; + static int on_url(http_parser* p, const char* data, size_t length) + { return 0; } @@ -408,33 +389,54 @@ private: return 0; } - static int on_url(http_parser* p, const char* data, size_t length) + static int on_headers_complete(http_parser* p) { - return 0; - } + ResponseDecoder* decoder = (ResponseDecoder*) p->data; -#if !(HTTP_PARSER_VERSION_MAJOR >= 2) - static int on_path(http_parser* p, const char* data, size_t length) - { - return 0; - } + // Add final header. + decoder->response->headers[decoder->field] = decoder->value; + decoder->field.clear(); + decoder->value.clear(); - static int on_query_string(http_parser* p, const char* data, size_t length) - { return 0; } - static int on_fragment(http_parser* p, const char* data, size_t length) + static int on_body(http_parser* p, const char* data, size_t length) { + ResponseDecoder* decoder = (ResponseDecoder*) p->data; + assert(decoder->response != NULL); + decoder->response->body.append(data, length); return 0; } -#endif // !(HTTP_PARSER_VERSION_MAJOR >= 2) - static int on_body(http_parser* p, const char* data, size_t length) + static int on_message_complete(http_parser* p) { ResponseDecoder* decoder = (ResponseDecoder*) p->data; - assert(decoder->response != NULL); - decoder->response->body.append(data, length); + + // Get the response status string. + if (http::statuses.contains(decoder->parser.status_code)) { + decoder->response->status = http::statuses[decoder->parser.status_code]; + } else { + decoder->failure = true; + return 1; + } + + // We can only provide the gzip encoding. + Option<std::string> encoding = + decoder->response->headers.get("Content-Encoding"); + if (encoding.isSome() && encoding.get() == "gzip") { + Try<std::string> decompressed = gzip::decompress(decoder->response->body); + if (decompressed.isError()) { + decoder->failure = true; + return 1; + } + decoder->response->body = decompressed.get(); + decoder->response->headers["Content-Length"] = + decoder->response->body.length(); + } + + decoder->responses.push_back(decoder->response); + decoder->response = NULL; return 0; }
