httpgw.cpp

   1 #include "swift.h"
   2
   3 using namespace swift;
   4
   5 #define HTTPGW_MAX_CLIENT 128
   6
   7 enum {
   8     HTTPGW_RANGE=0,
   9     HTTPGW_MAX_HEADER=1
  10 };
  11 const char * HTTPGW_HEADERS[HTTPGW_MAX_HEADER] = {
  12     "Content-Range"
  13 };
  14
  15
  16 struct http_gw_t {
  17     int      id;
  18     uint64_t offset;
  19     uint64_t tosend;
  20     int      transfer;
  21     SOCKET   sink;
  22     char*    headers[HTTPGW_MAX_HEADER];
  23 } http_requests[HTTPGW_MAX_CLIENT];
  24
  25
  26 int http_gw_reqs_open = 0;
  27 int http_gw_reqs_count = 0;
  28
  29 void HttpGwNewRequestCallback (SOCKET http_conn);
  30 void HttpGwNewRequestCallback (SOCKET http_conn);
  31
  32 http_gw_t* HttpGwFindRequest (SOCKET sock) {
  33     for(int i=0; i<http_gw_reqs_open; i++)
  34         if (http_requests[i].sink==sock)
  35             return http_requests+i;
  36     return NULL;
  37 }
  38
  39
  40 void HttpGwCloseConnection (SOCKET sock) {
  41     http_gw_t* req = HttpGwFindRequest(sock);
  42     if (req) {
  43         dprintf("%s @%i closed http connection %i\n",tintstr(),req->id,sock);
  44         for(int i=0; i<HTTPGW_MAX_HEADER; i++)
  45             if (req->headers[i]) {
  46                 free(req->headers[i]);
  47                 req->headers[i] = NULL;
  48             }
  49         *req = http_requests[--http_gw_reqs_open];
  50     }
  51     swift::close_socket(sock);
  52     swift::Datagram::Listen3rdPartySocket(sckrwecb_t(sock));
  53 }
  54
  55
  56 void HttpGwMayWriteCallback (SOCKET sink) {
  57     http_gw_t* req = HttpGwFindRequest(sink);
  58     uint64_t complete = swift::SeqComplete(req->transfer);
  59     if (complete>req->offset) { // send data
  60         char buf[1<<12];
  61         uint64_t tosend = std::min((uint64_t)1<<12,complete-req->offset);
  62         size_t rd = pread(req->transfer,buf,tosend,req->offset); // hope it is cached
  63         if (rd<0) {
  64             HttpGwCloseConnection(sink);
  65             return;
  66         }
  67         int wn = send(sink, buf, rd, 0);
  68         if (wn<0) {
  69             print_error("send fails");
  70             HttpGwCloseConnection(sink);
  71             return;
  72         }
  73         dprintf("%s @%i sent %ib\n",tintstr(),req->id,(int)wn);
  74         req->offset += wn;
  75         req->tosend -= wn;
  76     } else {
  77         if (req->tosend==0) { // done; wait for new request
  78             dprintf("%s @%i done\n",tintstr(),req->id);
  79             sckrwecb_t wait_new_req
  80             //  (req->sink,HttpGwNewRequestCallback,NULL,HttpGwCloseConnection);
  81               (req->sink,NULL,NULL,NULL);
  82             HttpGwCloseConnection(sink);
  83             swift::Datagram::Listen3rdPartySocket (wait_new_req);
  84         } else { // wait for data
  85             dprintf("%s @%i waiting for data\n",tintstr(),req->id);
  86             sckrwecb_t wait_swift_data(req->sink,NULL,NULL,HttpGwCloseConnection);
  87             swift::Datagram::Listen3rdPartySocket(wait_swift_data);
  88         }
  89     }
  90 }
  91
  92
  93 void HttpGwSwiftProgressCallback (int transfer, bin64_t bin) {
  94     dprintf("%s @A pcb: %s\n",tintstr(),bin.str());
  95     for (int httpc=0; httpc<http_gw_reqs_open; httpc++)
  96         if (http_requests[httpc].transfer==transfer)
  97             if ( (bin.base_offset()<<10) <= http_requests[httpc].offset &&
  98                   ((bin.base_offset()+bin.width())<<10) > http_requests[httpc].offset  ) {
  99                 dprintf("%s @%i progress: %s\n",tintstr(),http_requests[httpc].id,bin.str());
 100                 sckrwecb_t maywrite_callbacks
 101                         (http_requests[httpc].sink,NULL,
 102                          HttpGwMayWriteCallback,HttpGwCloseConnection);
 103                 Datagram::Listen3rdPartySocket (maywrite_callbacks);
 104             }
 105 }
 106
 107
 108 void HttpGwFirstProgressCallback (int transfer, bin64_t bin) {
 109     if (bin!=bin64_t(0,0)) // need the first packet
 110         return;
 111     swift::RemoveProgressCallback(transfer,&HttpGwFirstProgressCallback);
 112     swift::AddProgressCallback(transfer,&HttpGwSwiftProgressCallback,0);
 113     for (int httpc=0; httpc<http_gw_reqs_open; httpc++) {
 114         http_gw_t * req = http_requests + httpc;
 115         if (req->transfer==transfer && req->tosend==0) { // FIXME states
 116             uint64_t file_size = swift::Size(transfer);
 117             char response[1024];
 118             sprintf(response,
 119                 "HTTP/1.1 200 OK\r\n"\
 120                 "Connection: keep-alive\r\n"\
 121                 "Content-Type: video/ogg\r\n"\
 122                 /*"X-Content-Duration: 32\r\n"*/\
 123                 "Content-Length: %lli\r\n"\
 124                 "Accept-Ranges: none\r\n"\
 125                 "\r\n",
 126                 file_size);
 127             send(req->sink,response,strlen(response),0);
 128             req->tosend = file_size;
 129             dprintf("%s @%i headers_sent size %lli\n",tintstr(),req->id,file_size);
 130         }
 131     }
 132     HttpGwSwiftProgressCallback(transfer,bin);
 133 }
 134
 135
 136 void HttpGwNewRequestCallback (SOCKET http_conn){
 137     http_gw_t* req = http_requests + http_gw_reqs_open++;
 138     req->id = ++http_gw_reqs_count;
 139     req->sink = http_conn;
 140     req->offset = 0;
 141     req->tosend = 0;
 142     dprintf("%s @%i new http request\n",tintstr(),req->id);
 143     // read headers - the thrilling part
 144     // we surely do not support pipelining => one request at a time
 145     #define HTTPGW_MAX_REQ_SIZE 1024
 146     char buf[HTTPGW_MAX_REQ_SIZE+1];
 147     int rd = recv(http_conn,buf,HTTPGW_MAX_REQ_SIZE,0);
 148     if (rd<=0) { // if conn is closed by the peer, rd==0
 149         HttpGwCloseConnection(http_conn);
 150         return;
 151     }
 152     buf[rd] = 0;
 153     // HTTP request line
 154     char* reqline = strtok(buf,"\r\n");
 155     char method[16], url[512], version[16], crlf[5];
 156     if (3!=sscanf(reqline,"%16s %512s %16s",method,url,version)) {
 157         HttpGwCloseConnection(http_conn);
 158         return;
 159     }
 160     // HTTP header fields
 161     char* headerline;
 162     while (headerline=strtok(NULL,"\n\r")) {
 163         char header[128], value[256];
 164         if (2!=sscanf(headerline,"%120[^: ]: %250[^\r\n]",header,value)) {
 165             HttpGwCloseConnection(http_conn);
 166             return;
 167         }
 168         for(int i=0; i<HTTPGW_MAX_HEADER; i++)
 169             if (0==strcasecmp(HTTPGW_HEADERS[i],header) && !req->headers[i])
 170                 req->headers[i] = strdup(value);
 171     }
 172     // parse URL
 173     char * hashch=strtok(url,"/"), hash[41];
 174     while (hashch && (1!=sscanf(hashch,"%40[0123456789abcdefABCDEF]",hash) || strlen(hash)!=40))
 175         hashch = strtok(NULL,"/");
 176     if (strlen(hash)!=40) {
 177         HttpGwCloseConnection(http_conn);
 178         return;
 179     }
 180     dprintf("%s @%i demands %s\n",tintstr(),req->id,hash);
 181     // initiate transmission
 182     Sha1Hash root_hash = Sha1Hash(true,hash);
 183     int file = swift::Find(root_hash);
 184     if (file==-1)
 185         file = swift::Open(hash,root_hash);
 186     req->transfer = file;
 187     if (swift::Size(file)) {
 188         HttpGwFirstProgressCallback(file,bin64_t(0,0));
 189     } else {
 190         swift::AddProgressCallback(file,&HttpGwFirstProgressCallback,0);
 191         sckrwecb_t install (http_conn,NULL,NULL,HttpGwCloseConnection);
 192         swift::Datagram::Listen3rdPartySocket(install);
 193     }
 194 }
 195
 196
 197 // be liberal in what you do, be conservative in what you accept
 198 void HttpGwNewConnectionCallback (SOCKET serv) {
 199     Address client_address;
 200     socklen_t len;
 201     SOCKET conn = accept (serv, (sockaddr*) & (client_address.addr), &len);
 202     if (conn==INVALID_SOCKET) {
 203         print_error("client conn fails");
 204         return;
 205     }
 206     make_socket_nonblocking(conn);
 207     // submit 3rd party socket to the swift loop
 208     sckrwecb_t install
 209         (conn,HttpGwNewRequestCallback,NULL,HttpGwCloseConnection);
 210     swift::Datagram::Listen3rdPartySocket(install);
 211 }
 212
 213
 214 void HttpGwError (SOCKET s) {
 215     print_error("httpgw is dead");
 216     dprintf("%s @0 closed http gateway\n",tintstr());
 217     close_socket(s);
 218     swift::Datagram::Listen3rdPartySocket(sckrwecb_t(s));
 219 }
 220
 221
 222 #include <signal.h>
 223 SOCKET InstallHTTPGateway (Address bind_to) {
 224     SOCKET fd;
 225     #define gw_ensure(x) { if (!(x)) { \
 226     print_error("http binding fails"); close_socket(fd); \
 227     return INVALID_SOCKET; } }
 228     gw_ensure ( (fd=socket(AF_INET, SOCK_STREAM, 0)) != INVALID_SOCKET );
 229     int enable = true;
 230     setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, (setsockoptptr_t)&enable, sizeof(int));
 231     //setsockopt(fd, SOL_SOCKET, SO_NOSIGPIPE, (setsockoptptr_t)&enable, sizeof(int));
 232     //struct sigaction act;
 233     //memset(&act,0,sizeof(struct sigaction));
 234     //act.sa_handler = SIG_IGN;
 235     //sigaction (SIGPIPE, &act, NULL); // FIXME
 236     signal( SIGPIPE, SIG_IGN );
 237     gw_ensure ( 0==bind(fd, (sockaddr*)&(bind_to.addr), sizeof(struct sockaddr_in)) );
 238     gw_ensure (make_socket_nonblocking(fd));
 239     gw_ensure ( 0==listen(fd,8) );
 240     sckrwecb_t install_http(fd,HttpGwNewConnectionCallback,NULL,HttpGwError);
 241     gw_ensure (swift::Datagram::Listen3rdPartySocket(install_http));
 242     dprintf("%s @0 installed http gateway on %s\n",tintstr(),bind_to.str());
 243     return fd;
 244 }