X-Git-Url: https://git.sesse.net/?p=cubemap;a=blobdiff_plain;f=httpinput.cpp;h=c346a1423ebbe938e4ee3bb7f6dfe49011bb005a;hp=f92ac13ca9ca010f1b2a8efb2d45cba68c40b12f;hb=8b466d2b206f26aedf802b06fe93ba1b67960c93;hpb=0cb56be70f7ca4f4564eea892a99d20032359a1d diff --git a/httpinput.cpp b/httpinput.cpp index f92ac13..c346a14 100644 --- a/httpinput.cpp +++ b/httpinput.cpp @@ -1,4 +1,3 @@ -#include #include #include #include @@ -9,6 +8,7 @@ #include #include #include +#include #include #include #include @@ -62,7 +62,9 @@ HTTPInput::HTTPInput(const InputProto &serialized) void HTTPInput::close_socket() { - safe_close(sock); + if (sock != -1) { + safe_close(sock); + } } InputProto HTTPInput::serialize() const @@ -84,17 +86,16 @@ int HTTPInput::lookup_and_connect(const string &host, const string &port) { addrinfo *ai; int err = getaddrinfo(host.c_str(), port.c_str(), NULL, &ai); - if (err == -1) { + if (err != 0) { log(WARNING, "[%s] Lookup of '%s' failed (%s).", url.c_str(), host.c_str(), gai_strerror(err)); - freeaddrinfo(ai); return -1; } addrinfo *base_ai = ai; // Connect to everything in turn until we have a socket. - while (ai && !should_stop()) { + for ( ; ai && !should_stop(); ai = ai->ai_next) { int sock = socket(ai->ai_family, SOCK_STREAM, IPPROTO_TCP); if (sock == -1) { // Could be e.g. EPROTONOSUPPORT. The show must go on. @@ -152,7 +153,6 @@ int HTTPInput::lookup_and_connect(const string &host, const string &port) } safe_close(sock); - ai = ai->ai_next; } // Give the last one as error. @@ -240,8 +240,8 @@ bool HTTPInput::parse_response(const std::string &request) http_header.append(it->first + ": " + it->second + "\r\n"); } - for (size_t i = 0; i < stream_ids.size(); ++i) { - servers->set_header(stream_ids[i], http_header, ""); + for (size_t i = 0; i < stream_indices.size(); ++i) { + servers->set_header(stream_indices[i], http_header, ""); } return true; @@ -264,8 +264,9 @@ void HTTPInput::do_work() request_bytes_sent = 0; response.clear(); pending_data.clear(); - for (size_t i = 0; i < stream_ids.size(); ++i) { - servers->set_header(stream_ids[i], "", ""); + has_metacube_header = false; + for (size_t i = 0; i < stream_indices.size(); ++i) { + servers->set_header(stream_indices[i], "", ""); } { @@ -385,7 +386,7 @@ void HTTPInput::do_work() if (ret == 0) { // This really shouldn't happen... - log(ERROR, "[%s] Socket unexpectedly closed while reading header", + log(ERROR, "[%s] Socket unexpectedly closed while reading data", url.c_str()); state = CLOSING_SOCKET; continue; @@ -457,6 +458,11 @@ void HTTPInput::process_data(char *ptr, size_t bytes) uint32_t size = ntohl(hdr->size); uint32_t flags = ntohl(hdr->flags); + if (size > 65535) { + log(WARNING, "[%s] Metacube block of %x bytes (flags=%x); corrupted header?", + url.c_str(), size, flags); + } + // See if we have the entire block. If not, wait for more data. if (pending_data.size() < sizeof(metacube_block_header) + size) { return; @@ -466,12 +472,12 @@ void HTTPInput::process_data(char *ptr, size_t bytes) char *inner_data = pending_data.data() + sizeof(metacube_block_header); if (flags & METACUBE_FLAGS_HEADER) { string header(inner_data, inner_data + size); - for (size_t i = 0; i < stream_ids.size(); ++i) { - servers->set_header(stream_ids[i], http_header, header); + for (size_t i = 0; i < stream_indices.size(); ++i) { + servers->set_header(stream_indices[i], http_header, header); } } else { - for (size_t i = 0; i < stream_ids.size(); ++i) { - servers->add_data(stream_ids[i], inner_data, size); + for (size_t i = 0; i < stream_indices.size(); ++i) { + servers->add_data(stream_indices[i], inner_data, size); } } @@ -490,6 +496,7 @@ void HTTPInput::drop_pending_data(size_t num_bytes) } log(WARNING, "[%s] Dropping %lld junk bytes from stream, maybe it is not a Metacube stream?", url.c_str(), (long long)num_bytes); + assert(pending_data.size() >= num_bytes); pending_data.erase(pending_data.begin(), pending_data.begin() + num_bytes); }