Make Input a bit more generic, to pave the way for UDP.
authorSteinar H. Gunderson <sgunderson@bigfoot.com>
Thu, 11 Apr 2013 19:06:03 +0000 (21:06 +0200)
committerSteinar H. Gunderson <sgunderson@bigfoot.com>
Thu, 11 Apr 2013 19:06:03 +0000 (21:06 +0200)
Makefile
httpinput.cpp [new file with mode: 0644]
httpinput.h [new file with mode: 0644]
input.cpp
input.h
main.cpp

index 647dfe9..5367daa 100644 (file)
--- a/Makefile
+++ b/Makefile
@@ -4,7 +4,7 @@ PROTOC=protoc
 CXXFLAGS=-Wall -O2 -g
 LDLIBS=-lpthread -lprotobuf
 
-OBJS=main.o server.o serverpool.o mutexlock.o input.o parse.o markpool.o acceptor.o stats.o thread.o state.pb.o
+OBJS=main.o server.o serverpool.o mutexlock.o input.o httpinput.o parse.o markpool.o acceptor.o stats.o thread.o state.pb.o
 
 all: cubemap
 
diff --git a/httpinput.cpp b/httpinput.cpp
new file mode 100644 (file)
index 0000000..1db4f91
--- /dev/null
@@ -0,0 +1,439 @@
+#include <stdio.h>
+#include <string.h>
+#include <stdint.h>
+#include <assert.h>
+#include <arpa/inet.h>
+#include <sys/socket.h>
+#include <pthread.h>
+#include <sys/types.h>
+#include <sys/ioctl.h>
+#include <sys/types.h>
+#include <sys/socket.h>
+#include <netdb.h>
+#include <poll.h>
+#include <signal.h>
+#include <errno.h>
+#include <vector>
+#include <string>
+#include <map>
+
+#include "metacube.h"
+#include "mutexlock.h"
+#include "httpinput.h"
+#include "server.h"
+#include "serverpool.h"
+#include "parse.h"
+#include "state.pb.h"
+
+using namespace std;
+
+extern ServerPool *servers;
+         
+HTTPInput::HTTPInput(const string &stream_id, const string &url)
+       : state(NOT_CONNECTED),
+         stream_id(stream_id),
+         url(url),
+         has_metacube_header(false),
+         sock(-1)
+{
+}
+
+HTTPInput::HTTPInput(const InputProto &serialized)
+       : state(State(serialized.state())),
+         stream_id(serialized.stream_id()),
+         url(serialized.url()),
+         request(serialized.request()),
+         request_bytes_sent(serialized.request_bytes_sent()),
+         response(serialized.response()),
+         http_header(serialized.http_header()),
+         has_metacube_header(serialized.has_metacube_header()),
+         sock(serialized.sock())
+{
+       pending_data.resize(serialized.pending_data().size());
+       memcpy(&pending_data[0], serialized.pending_data().data(), serialized.pending_data().size());
+
+       string protocol;
+       parse_url(url, &protocol, &host, &port, &path);  // Don't care if it fails.
+}
+
+InputProto HTTPInput::serialize() const
+{
+       InputProto serialized;
+       serialized.set_state(state);
+       serialized.set_stream_id(stream_id);
+       serialized.set_url(url);
+       serialized.set_request(request);
+       serialized.set_request_bytes_sent(request_bytes_sent);
+       serialized.set_response(response);
+       serialized.set_http_header(http_header);
+       serialized.set_pending_data(string(pending_data.begin(), pending_data.end()));
+       serialized.set_has_metacube_header(has_metacube_header);
+       serialized.set_sock(sock);
+       return serialized;
+}
+
+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) {
+               fprintf(stderr, "WARNING: Lookup of '%s' failed (%s).\n",
+                       host.c_str(), gai_strerror(err));
+               freeaddrinfo(ai);
+               return -1;
+       }
+
+       // Connect to everything in turn until we have a socket.
+       while (ai && !should_stop) {
+               int sock = socket(ai->ai_family, SOCK_STREAM, IPPROTO_TCP);
+               if (sock == -1) {
+                       // Could be e.g. EPROTONOSUPPORT. The show must go on.
+                       continue;
+               }
+
+               do {
+                       err = connect(sock, ai->ai_addr, ai->ai_addrlen);
+               } while (err == -1 && errno == EINTR);
+
+               if (err != -1) {
+                       freeaddrinfo(ai);
+                       return sock;
+               }
+
+               ai = ai->ai_next;
+       }
+
+       // Give the last one as error.
+       fprintf(stderr, "WARNING: Connect to '%s' failed (%s)\n",
+               host.c_str(), strerror(errno));
+       freeaddrinfo(ai);
+       return -1;
+}
+       
+bool HTTPInput::parse_response(const std::string &request)
+{
+       vector<string> lines = split_lines(response);
+       if (lines.empty()) {
+               fprintf(stderr, "WARNING: Empty HTTP response from input.\n");
+               return false;
+       }
+
+       vector<string> first_line_tokens = split_tokens(lines[0]);
+       if (first_line_tokens.size() < 2) {
+               fprintf(stderr, "WARNING: Malformed response line '%s' from input.\n",
+                       lines[0].c_str());
+               return false;
+       }
+
+       int response = atoi(first_line_tokens[1].c_str());
+       if (response != 200) {
+               fprintf(stderr, "WARNING: Non-200 response '%s' from input.\n",
+                       lines[0].c_str());
+               return false;
+       }
+
+       multimap<string, string> parameters;
+       for (size_t i = 1; i < lines.size(); ++i) {
+               size_t split = lines[i].find(":");
+               if (split == string::npos) {
+                       fprintf(stderr, "WARNING: Ignoring malformed HTTP response line '%s'\n",
+                               lines[i].c_str());
+                       continue;
+               }
+
+               string key(lines[i].begin(), lines[i].begin() + split);
+
+               // Skip any spaces after the colon.
+               do {
+                       ++split;
+               } while (split < lines[i].size() && lines[i][split] == ' ');
+
+               string value(lines[i].begin() + split, lines[i].end());
+
+               // Remove “Content-encoding: metacube”.
+               // TODO: Make case-insensitive.
+               if (key == "Content-encoding" && value == "metacube") {
+                       continue;
+               }
+
+               parameters.insert(make_pair(key, value));
+       }
+
+       // Change “Server: foo” to “Server: metacube/0.1 (reflecting: foo)”
+       // TODO: Make case-insensitive.
+       // XXX: Use a Via: instead?
+       if (parameters.count("Server") == 0) {
+               parameters.insert(make_pair("Server", "metacube/0.1"));
+       } else {
+               for (multimap<string, string>::iterator it = parameters.begin();
+                    it != parameters.end();
+                    ++it) {
+                       if (it->first != "Server") {
+                               continue;
+                       }
+                       it->second = "metacube/0.1 (reflecting: " + it->second + ")";
+               }
+       }
+
+       // Construct the new HTTP header.
+       http_header = "HTTP/1.0 200 OK\r\n";
+       for (multimap<string, string>::iterator it = parameters.begin();
+            it != parameters.end();
+            ++it) {
+               http_header.append(it->first + ": " + it->second + "\r\n");
+       }
+       http_header.append("\r\n");     
+       servers->set_header(stream_id, http_header);
+
+       return true;
+}
+
+void HTTPInput::do_work()
+{
+       while (!should_stop) {
+               if (state == SENDING_REQUEST || state == RECEIVING_HEADER || state == RECEIVING_DATA) {
+                       // Since we are non-blocking, we need to wait for the right state first.
+                       // Wait up to 50 ms, then check should_stop.
+                       pollfd pfd;
+                       pfd.fd = sock;
+                       pfd.events = (state == SENDING_REQUEST) ? POLLOUT : POLLIN;
+                       pfd.events |= POLLRDHUP;
+
+                       int nfds = poll(&pfd, 1, 50);
+                       if (nfds == 0 || (nfds == -1 && errno == EINTR)) {
+                               continue;
+                       }
+                       if (nfds == -1) {
+                               perror("poll");
+                               state = CLOSING_SOCKET;
+                       }
+               }
+
+               switch (state) {
+               case NOT_CONNECTED:
+                       request.clear();
+                       request_bytes_sent = 0;
+                       response.clear();
+
+                       {
+                               string protocol;  // Thrown away.
+                               if (!parse_url(url, &protocol, &host, &port, &path)) {
+                                       fprintf(stderr, "Failed to parse URL '%s'\n", url.c_str());
+                                       break;
+                               }
+                       }
+
+                       sock = lookup_and_connect(host, port);
+                       if (sock != -1) {
+                               // Yay, successful connect. Try to set it as nonblocking.
+                               int one = 1;
+                               if (ioctl(sock, FIONBIO, &one) == -1) {
+                                       perror("ioctl(FIONBIO)");
+                                       state = CLOSING_SOCKET;
+                               } else {
+                                       state = SENDING_REQUEST;
+                                       request = "GET " + path + " HTTP/1.0\r\nUser-Agent: cubemap\r\n\r\n";
+                                       request_bytes_sent = 0;
+                               }
+                       }
+                       break;
+               case SENDING_REQUEST: {
+                       size_t to_send = request.size() - request_bytes_sent;
+                       int ret;
+
+                       do {
+                               ret = write(sock, request.data() + request_bytes_sent, to_send);
+                       } while (ret == -1 && errno == EINTR);
+
+                       if (ret == -1) {
+                               perror("write");
+                               state = CLOSING_SOCKET;
+                               continue;
+                       }
+
+                       assert(ret >= 0);
+                       request_bytes_sent += ret;
+
+                       if (request_bytes_sent == request.size()) {
+                               state = RECEIVING_HEADER;
+                       }
+                       break;
+               }
+               case RECEIVING_HEADER: {
+                       char buf[4096];
+                       int ret;
+
+                       do {
+                               ret = read(sock, buf, sizeof(buf));
+                       } while (ret == -1 && errno == EINTR);
+
+                       if (ret == -1) {
+                               perror("read");
+                               state = CLOSING_SOCKET;
+                               continue;
+                       }
+
+                       if (ret == 0) {
+                               // This really shouldn't happen...
+                               fprintf(stderr, "Socket unexpectedly closed while reading header\n");
+                               state = CLOSING_SOCKET;
+                               continue;
+                       }
+                       
+                       RequestParseStatus status = wait_for_double_newline(&response, buf, ret);
+                       
+                       if (status == RP_OUT_OF_SPACE) {
+                               fprintf(stderr, "WARNING: fd %d sent overlong response!\n", sock);
+                               state = CLOSING_SOCKET;
+                               continue;
+                       } else if (status == RP_NOT_FINISHED_YET) {
+                               continue;
+                       }
+       
+                       // OK, so we're fine, but there might be some of the actual data after the response.
+                       // We'll need to deal with that separately.
+                       string extra_data;
+                       if (status == RP_EXTRA_DATA) {
+                               char *ptr = static_cast<char *>(
+                                       memmem(response.data(), response.size(), "\r\n\r\n", 4));
+                               assert(ptr != NULL);
+                               extra_data = string(ptr, &response[0] + response.size());
+                               response.resize(ptr - response.data());
+                       }
+
+                       if (!parse_response(response)) {
+                               state = CLOSING_SOCKET;
+                               continue;
+                       }
+
+                       if (!extra_data.empty()) {
+                               process_data(&extra_data[0], extra_data.size());
+                       }
+
+                       state = RECEIVING_DATA;
+                       break;
+               }
+               case RECEIVING_DATA: {
+                       char buf[4096];
+                       int ret;
+
+                       do {
+                               ret = read(sock, buf, sizeof(buf));
+                       } while (ret == -1 && errno == EINTR);
+
+                       if (ret == -1) {
+                               perror("read");
+                               state = CLOSING_SOCKET;
+                               continue;
+                       }
+
+                       if (ret == 0) {
+                               // This really shouldn't happen...
+                               fprintf(stderr, "Socket unexpectedly closed while reading header\n");
+                               state = CLOSING_SOCKET;
+                               continue;
+                       }
+
+                       process_data(buf, ret);
+                       break;
+               }
+               case CLOSING_SOCKET: {
+                       int err;
+                       do {
+                               err = close(sock);
+                       } while (err == -1 && errno == EINTR);
+
+                       if (err == -1) {
+                               perror("close");
+                       }
+
+                       state = NOT_CONNECTED;
+                       break;
+               }
+               default:
+                       assert(false);
+               }
+
+               // If we are still in NOT_CONNECTED, either something went wrong,
+               // or the connection just got closed.
+               // The earlier steps have already given the error message, if any.
+               if (state == NOT_CONNECTED && !should_stop) {
+                       fprintf(stderr, "Waiting 0.2 second and restarting...\n");
+                       usleep(200000);
+               }
+       }
+}
+
+void HTTPInput::process_data(char *ptr, size_t bytes)
+{
+       pending_data.insert(pending_data.end(), ptr, ptr + bytes);
+
+       for ( ;; ) {
+               // If we don't have enough data (yet) for even the Metacube header, just return.
+               if (pending_data.size() < sizeof(metacube_block_header)) {
+                       return;
+               }
+
+               // Make sure we have the Metacube sync header at the start.
+               // We may need to skip over junk data (it _should_ not happen, though).
+               if (!has_metacube_header) {
+                       char *ptr = static_cast<char *>(
+                               memmem(pending_data.data(), pending_data.size(),
+                                      METACUBE_SYNC, strlen(METACUBE_SYNC)));
+                       if (ptr == NULL) {
+                               // OK, so we didn't find the sync marker. We know then that
+                               // we do not have the _full_ marker in the buffer, but we
+                               // could have N-1 bytes. Drop everything before that,
+                               // and then give up.
+                               drop_pending_data(pending_data.size() - (strlen(METACUBE_SYNC) - 1));
+                               return;
+                       } else {
+                               // Yay, we found the header. Drop everything (if anything) before it.
+                               drop_pending_data(ptr - pending_data.data());
+                               has_metacube_header = true;
+
+                               // Re-check that we have the entire header; we could have dropped data.
+                               if (pending_data.size() < sizeof(metacube_block_header)) {
+                                       return;
+                               }
+                       }
+               }
+
+               // Now it's safe to read the header.
+               metacube_block_header *hdr = reinterpret_cast<metacube_block_header *>(pending_data.data());    
+               assert(memcmp(hdr->sync, METACUBE_SYNC, sizeof(hdr->sync)) == 0);
+               uint32_t size = ntohl(hdr->size);
+               uint32_t flags = ntohl(hdr->flags);
+
+               // See if we have the entire block. If not, wait for more data.
+               if (pending_data.size() < sizeof(metacube_block_header) + size) {
+                       return;
+               }
+
+               // Send this block on to the data.
+               char *inner_data = pending_data.data() + sizeof(metacube_block_header);
+               if (flags & METACUBE_FLAGS_HEADER) {
+                       string header(inner_data, inner_data + size);
+                       servers->set_header(stream_id, http_header + header);
+               } else { 
+                       servers->add_data(stream_id, inner_data, size);
+               }
+
+               // Consume the block. This isn't the most efficient way of dealing with things
+               // should we have many blocks, but these routines don't need to be too efficient
+               // anyway.
+               pending_data.erase(pending_data.begin(), pending_data.begin() + sizeof(metacube_block_header) + size);
+               has_metacube_header = false;
+       }
+}
+
+void HTTPInput::drop_pending_data(size_t num_bytes)
+{
+       if (num_bytes == 0) {
+               return;
+       }
+       fprintf(stderr, "Warning: Dropping %lld junk bytes from stream, maybe it is not a Metacube stream?\n",
+               (long long)num_bytes);
+       pending_data.erase(pending_data.begin(), pending_data.begin() + num_bytes);
+}
+
diff --git a/httpinput.h b/httpinput.h
new file mode 100644 (file)
index 0000000..894d8e9
--- /dev/null
@@ -0,0 +1,79 @@
+#ifndef _HTTPINPUT_H
+#define _HTTPINPUT_H 1
+
+#include <vector>
+#include <string>
+
+#include "input.h"
+
+class InputProto;
+
+class HTTPInput : public Input {
+public:
+       HTTPInput(const std::string &stream_id, const std::string &url);
+
+       // Serialization/deserialization.
+       HTTPInput(const InputProto &serialized);
+       virtual InputProto serialize() const;
+
+       virtual std::string get_url() const { return url; }
+
+private:
+       // Actually does the download.
+       virtual void do_work();
+       
+       // Open a socket that connects to the given host and port. Does DNS resolving.
+       int lookup_and_connect(const std::string &host, const std::string &port);
+
+       // Parses a HTTP response. Returns false if it not a 200.
+       bool parse_response(const std::string &response);
+
+       // Stores the given data, looks for Metacube blocks (skipping data if needed),
+       // and calls process_block() for each one.
+       void process_data(char *ptr, size_t bytes);
+
+       // Drops <num_bytes> bytes from the head of <pending_data>,
+       // and outputs a warning.
+       void drop_pending_data(size_t num_bytes);
+
+       enum State {
+               NOT_CONNECTED,
+               SENDING_REQUEST,
+               RECEIVING_HEADER,
+               RECEIVING_DATA,
+               CLOSING_SOCKET,  // Due to error.
+       };
+       State state;
+
+       std::string stream_id;
+
+       // The URL and its parsed components.
+       std::string url;
+       std::string host, port, path;
+
+       // The HTTP request, with headers and all.
+       // Only relevant for SENDING_REQUEST.
+       std::string request;
+
+       // How many bytes we've sent of the request so far.
+       // Only relevant for SENDING_REQUEST.
+       size_t request_bytes_sent;
+
+       // The HTTP response we've received so far. Only relevant for RECEIVING_HEADER.
+       std::string response;
+
+       // The HTTP respones headers we want to give clients for this input.
+       std::string http_header;
+
+       // Data we have received but not fully processed yet.
+       std::vector<char> pending_data;
+
+       // If <pending_data> starts with a Metacube header,
+       // this is true.
+       bool has_metacube_header;
+
+       // The socket we are downloading on (or -1).
+       int sock;       
+};
+
+#endif  // !defined(_HTTPINPUT_H)
index a54e6fb..1e61780 100644 (file)
--- a/input.cpp
+++ b/input.cpp
@@ -1,40 +1,18 @@
-#include <stdio.h>
 #include <string.h>
-#include <stdint.h>
-#include <assert.h>
-#include <arpa/inet.h>
-#include <sys/socket.h>
-#include <pthread.h>
-#include <sys/types.h>
-#include <sys/ioctl.h>
-#include <sys/types.h>
-#include <sys/socket.h>
-#include <netdb.h>
-#include <poll.h>
-#include <signal.h>
-#include <errno.h>
-#include <vector>
 #include <string>
-#include <map>
 
-#include "metacube.h"
-#include "mutexlock.h"
 #include "input.h"
-#include "server.h"
-#include "serverpool.h"
-#include "parse.h"
-#include "state.pb.h"
 
 using namespace std;
 
-extern ServerPool *servers;
-         
 // Extremely rudimentary URL parsing.
-bool parse_url(const string &url, string *host, string *port, string *path)
+bool parse_url(const string &url, string *protocol, string *host, string *port, string *path)
 {
        if (url.find("http://") != 0) {
                return false;
        }
+
+       *protocol = "http";
        
        string rest = url.substr(strlen("http://"));
        size_t split = rest.find_first_of(":/");
@@ -72,407 +50,5 @@ bool parse_url(const string &url, string *host, string *port, string *path)
        return true;
 }
 
-Input::Input(const string &stream_id, const string &url)
-       : state(NOT_CONNECTED),
-         stream_id(stream_id),
-         url(url),
-         has_metacube_header(false),
-         sock(-1)
-{
-}
-
-Input::Input(const InputProto &serialized)
-       : state(State(serialized.state())),
-         stream_id(serialized.stream_id()),
-         url(serialized.url()),
-         request(serialized.request()),
-         request_bytes_sent(serialized.request_bytes_sent()),
-         response(serialized.response()),
-         http_header(serialized.http_header()),
-         has_metacube_header(serialized.has_metacube_header()),
-         sock(serialized.sock())
-{
-       pending_data.resize(serialized.pending_data().size());
-       memcpy(&pending_data[0], serialized.pending_data().data(), serialized.pending_data().size());
-
-       parse_url(url, &host, &port, &path);  // Don't care if it fails.
-}
-
-InputProto Input::serialize() const
-{
-       InputProto serialized;
-       serialized.set_state(state);
-       serialized.set_stream_id(stream_id);
-       serialized.set_url(url);
-       serialized.set_request(request);
-       serialized.set_request_bytes_sent(request_bytes_sent);
-       serialized.set_response(response);
-       serialized.set_http_header(http_header);
-       serialized.set_pending_data(string(pending_data.begin(), pending_data.end()));
-       serialized.set_has_metacube_header(has_metacube_header);
-       serialized.set_sock(sock);
-       return serialized;
-}
-
-int Input::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) {
-               fprintf(stderr, "WARNING: Lookup of '%s' failed (%s).\n",
-                       host.c_str(), gai_strerror(err));
-               freeaddrinfo(ai);
-               return -1;
-       }
-
-       // Connect to everything in turn until we have a socket.
-       while (ai && !should_stop) {
-               int sock = socket(ai->ai_family, SOCK_STREAM, IPPROTO_TCP);
-               if (sock == -1) {
-                       // Could be e.g. EPROTONOSUPPORT. The show must go on.
-                       continue;
-               }
-
-               do {
-                       err = connect(sock, ai->ai_addr, ai->ai_addrlen);
-               } while (err == -1 && errno == EINTR);
-
-               if (err != -1) {
-                       freeaddrinfo(ai);
-                       return sock;
-               }
-
-               ai = ai->ai_next;
-       }
-
-       // Give the last one as error.
-       fprintf(stderr, "WARNING: Connect to '%s' failed (%s)\n",
-               host.c_str(), strerror(errno));
-       freeaddrinfo(ai);
-       return -1;
-}
-       
-bool Input::parse_response(const std::string &request)
-{
-       vector<string> lines = split_lines(response);
-       if (lines.empty()) {
-               fprintf(stderr, "WARNING: Empty HTTP response from input.\n");
-               return false;
-       }
-
-       vector<string> first_line_tokens = split_tokens(lines[0]);
-       if (first_line_tokens.size() < 2) {
-               fprintf(stderr, "WARNING: Malformed response line '%s' from input.\n",
-                       lines[0].c_str());
-               return false;
-       }
-
-       int response = atoi(first_line_tokens[1].c_str());
-       if (response != 200) {
-               fprintf(stderr, "WARNING: Non-200 response '%s' from input.\n",
-                       lines[0].c_str());
-               return false;
-       }
-
-       multimap<string, string> parameters;
-       for (size_t i = 1; i < lines.size(); ++i) {
-               size_t split = lines[i].find(":");
-               if (split == string::npos) {
-                       fprintf(stderr, "WARNING: Ignoring malformed HTTP response line '%s'\n",
-                               lines[i].c_str());
-                       continue;
-               }
-
-               string key(lines[i].begin(), lines[i].begin() + split);
-
-               // Skip any spaces after the colon.
-               do {
-                       ++split;
-               } while (split < lines[i].size() && lines[i][split] == ' ');
-
-               string value(lines[i].begin() + split, lines[i].end());
-
-               // Remove “Content-encoding: metacube”.
-               // TODO: Make case-insensitive.
-               if (key == "Content-encoding" && value == "metacube") {
-                       continue;
-               }
-
-               parameters.insert(make_pair(key, value));
-       }
-
-       // Change “Server: foo” to “Server: metacube/0.1 (reflecting: foo)”
-       // TODO: Make case-insensitive.
-       // XXX: Use a Via: instead?
-       if (parameters.count("Server") == 0) {
-               parameters.insert(make_pair("Server", "metacube/0.1"));
-       } else {
-               for (multimap<string, string>::iterator it = parameters.begin();
-                    it != parameters.end();
-                    ++it) {
-                       if (it->first != "Server") {
-                               continue;
-                       }
-                       it->second = "metacube/0.1 (reflecting: " + it->second + ")";
-               }
-       }
-
-       // Construct the new HTTP header.
-       http_header = "HTTP/1.0 200 OK\r\n";
-       for (multimap<string, string>::iterator it = parameters.begin();
-            it != parameters.end();
-            ++it) {
-               http_header.append(it->first + ": " + it->second + "\r\n");
-       }
-       http_header.append("\r\n");     
-       servers->set_header(stream_id, http_header);
-
-       return true;
-}
-
-void Input::do_work()
-{
-       while (!should_stop) {
-               if (state == SENDING_REQUEST || state == RECEIVING_HEADER || state == RECEIVING_DATA) {
-                       // Since we are non-blocking, we need to wait for the right state first.
-                       // Wait up to 50 ms, then check should_stop.
-                       pollfd pfd;
-                       pfd.fd = sock;
-                       pfd.events = (state == SENDING_REQUEST) ? POLLOUT : POLLIN;
-                       pfd.events |= POLLRDHUP;
-
-                       int nfds = poll(&pfd, 1, 50);
-                       if (nfds == 0 || (nfds == -1 && errno == EINTR)) {
-                               continue;
-                       }
-                       if (nfds == -1) {
-                               perror("poll");
-                               state = CLOSING_SOCKET;
-                       }
-               }
-
-               switch (state) {
-               case NOT_CONNECTED:
-                       request.clear();
-                       request_bytes_sent = 0;
-                       response.clear();
-       
-                       if (!parse_url(url, &host, &port, &path)) {
-                               fprintf(stderr, "Failed to parse URL '%s'\n", url.c_str());
-                               break;
-                       }
-
-                       sock = lookup_and_connect(host, port);
-                       if (sock != -1) {
-                               // Yay, successful connect. Try to set it as nonblocking.
-                               int one = 1;
-                               if (ioctl(sock, FIONBIO, &one) == -1) {
-                                       perror("ioctl(FIONBIO)");
-                                       state = CLOSING_SOCKET;
-                               } else {
-                                       state = SENDING_REQUEST;
-                                       request = "GET " + path + " HTTP/1.0\r\nUser-Agent: cubemap\r\n\r\n";
-                                       request_bytes_sent = 0;
-                               }
-                       }
-                       break;
-               case SENDING_REQUEST: {
-                       size_t to_send = request.size() - request_bytes_sent;
-                       int ret;
-
-                       do {
-                               ret = write(sock, request.data() + request_bytes_sent, to_send);
-                       } while (ret == -1 && errno == EINTR);
-
-                       if (ret == -1) {
-                               perror("write");
-                               state = CLOSING_SOCKET;
-                               continue;
-                       }
-
-                       assert(ret >= 0);
-                       request_bytes_sent += ret;
-
-                       if (request_bytes_sent == request.size()) {
-                               state = RECEIVING_HEADER;
-                       }
-                       break;
-               }
-               case RECEIVING_HEADER: {
-                       char buf[4096];
-                       int ret;
-
-                       do {
-                               ret = read(sock, buf, sizeof(buf));
-                       } while (ret == -1 && errno == EINTR);
-
-                       if (ret == -1) {
-                               perror("read");
-                               state = CLOSING_SOCKET;
-                               continue;
-                       }
-
-                       if (ret == 0) {
-                               // This really shouldn't happen...
-                               fprintf(stderr, "Socket unexpectedly closed while reading header\n");
-                               state = CLOSING_SOCKET;
-                               continue;
-                       }
-                       
-                       RequestParseStatus status = wait_for_double_newline(&response, buf, ret);
-                       
-                       if (status == RP_OUT_OF_SPACE) {
-                               fprintf(stderr, "WARNING: fd %d sent overlong response!\n", sock);
-                               state = CLOSING_SOCKET;
-                               continue;
-                       } else if (status == RP_NOT_FINISHED_YET) {
-                               continue;
-                       }
-       
-                       // OK, so we're fine, but there might be some of the actual data after the response.
-                       // We'll need to deal with that separately.
-                       string extra_data;
-                       if (status == RP_EXTRA_DATA) {
-                               char *ptr = static_cast<char *>(
-                                       memmem(response.data(), response.size(), "\r\n\r\n", 4));
-                               assert(ptr != NULL);
-                               extra_data = string(ptr, &response[0] + response.size());
-                               response.resize(ptr - response.data());
-                       }
-
-                       if (!parse_response(response)) {
-                               state = CLOSING_SOCKET;
-                               continue;
-                       }
-
-                       if (!extra_data.empty()) {
-                               process_data(&extra_data[0], extra_data.size());
-                       }
-
-                       state = RECEIVING_DATA;
-                       break;
-               }
-               case RECEIVING_DATA: {
-                       char buf[4096];
-                       int ret;
-
-                       do {
-                               ret = read(sock, buf, sizeof(buf));
-                       } while (ret == -1 && errno == EINTR);
-
-                       if (ret == -1) {
-                               perror("read");
-                               state = CLOSING_SOCKET;
-                               continue;
-                       }
-
-                       if (ret == 0) {
-                               // This really shouldn't happen...
-                               fprintf(stderr, "Socket unexpectedly closed while reading header\n");
-                               state = CLOSING_SOCKET;
-                               continue;
-                       }
-
-                       process_data(buf, ret);
-                       break;
-               }
-               case CLOSING_SOCKET: {
-                       int err;
-                       do {
-                               err = close(sock);
-                       } while (err == -1 && errno == EINTR);
-
-                       if (err == -1) {
-                               perror("close");
-                       }
-
-                       state = NOT_CONNECTED;
-                       break;
-               }
-               default:
-                       assert(false);
-               }
-
-               // If we are still in NOT_CONNECTED, either something went wrong,
-               // or the connection just got closed.
-               // The earlier steps have already given the error message, if any.
-               if (state == NOT_CONNECTED && !should_stop) {
-                       fprintf(stderr, "Waiting 0.2 second and restarting...\n");
-                       usleep(200000);
-               }
-       }
-}
-
-void Input::process_data(char *ptr, size_t bytes)
-{
-       pending_data.insert(pending_data.end(), ptr, ptr + bytes);
-
-       for ( ;; ) {
-               // If we don't have enough data (yet) for even the Metacube header, just return.
-               if (pending_data.size() < sizeof(metacube_block_header)) {
-                       return;
-               }
-
-               // Make sure we have the Metacube sync header at the start.
-               // We may need to skip over junk data (it _should_ not happen, though).
-               if (!has_metacube_header) {
-                       char *ptr = static_cast<char *>(
-                               memmem(pending_data.data(), pending_data.size(),
-                                      METACUBE_SYNC, strlen(METACUBE_SYNC)));
-                       if (ptr == NULL) {
-                               // OK, so we didn't find the sync marker. We know then that
-                               // we do not have the _full_ marker in the buffer, but we
-                               // could have N-1 bytes. Drop everything before that,
-                               // and then give up.
-                               drop_pending_data(pending_data.size() - (strlen(METACUBE_SYNC) - 1));
-                               return;
-                       } else {
-                               // Yay, we found the header. Drop everything (if anything) before it.
-                               drop_pending_data(ptr - pending_data.data());
-                               has_metacube_header = true;
-
-                               // Re-check that we have the entire header; we could have dropped data.
-                               if (pending_data.size() < sizeof(metacube_block_header)) {
-                                       return;
-                               }
-                       }
-               }
-
-               // Now it's safe to read the header.
-               metacube_block_header *hdr = reinterpret_cast<metacube_block_header *>(pending_data.data());    
-               assert(memcmp(hdr->sync, METACUBE_SYNC, sizeof(hdr->sync)) == 0);
-               uint32_t size = ntohl(hdr->size);
-               uint32_t flags = ntohl(hdr->flags);
-
-               // See if we have the entire block. If not, wait for more data.
-               if (pending_data.size() < sizeof(metacube_block_header) + size) {
-                       return;
-               }
-
-               // Send this block on to the data.
-               char *inner_data = pending_data.data() + sizeof(metacube_block_header);
-               if (flags & METACUBE_FLAGS_HEADER) {
-                       string header(inner_data, inner_data + size);
-                       servers->set_header(stream_id, http_header + header);
-               } else { 
-                       servers->add_data(stream_id, inner_data, size);
-               }
-
-               // Consume the block. This isn't the most efficient way of dealing with things
-               // should we have many blocks, but these routines don't need to be too efficient
-               // anyway.
-               pending_data.erase(pending_data.begin(), pending_data.begin() + sizeof(metacube_block_header) + size);
-               has_metacube_header = false;
-       }
-}
-
-void Input::drop_pending_data(size_t num_bytes)
-{
-       if (num_bytes == 0) {
-               return;
-       }
-       fprintf(stderr, "Warning: Dropping %lld junk bytes from stream, maybe it is not a Metacube stream?\n",
-               (long long)num_bytes);
-       pending_data.erase(pending_data.begin(), pending_data.begin() + num_bytes);
-}
+Input::~Input() {}
 
diff --git a/input.h b/input.h
index 14537c8..3e86510 100644 (file)
--- a/input.h
+++ b/input.h
@@ -1,79 +1,20 @@
 #ifndef _INPUT_H
 #define _INPUT_H 1
 
-#include <vector>
 #include <string>
 
 #include "thread.h"
 
 class InputProto;
 
+// Extremely rudimentary URL parsing.
+bool parse_url(const std::string &url, std::string *protocol, std::string *host, std::string *port, std::string *path);
+
 class Input : public Thread {
 public:
-       Input(const std::string &stream_id, const std::string &url);
-
-       // Serialization/deserialization.
-       Input(const InputProto &serialized);
-       InputProto serialize() const;
-
-       std::string get_url() const { return url; }
-
-private:
-       // Actually does the download.
-       virtual void do_work();
-       
-       // Open a socket that connects to the given host and port. Does DNS resolving.
-       int lookup_and_connect(const std::string &host, const std::string &port);
-
-       // Parses a HTTP response. Returns false if it not a 200.
-       bool parse_response(const std::string &response);
-
-       // Stores the given data, looks for Metacube blocks (skipping data if needed),
-       // and calls process_block() for each one.
-       void process_data(char *ptr, size_t bytes);
-
-       // Drops <num_bytes> bytes from the head of <pending_data>,
-       // and outputs a warning.
-       void drop_pending_data(size_t num_bytes);
-
-       enum State {
-               NOT_CONNECTED,
-               SENDING_REQUEST,
-               RECEIVING_HEADER,
-               RECEIVING_DATA,
-               CLOSING_SOCKET,  // Due to error.
-       };
-       State state;
-
-       std::string stream_id;
-
-       // The URL and its parsed components.
-       std::string url;
-       std::string host, port, path;
-
-       // The HTTP request, with headers and all.
-       // Only relevant for SENDING_REQUEST.
-       std::string request;
-
-       // How many bytes we've sent of the request so far.
-       // Only relevant for SENDING_REQUEST.
-       size_t request_bytes_sent;
-
-       // The HTTP response we've received so far. Only relevant for RECEIVING_HEADER.
-       std::string response;
-
-       // The HTTP respones headers we want to give clients for this input.
-       std::string http_header;
-
-       // Data we have received but not fully processed yet.
-       std::vector<char> pending_data;
-
-       // If <pending_data> starts with a Metacube header,
-       // this is true.
-       bool has_metacube_header;
-
-       // The socket we are downloading on (or -1).
-       int sock;       
+       virtual ~Input();
+       virtual InputProto serialize() const = 0;
+       virtual std::string get_url() const = 0;
 };
 
 #endif  // !defined(_INPUT_H)
index db3be63..04d0d3d 100644 (file)
--- a/main.cpp
+++ b/main.cpp
@@ -25,6 +25,7 @@
 #include "server.h"
 #include "serverpool.h"
 #include "input.h"
+#include "httpinput.h"
 #include "stats.h"
 #include "state.pb.h"
 
@@ -267,7 +268,7 @@ vector<Input *> create_inputs(const vector<ConfigLine> &config,
                        deserialized_inputs->erase(deserialized_input_it);
                }
                if (input == NULL) {
-                       input = new Input(stream_id, src);
+                       input = new HTTPInput(stream_id, src);
                }
                input->run();
                inputs.push_back(input);
@@ -358,7 +359,7 @@ int main(int argc, char **argv)
                for (int i = 0; i < loaded_state.inputs_size(); ++i) {
                        deserialized_inputs.insert(make_pair(
                                loaded_state.inputs(i).stream_id(),
-                               new Input(loaded_state.inputs(i))));
+                               new HTTPInput(loaded_state.inputs(i))));
                } 
 
                // Convert the acceptor from older serialized formats.