0
@@ -45,6 +45,7 @@ module Ebb
0
   end
0
   
0
   def self.process(app, client)
0
+    #p client.env
0
     status, headers, body = app.call(client.env)
0
     
0
     # Write the status
0
@@ -52,9 +53,9 @@ module Ebb
0
     
0
     # Add Content-Length to the headers.
0
     if headers.respond_to?(:[]=) and body.respond_to?(:length) and status != 304
0
-      headers['Connection'] = 'close'
0
       headers['Content-Length'] = body.length.to_s
0
     end
0
+    headers['Connection'] = client.keep_alive? ? 'Keep-Alive' : 'close'
0
     
0
     # Write the headers
0
     headers.each { |field, value| client.write_header(field, value) }
0
@@ -125,6 +126,10 @@ module Ebb
0
     def release
0
       FFI::client_release(self)
0
     end
0
+    
0
+    def keep_alive?
0
+      FFI::client_keep_alive?(self)
0
+    end
0
   end
0
   
0
   class RequestBody

0
@@ -30,6 +30,14 @@
0
 #define ramp(a) (a > 0 ? a : 0)
0
 
0
 static int server_socket_unix(const char *path, int access_mask);
0
+static void client_init(ebb_client *client);
0
+
0
+static void set_nonblock(int fd)
0
+{
0
+  int flags = fcntl(fd, F_GETFL, 0);
0
+  assert(0 <= fcntl(fd, F_SETFL, flags | O_NONBLOCK) && "Setting socket non-block failed!");
0
+}
0
+
0
 
0
 void env_add(ebb_client *client, const char *field, int flen, const char *value, int vlen)
0
 {
0
@@ -73,6 +81,16 @@ void http_field_cb(void *data, const char *field, size_t flen, const char *value
0
 void on_element(void *data, int type, const char *at, size_t length)
0
 {
0
   ebb_client *client = (ebb_client*)(data);
0
+  switch(type) {
0
+  case MONGREL_HTTP_VERSION:
0
+    /* "HTTP/1.1" by default is keep-alive true, otherwise not*/
0
+    /* note that the version will always come first */
0
+    client->keep_alive = (strncmp(at+5, "1.1", 3) == 0);
0
+    break;
0
+  case MONGREL_CONNECTION:
0
+    client->keep_alive = (strncmp(at, "close", 5) != 0);
0
+    break;
0
+  }
0
   env_add_const(client, type, at, length);
0
 }
0
 
0
@@ -83,6 +101,12 @@ static void dispatch(ebb_client *client)
0
   if(client->open == FALSE)
0
     return;
0
   client->in_use = TRUE;
0
+  
0
+  /* decide if to use keep-alive or not */
0
+  
0
+
0
+  
0
+  
0
   server->request_cb(client, server->request_cb_data);
0
 }
0
 
0
@@ -118,9 +142,9 @@ static void* read_body_into_file(void *_client)
0
   int flags = fcntl(client->fd, F_GETFL, 0);
0
   assert(0 <= fcntl(client->fd, F_SETFL, flags & ~O_NONBLOCK));
0
   
0
-  sprintf(client->upload_file_filename, "/tmp/ebb_upload_%010d", id++);
0
-  tmpfile = fopen(client->upload_file_filename, "w+");
0
-  if(tmpfile == NULL) g_message("Cannot open tmpfile %s", client->upload_file_filename);
0
+  sprintf(client->upload_filename, "/tmp/ebb_upload_%010d", id++);
0
+  tmpfile = fopen(client->upload_filename, "w+");
0
+  if(tmpfile == NULL) g_message("Cannot open tmpfile %s", client->upload_filename);
0
   client->upload_file = tmpfile;
0
   
0
   size_t body_head_length = client->read - client->parser.nread;
0
@@ -164,7 +188,7 @@ static void* read_body_into_file(void *_client)
0
     written += received;
0
   }
0
   rewind(tmpfile);
0
-  // g_debug("%d bytes written to file %s", written, client->upload_file_filename);
0
+  // g_debug("%d bytes written to file %s", written, client->upload_filename);
0
   dispatch(client);
0
   return NULL;
0
 error:
0
@@ -188,7 +212,7 @@ static void on_client_readable(struct ev_loop *loop, ev_io *watcher, int revents
0
                      , EBB_BUFFERSIZE - client->read
0
                      , 0
0
                      );
0
-  if(read < 0) goto error; 
0
+  if(read < 0) goto error;
0
   if(read == 0) goto error; /* XXX is this the right action to take for read==0 ? */
0
   client->read += read;
0
   ev_timer_again(loop, &client->timeout_watcher);
0
@@ -222,7 +246,9 @@ static void on_client_readable(struct ev_loop *loop, ev_io *watcher, int revents
0
   }
0
   return;
0
 error:
0
+#ifdef DEBUG
0
   if(read < 0) g_message("Error recving data: %s", strerror(errno));
0
+#endif
0
   ebb_client_close(client);
0
 }
0
 
0
@@ -265,12 +291,17 @@ static void on_client_writable(struct ev_loop *loop, ev_io *watcher, int revents
0
   assert(client->written <= client->response_buffer->len);
0
   //g_message("wrote %d bytes. total: %d", (int)sent, (int)(client->written));
0
   
0
-  ev_timer_again(loop, &(client->timeout_watcher));
0
+  ev_timer_again(loop, &client->timeout_watcher);
0
   
0
   if(client->written == client->response_buffer->len) {
0
+    /* stop the write watcher. to be restarted by the next call to ebb_client_write_body
0
+     * or if client->body_written is set (by using ebb_client_release) then
0
+     * we close the connection
0
+     */
0
     ev_io_stop(loop, watcher);
0
-    if(client->body_written)
0
-      ebb_client_close(client);
0
+    if(client->body_written) {
0
+      client->keep_alive ? client_init(client) : ebb_client_close(client);
0
+    }
0
   }
0
   return;
0
 error:
0
@@ -278,15 +309,26 @@ error:
0
 }
0
 
0
 
0
-static client_init(ebb_client *client, int fd)
0
+static void client_init(ebb_client *client)
0
 {
0
   assert(client->in_use == FALSE);
0
   
0
-  client->open = TRUE;
0
-  client->fd = fd;
0
+  /* If the client is already open, reuse the fd, just reset all the parameters
0
+   * this would happen in the case of a keep_alive request
0
+   */
0
+  if(!client->open) {
0
+    /* DO SOCKET STUFF */
0
+    socklen_t len;
0
+    int fd = accept(client->server->fd, (struct sockaddr*)&(client->server->sockaddr), &len);
0
+    if(fd < 0) {
0
+      perror("accept()");
0
+      return;
0
+    }
0
+    client->open = TRUE;
0
+    client->fd = fd;
0
+  }
0
   
0
-  int flags = fcntl(client->fd, F_GETFL, 0);
0
-  assert(0 <= fcntl(client->fd, F_SETFL, flags | O_NONBLOCK));
0
+  set_nonblock(client->fd);
0
   
0
   /* INITIALIZE http_parser */
0
   http_parser_init(&client->parser);
0
@@ -297,15 +339,18 @@ static client_init(ebb_client *client, int fd)
0
   /* OTHER */
0
   client->env_size = 0;
0
   client->read = client->nread_from_body = 0;
0
-  client->response_buffer->len = 0; /* see note in ebb_client_close */
0
   if(client->request_buffer == NULL) {
0
+    /* Only allocate the request_buffer once */
0
     client->request_buffer = (char*)malloc(EBB_BUFFERSIZE);
0
   }
0
-  
0
-  client->status_written = FALSE;
0
-  client->headers_written = FALSE;
0
-  client->body_written = FALSE;
0
+  client->keep_alive = TRUE;
0
+  client->status_written = client->headers_written = client->body_written = FALSE;
0
   client->written = 0;
0
+  /* here we do not free the already allocated GString client->response_buffer
0
+   * that we're holding the response in. we reuse it again - presumably 
0
+   * because the backend is going to keep sending such long requests.
0
+   */
0
+  client->response_buffer->len = 0;
0
   
0
   /* SETUP READ AND TIMEOUT WATCHERS */
0
   client->write_watcher.data = client;
0
@@ -325,12 +370,6 @@ static client_init(ebb_client *client, int fd)
0
 }
0
 
0
 
0
-static client_reinit(ebb_client *client)
0
-{
0
-  client_init(client, client->fd);
0
-}
0
-
0
-
0
 static void on_request(struct ev_loop *loop, ev_io *watcher, int revents)
0
 {
0
   ebb_server *server = (ebb_server*)(watcher->data);
0
@@ -368,15 +407,7 @@ static void on_request(struct ev_loop *loop, ev_io *watcher, int revents)
0
   g_debug("%d open connections", count);
0
 #endif
0
   
0
-  /* DO SOCKET STUFF */
0
-  socklen_t len;
0
-  int client_fd = accept(server->fd, (struct sockaddr*)&(server->sockaddr), &len);
0
-  if(client_fd < 0) {
0
-    perror("accept()");
0
-    return;
0
-  }
0
-  
0
-  client_init(client, client_fd);
0
+  client_init(client);
0
 }
0
 
0
 
0
@@ -454,11 +485,7 @@ int ebb_server_listen_on_port(ebb_server *server, const int port)
0
     goto error;
0
   }
0
   
0
-  flags = fcntl(sfd, F_GETFL, 0);
0
-  if(fcntl(sfd, F_SETFL, flags | O_NONBLOCK) < 0) {
0
-    perror("setting O_NONBLOCK");
0
-    goto error;
0
-  }
0
+  set_nonblock(sfd);
0
   
0
   flags = 1;
0
   setsockopt(sfd, SOL_SOCKET, SO_REUSEADDR, (void *)&flags, sizeof(flags));
0
@@ -532,16 +559,9 @@ void ebb_client_close(ebb_client *client)
0
     
0
     if(client->upload_file) {
0
       fclose(client->upload_file);
0
-      unlink(client->upload_file_filename);
0
+      unlink(client->upload_filename);
0
     }
0
     
0
-    /* here we do not free the already allocated GString client->response_buffer
0
-     * that we're holding the response in. we reuse it again - 
0
-     * presumably because the backend is going to keep sending such long
0
-     * requests.
0
-     */
0
-    client->response_buffer->len = 0;
0
-    
0
     close(client->fd);
0
     client->open = FALSE;
0
   }
0
@@ -582,6 +602,7 @@ void ebb_client_write_body(ebb_client *client, const char *data, int length)
0
   
0
   if(client->headers_written == FALSE) {
0
     g_string_append(client->response_buffer, "\r\n");
0
+    client->headers_written = TRUE;
0
   }
0
   
0
   g_string_append_len(client->response_buffer, data, length);
0
@@ -590,11 +611,7 @@ void ebb_client_write_body(ebb_client *client, const char *data, int length)
0
    * we're streaming and the watcher has been stopped. In that case we 
0
    * start it again since we have more to write. */
0
   if(ev_is_active(&client->write_watcher) == FALSE) {
0
-    /* assure the socket is still in non-blocking mode */
0
-    int flags = fcntl(client->fd, F_GETFL, 0);
0
-    if(0 > fcntl(client->fd, F_SETFL, flags | O_NONBLOCK))
0
-      perror("fcntl() setting non-block");
0
-    client->headers_written = TRUE;
0
+    set_nonblock(client->fd);
0
     ev_io_start(client->server->loop, &client->write_watcher);
0
   }
0
 }

0
@@ -54,7 +54,7 @@ struct ebb_client {
0
   ev_io read_watcher;
0
   size_t read, nread_from_body;
0
   
0
-  char upload_file_filename[200];
0
+  char upload_filename[200];
0
   FILE *upload_file;
0
   
0
   ev_io write_watcher;
0
@@ -62,7 +62,7 @@ struct ebb_client {
0
   size_t written;
0
   
0
   ev_timer timeout_watcher;
0
-  
0
+  unsigned int keep_alive : 1;
0
   unsigned int status_written : 1;
0
   unsigned int headers_written : 1;
0
   unsigned int body_written : 1;

0
@@ -239,6 +239,12 @@ VALUE client_release(VALUE _, VALUE rb_client)
0
   return Qnil;
0
 }
0
 
0
+VALUE client_keep_alive_p(VALUE _, VALUE rb_client)
0
+{
0
+  ebb_client *client;
0
+  Data_Get_Struct(rb_client, ebb_client, client);
0
+  return client->keep_alive ? Qtrue : Qfalse;
0
+}
0
 
0
 void Init_ebb_ext()
0
 {
0
@@ -274,6 +280,7 @@ void Init_ebb_ext()
0
   rb_define_singleton_method(mFFI, "client_write_status", client_write_status, 3);
0
   rb_define_singleton_method(mFFI, "client_write_header", client_write_header, 3);
0
   rb_define_singleton_method(mFFI, "client_write_body", client_write_body, 2);
0
+  rb_define_singleton_method(mFFI, "client_keep_alive?", client_keep_alive_p, 1);
0
   rb_define_singleton_method(mFFI, "client_env", client_env, 1);
0
   rb_define_singleton_method(mFFI, "client_release", client_release, 1);
0