X-Git-Url: https://git.sesse.net/?p=cubemap;a=blobdiff_plain;f=server.cpp;h=d3b936f0b9c1f869c34c5dbf05c1beb32ad90a8e;hp=e99dbeaa89b311788a9c0fc4eb132a667db3f5b1;hb=6544fa0ec3f3a501bcb89ea977756911bd7f3ebd;hpb=150a9636b58ac912ed4d46e6d8379beb5632f2f8 diff --git a/server.cpp b/server.cpp index e99dbea..d3b936f 100644 --- a/server.cpp +++ b/server.cpp @@ -140,7 +140,7 @@ void Server::do_work() } timeout_time.tv_sec -= REQUEST_READ_TIMEOUT_SEC; while (!clients_ordered_by_connect_time.empty()) { - pair &connect_time_and_fd = clients_ordered_by_connect_time.front(); + const pair &connect_time_and_fd = clients_ordered_by_connect_time.front(); // See if we have reached the end of clients to process. if (is_earlier(timeout_time, connect_time_and_fd.first)) { @@ -149,7 +149,7 @@ void Server::do_work() // If this client doesn't exist anymore, just ignore it // (it was deleted earlier). - std::map::iterator client_it = clients.find(connect_time_and_fd.second); + map::iterator client_it = clients.find(connect_time_and_fd.second); if (client_it == clients.end()) { clients_ordered_by_connect_time.pop(); continue; @@ -286,27 +286,27 @@ void Server::add_client_from_serialized(const ClientProto &client) } } -int Server::lookup_stream_by_url(const std::string &url) const +int Server::lookup_stream_by_url(const string &url) const { - map::const_iterator url_it = url_map.find(url); - if (url_it == url_map.end()) { + map::const_iterator stream_url_it = stream_url_map.find(url); + if (stream_url_it == stream_url_map.end()) { return -1; } - return url_it->second; + return stream_url_it->second; } -int Server::add_stream(const string &url, size_t backlog_size, size_t prebuffering_bytes, Stream::Encoding encoding) +int Server::add_stream(const string &url, size_t backlog_size, size_t prebuffering_bytes, Stream::Encoding encoding, Stream::Encoding src_encoding) { MutexLock lock(&mutex); - url_map.insert(make_pair(url, streams.size())); - streams.push_back(new Stream(url, backlog_size, prebuffering_bytes, encoding)); + stream_url_map.insert(make_pair(url, streams.size())); + streams.push_back(new Stream(url, backlog_size, prebuffering_bytes, encoding, src_encoding)); return streams.size() - 1; } int Server::add_stream_from_serialized(const StreamProto &stream, int data_fd) { MutexLock lock(&mutex); - url_map.insert(make_pair(stream.url(), streams.size())); + stream_url_map.insert(make_pair(stream.url(), streams.size())); streams.push_back(new Stream(stream, data_fd)); return streams.size() - 1; } @@ -331,6 +331,13 @@ void Server::set_encoding(int stream_index, Stream::Encoding encoding) assert(stream_index >= 0 && stream_index < ssize_t(streams.size())); streams[stream_index]->encoding = encoding; } + +void Server::set_src_encoding(int stream_index, Stream::Encoding encoding) +{ + MutexLock lock(&mutex); + assert(stream_index >= 0 && stream_index < ssize_t(streams.size())); + streams[stream_index]->src_encoding = encoding; +} void Server::set_header(int stream_index, const string &http_header, const string &stream_header) { @@ -359,6 +366,13 @@ void Server::set_pacing_rate(int stream_index, uint32_t pacing_rate) streams[stream_index]->pacing_rate = pacing_rate; } +void Server::add_gen204(const std::string &url, const std::string &allow_origin) +{ + MutexLock lock(&mutex); + assert(clients.empty()); + ping_url_map[url] = allow_origin; +} + void Server::add_data_deferred(int stream_index, const char *data, size_t bytes, StreamStartSuitability suitable_for_stream_start) { assert(stream_index >= 0 && stream_index < ssize_t(streams.size())); @@ -418,22 +432,24 @@ read_request_again: int error_code = parse_request(client); if (error_code == 200) { construct_header(client); + } else if (error_code == 204) { + construct_204(client); } else { construct_error(client, error_code); } // We've changed states, so fall through. - assert(client->state == Client::SENDING_ERROR || + assert(client->state == Client::SENDING_SHORT_RESPONSE || client->state == Client::SENDING_HEADER); } - case Client::SENDING_ERROR: + case Client::SENDING_SHORT_RESPONSE: case Client::SENDING_HEADER: { -sending_header_or_error_again: +sending_header_or_short_response_again: int ret; do { ret = write(client->sock, - client->header_or_error.data() + client->header_or_error_bytes_sent, - client->header_or_error.size() - client->header_or_error_bytes_sent); + client->header_or_short_response.data() + client->header_or_short_response_bytes_sent, + client->header_or_short_response.size() - client->header_or_short_response_bytes_sent); } while (ret == -1 && errno == EINTR); if (ret == -1 && errno == EAGAIN) { @@ -451,18 +467,18 @@ sending_header_or_error_again: return; } - client->header_or_error_bytes_sent += ret; - assert(client->header_or_error_bytes_sent <= client->header_or_error.size()); + client->header_or_short_response_bytes_sent += ret; + assert(client->header_or_short_response_bytes_sent <= client->header_or_short_response.size()); - if (client->header_or_error_bytes_sent < client->header_or_error.size()) { + if (client->header_or_short_response_bytes_sent < client->header_or_short_response.size()) { // We haven't sent all yet. Fine; go another round. - goto sending_header_or_error_again; + goto sending_header_or_short_response_again; } // We're done sending the header or error! Clear it to release some memory. - client->header_or_error.clear(); + client->header_or_short_response.clear(); - if (client->state == Client::SENDING_ERROR) { + if (client->state == Client::SENDING_SHORT_RESPONSE) { // We're done sending the error, so now close. // This is postcondition #1. close_client(client); @@ -472,7 +488,7 @@ sending_header_or_error_again: Stream *stream = client->stream; if (client->stream_pos == size_t(-2)) { // Start sending from the beginning of the backlog. - client->stream_pos = std::min( + client->stream_pos = min( stream->bytes_received - stream->backlog_size, 0); client->state = Client::SENDING_DATA; @@ -636,6 +652,7 @@ int Server::parse_request(Client *client) } string url = request_tokens[1]; + client->url = url; if (url.find("?backlog") == url.size() - 8) { client->stream_pos = -2; url = url.substr(0, url.size() - 8); @@ -643,18 +660,21 @@ int Server::parse_request(Client *client) client->stream_pos = -1; } - map::const_iterator url_map_it = url_map.find(url); - if (url_map_it == url_map.end()) { - return 404; // Not found. + map::const_iterator stream_url_map_it = stream_url_map.find(url); + if (stream_url_map_it == stream_url_map.end()) { + map::const_iterator ping_url_map_it = ping_url_map.find(url); + if (ping_url_map_it == ping_url_map.end()) { + return 404; // Not found. + } else { + return 204; // No error. + } } - Stream *stream = streams[url_map_it->second]; + Stream *stream = streams[stream_url_map_it->second]; if (stream->http_header.empty()) { return 503; // Service unavailable. } - client->url = request_tokens[1]; - client->stream = stream; if (setsockopt(client->sock, SOL_SOCKET, SO_MAX_PACING_RATE, &client->stream->pacing_rate, sizeof(client->stream->pacing_rate)) == -1) { if (client->stream->pacing_rate != ~0U) { @@ -670,11 +690,11 @@ void Server::construct_header(Client *client) { Stream *stream = client->stream; if (stream->encoding == Stream::STREAM_ENCODING_RAW) { - client->header_or_error = stream->http_header + + client->header_or_short_response = stream->http_header + "\r\n" + stream->stream_header; } else if (stream->encoding == Stream::STREAM_ENCODING_METACUBE) { - client->header_or_error = stream->http_header + + client->header_or_short_response = stream->http_header + "Content-encoding: metacube\r\n" + "\r\n"; if (!stream->stream_header.empty()) { @@ -683,10 +703,10 @@ void Server::construct_header(Client *client) hdr.size = htonl(stream->stream_header.size()); hdr.flags = htons(METACUBE_FLAGS_HEADER); hdr.csum = htons(metacube2_compute_crc(&hdr)); - client->header_or_error.append( + client->header_or_short_response.append( string(reinterpret_cast(&hdr), sizeof(hdr))); } - client->header_or_error.append(stream->stream_header); + client->header_or_short_response.append(stream->stream_header); } else { assert(false); } @@ -709,10 +729,42 @@ void Server::construct_error(Client *client, int error_code) char error[256]; snprintf(error, 256, "HTTP/1.0 %d Error\r\nContent-type: text/plain\r\n\r\nSomething went wrong. Sorry.\r\n", error_code); - client->header_or_error = error; + client->header_or_short_response = error; + + // Switch states. + client->state = Client::SENDING_SHORT_RESPONSE; + + epoll_event ev; + ev.events = EPOLLOUT | EPOLLET | EPOLLRDHUP; + ev.data.u64 = reinterpret_cast(client); + + if (epoll_ctl(epoll_fd, EPOLL_CTL_MOD, client->sock, &ev) == -1) { + log_perror("epoll_ctl(EPOLL_CTL_MOD)"); + exit(1); + } +} + +void Server::construct_204(Client *client) +{ + map::const_iterator ping_url_map_it = ping_url_map.find(client->url); + assert(ping_url_map_it != ping_url_map.end()); + + if (ping_url_map_it->second.empty()) { + client->header_or_short_response = + "HTTP/1.0 204 No Content\r\n" + "\r\n"; + } else { + char response[256]; + snprintf(response, 256, + "HTTP/1.0 204 No Content\r\n" + "Access-Control-Allow-Origin: %s\r\n" + "\r\n", + ping_url_map_it->second.c_str()); + client->header_or_short_response = response; + } // Switch states. - client->state = Client::SENDING_ERROR; + client->state = Client::SENDING_SHORT_RESPONSE; epoll_event ev; ev.events = EPOLLOUT | EPOLLET | EPOLLRDHUP;