X-Git-Url: https://git.sesse.net/?p=cubemap;a=blobdiff_plain;f=server.cpp;h=7e4352276ea1b2d1db6a84f88bae5e76239b0824;hp=c80f2da2e2c354729702ef04ba9f7068ed5ec14c;hb=17d773d2d45d495704e974b9246eccb21faa8635;hpb=6d062eab70fd6528219545846e6c19c1cad35a3d diff --git a/server.cpp b/server.cpp index c80f2da..7e43522 100644 --- a/server.cpp +++ b/server.cpp @@ -18,6 +18,7 @@ #include "metacube.h" #include "server.h" #include "mutexlock.h" +#include "parse.h" #include "state.pb.h" using namespace std; @@ -25,7 +26,7 @@ using namespace std; Client::Client(int sock) : sock(sock), state(Client::READING_REQUEST), - header_bytes_sent(0), + header_or_error_bytes_sent(0), bytes_sent(0) { request.reserve(1024); @@ -36,8 +37,8 @@ Client::Client(const ClientProto &serialized) state(State(serialized.state())), request(serialized.request()), stream_id(serialized.stream_id()), - header(serialized.header()), - header_bytes_sent(serialized.header_bytes_sent()), + header_or_error(serialized.header_or_error()), + header_or_error_bytes_sent(serialized.header_or_error_bytes_sent()), bytes_sent(serialized.bytes_sent()) { } @@ -49,8 +50,8 @@ ClientProto Client::serialize() const serialized.set_state(state); serialized.set_request(request); serialized.set_stream_id(stream_id); - serialized.set_header(header); - serialized.set_header_bytes_sent(serialized.header_bytes_sent()); + serialized.set_header_or_error(header_or_error); + serialized.set_header_or_error_bytes_sent(serialized.header_or_error_bytes_sent()); serialized.set_bytes_sent(bytes_sent); return serialized; } @@ -69,7 +70,8 @@ Stream::~Stream() } Stream::Stream(const StreamProto &serialized) - : header(serialized.header()), + : stream_id(serialized.stream_id()), + header(serialized.header()), data(new char[BACKLOG_SIZE]), data_size(serialized.data_size()) { @@ -83,6 +85,7 @@ StreamProto Stream::serialize() const serialized.set_header(header); serialized.set_data(string(data, data + BACKLOG_SIZE)); serialized.set_data_size(data_size); + serialized.set_stream_id(stream_id); return serialized; } @@ -97,6 +100,11 @@ Server::Server() } } +Server::~Server() +{ + close(epoll_fd); +} + void Server::run() { should_stop = false; @@ -189,17 +197,57 @@ void Server::add_client(int sock) exit(1); } } - + +void Server::add_client_from_serialized(const ClientProto &client) +{ + MutexLock lock(&mutex); + clients.insert(make_pair(client.sock(), Client(client))); + + // Start listening on data from this socket. + epoll_event ev; + if (client.state() == Client::READING_REQUEST) { + ev.events = EPOLLIN | EPOLLRDHUP; + } else { + // If we don't have more data for this client, we'll be putting it into + // the sleeping array again soon. + ev.events = EPOLLOUT | EPOLLRDHUP; + } + ev.data.u64 = 0; // Keep Valgrind happy. + ev.data.fd = client.sock(); + if (epoll_ctl(epoll_fd, EPOLL_CTL_ADD, client.sock(), &ev) == -1) { + perror("epoll_ctl(EPOLL_CTL_ADD)"); + exit(1); + } +} + void Server::add_stream(const string &stream_id) { MutexLock lock(&mutex); streams.insert(make_pair(stream_id, new Stream(stream_id))); } + +void Server::add_stream_from_serialized(const StreamProto &stream) +{ + MutexLock lock(&mutex); + streams.insert(make_pair(stream.stream_id(), new Stream(stream))); +} void Server::set_header(const string &stream_id, const string &header) { MutexLock lock(&mutex); find_stream(stream_id)->header = header; + + // If there are clients we haven't sent anything to yet, we should give + // them the header, so push back into the SENDING_HEADER state. + for (map::iterator client_it = clients.begin(); + client_it != clients.end(); + ++client_it) { + Client *client = &client_it->second; + if (client->state == Client::SENDING_DATA && + client->bytes_sent == 0) { + construct_header(client); + } + } } void Server::add_data(const string &stream_id, const char *data, size_t bytes) @@ -272,34 +320,44 @@ void Server::process_client(Client *client) return; } - parse_request(client); + int error_code = parse_request(client); + if (error_code == 200) { + construct_header(client); + } else { + construct_error(client, error_code); + } break; } + case Client::SENDING_ERROR: case Client::SENDING_HEADER: { int ret = write(client->sock, - client->header.data() + client->header_bytes_sent, - client->header.size() - client->header_bytes_sent); + client->header_or_error.data() + client->header_or_error_bytes_sent, + client->header_or_error.size() - client->header_or_error_bytes_sent); if (ret == -1) { perror("write"); close_client(client); return; } - client->header_bytes_sent += ret; - assert(client->header_bytes_sent <= client->header.size()); + client->header_or_error_bytes_sent += ret; + assert(client->header_or_error_bytes_sent <= client->header_or_error.size()); - if (client->header_bytes_sent < client->header.size()) { + if (client->header_or_error_bytes_sent < client->header_or_error.size()) { // We haven't sent all yet. Fine; we'll do that later. return; } - // We're done sending the header! Clear the entire header to release some memory. - client->header.clear(); + // We're done sending the header or error! Clear it to release some memory. + client->header_or_error.clear(); - // Start sending from the end. In other words, we won't send any of the backlog, - // but we'll start sending immediately as we get data. - client->state = Client::SENDING_DATA; - client->bytes_sent = find_stream(client->stream_id)->data_size; + if (client->state == Client::SENDING_ERROR) { + close_client(client); + } else { + // Start sending from the end. In other words, we won't send any of the backlog, + // but we'll start sending immediately as we get data. + client->state = Client::SENDING_DATA; + client->bytes_sent = find_stream(client->stream_id)->data_size; + } break; } case Client::SENDING_DATA: { @@ -351,14 +409,33 @@ void Server::process_client(Client *client) } } -void Server::parse_request(Client *client) +int Server::parse_request(Client *client) { - // TODO: Actually parse the request. :-) - client->stream_id = "stream"; + vector lines = split_lines(client->request); + if (lines.empty()) { + return 400; // Bad request (empty). + } + + vector request_tokens = split_tokens(lines[0]); + if (request_tokens.size() < 2) { + return 400; // Bad request (empty). + } + if (request_tokens[0] != "GET") { + return 400; // Should maybe be 405 instead? + } + if (streams.count(request_tokens[1]) == 0) { + return 404; // Not found. + } + + client->stream_id = request_tokens[1]; client->request.clear(); - // Construct the header. - client->header = "HTTP/1.0 200 OK\r\n Content-type: video/x-flv\r\nCache-Control: no-cache\r\nContent-type: todo/fixme\r\n\r\n" + + return 200; // OK! +} + +void Server::construct_header(Client *client) +{ + client->header_or_error = "HTTP/1.0 200 OK\r\nContent-type: video/x-flv\r\nCache-Control: no-cache\r\n\r\n" + find_stream(client->stream_id)->header; // Switch states. @@ -375,6 +452,27 @@ void Server::parse_request(Client *client) } } +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; + + // Switch states. + client->state = Client::SENDING_ERROR; + + epoll_event ev; + ev.events = EPOLLOUT | EPOLLRDHUP; + ev.data.u64 = 0; // Keep Valgrind happy. + ev.data.fd = client->sock; + + if (epoll_ctl(epoll_fd, EPOLL_CTL_MOD, client->sock, &ev) == -1) { + perror("epoll_ctl(EPOLL_CTL_MOD)"); + exit(1); + } +} + void Server::close_client(Client *client) { if (epoll_ctl(epoll_fd, EPOLL_CTL_DEL, client->sock, NULL) == -1) {