0
@@ -100,6 +100,8 @@ class ServerTest
0
       @pid = fork { start_mongrel }
0
     when 'thin'
0
       @pid = fork { start_thin }
0
+    when 'fcgi'
0
+      @pid = fork { start_fcgi }
0
     end
0
   end
0
   
0
@@ -130,6 +132,10 @@ class ServerTest
0
     Rack::Handler::Thin.run(app, :Port => @port)
0
   end
0
   
0
+  def start_fcgi
0
+    Rack::Handler::FastCGI.run(app, :Port => @port)
0
+  end
0
+  
0
   def trial(ab_cmd)
0
     cmd = ab_cmd.sub('PORT', @port.to_s)
0
     

0
@@ -9,10 +9,14 @@ module Ebb
0
   
0
   def self.start_server(app, options={})
0
     port = (options[:port] || 4001).to_i
0
+    
0
     nworkers = options[:workers] || 1
0
+    Client::BASE_ENV['rack.multithread'] = true if nworkers > 1
0
+    
0
     # socket = options[:socket]
0
     # timeout =  options[:timeout]
0
     
0
+    
0
     FFI::server_listen_on_port(port)
0
     
0
     puts "Ebb listening at http://0.0.0.0:#{port}/ (#{nworkers} worker#{'s' if nworkers > 1})"
0
@@ -64,10 +68,13 @@ module Ebb
0
     # until the rest of ebb is more developed.
0
     if body.kind_of?(String)
0
       client.write body
0
+      client.body_written = true
0
+      client.begin_transmission
0
     else
0
-      body.each { |p| client.write p }
0
+      client.begin_transmission
0
+      body.each { |p| client.write(p) }
0
+      client.body_written = true
0
     end
0
-    client.finished
0
   end
0
   
0
   def FFI.waiting_clients
0
@@ -95,8 +102,12 @@ module Ebb
0
       end
0
     end
0
     
0
-    def finished
0
-      FFI::client_finished(self)
0
+    def begin_transmission
0
+      FFI::client_begin_transmission(self)
0
+    end
0
+    
0
+    def body_written=(v)
0
+      FFI::client_set_body_written(self, v)
0
     end
0
     
0
     def write(data)

0
@@ -29,10 +29,9 @@
0
 #define min(a,b) (a < b ? a : b)
0
 #define ramp(a) (a > 0 ? a : 0)
0
 
0
-static int server_socket(const int port);
0
 static int server_socket_unix(const char *path, int access_mask);
0
 
0
-void env_add(ebb_client *client, const char *field, int flen, const char *value, int vlen)
0
+static void env_add(ebb_client *client, const char *field, int flen, const char *value, int vlen)
0
 {
0
   if(client->env_size >= EBB_MAX_ENV) {
0
     client->parser.overflow_error = TRUE;
0
@@ -47,7 +46,7 @@ void env_add(ebb_client *client, const char *field, int flen, const char *value,
0
 }
0
 
0
 
0
-void env_add_const(ebb_client *client, int type, const char *value, int vlen)
0
+static void env_add_const(ebb_client *client, int type, const char *value, int vlen)
0
 {
0
   if(client->env_size >= EBB_MAX_ENV) {
0
     client->parser.overflow_error = TRUE;
0
@@ -62,7 +61,7 @@ void env_add_const(ebb_client *client, int type, const char *value, int vlen)
0
 }
0
 
0
 
0
-void http_field_cb(void *data, const char *field, size_t flen, const char *value, size_t vlen)
0
+static void http_field_cb(void *data, const char *field, size_t flen, const char *value, size_t vlen)
0
 {
0
   ebb_client *client = (ebb_client*)(data);
0
   assert(field != NULL);
0
@@ -71,49 +70,49 @@ void http_field_cb(void *data, const char *field, size_t flen, const char *value
0
 }
0
 
0
 
0
-void request_method_cb(void *data, const char *at, size_t length)
0
+static void request_method_cb(void *data, const char *at, size_t length)
0
 {
0
   ebb_client *client = (ebb_client*)(data);
0
   env_add_const(client, EBB_REQUEST_METHOD, at, length);
0
 }
0
 
0
 
0
-void request_uri_cb(void *data, const char *at, size_t length)
0
+static void request_uri_cb(void *data, const char *at, size_t length)
0
 {
0
   ebb_client *client = (ebb_client*)(data);
0
   env_add_const(client, EBB_REQUEST_URI, at, length);
0
 }
0
 
0
 
0
-void fragment_cb(void *data, const char *at, size_t length)
0
+static void fragment_cb(void *data, const char *at, size_t length)
0
 {
0
   ebb_client *client = (ebb_client*)(data);
0
   env_add_const(client, EBB_FRAGMENT, at, length);
0
 }
0
 
0
 
0
-void request_path_cb(void *data, const char *at, size_t length)
0
+static void request_path_cb(void *data, const char *at, size_t length)
0
 {
0
   ebb_client *client = (ebb_client*)(data);
0
   env_add_const(client, EBB_REQUEST_PATH, at, length);
0
 }
0
 
0
 
0
-void query_string_cb(void *data, const char *at, size_t length)
0
+static void query_string_cb(void *data, const char *at, size_t length)
0
 {
0
   ebb_client *client = (ebb_client*)(data);
0
   env_add_const(client, EBB_QUERY_STRING, at, length);
0
 }
0
 
0
 
0
-void http_version_cb(void *data, const char *at, size_t length)
0
+static void http_version_cb(void *data, const char *at, size_t length)
0
 {
0
   ebb_client *client = (ebb_client*)(data);
0
   env_add_const(client, EBB_HTTP_VERSION, at, length);
0
 }
0
 
0
 
0
-void content_length_cb(void *data, const char *at, size_t length)
0
+static void content_length_cb(void *data, const char *at, size_t length)
0
 {
0
   ebb_client *client = (ebb_client*)(data);
0
   env_add_const(client, EBB_CONTENT_LENGTH, at, length);
0
@@ -125,7 +124,7 @@ void content_length_cb(void *data, const char *at, size_t length)
0
 
0
 
0
 const char* localhost_str = "0.0.0.0";
0
-void dispatch(ebb_client *client)
0
+static void dispatch(ebb_client *client)
0
 {
0
   ebb_server *server = client->server;
0
   
0
@@ -147,7 +146,7 @@ void dispatch(ebb_client *client)
0
 }
0
 
0
 
0
-void on_timeout(struct ev_loop *loop, ev_timer *watcher, int revents)
0
+static void on_timeout(struct ev_loop *loop, ev_timer *watcher, int revents)
0
 {
0
   ebb_client *client = (ebb_client*)(watcher->data);
0
   
0
@@ -163,7 +162,7 @@ void on_timeout(struct ev_loop *loop, ev_timer *watcher, int revents)
0
 #define client_finished_parsing http_parser_is_finished(&client->parser)
0
 #define total_request_size (client->content_length + client->parser.nread)
0
 
0
-void* read_body_into_file(void *_client)
0
+static void* read_body_into_file(void *_client)
0
 {
0
   ebb_client *client = (ebb_client*)_client;
0
   static unsigned int id;
0
@@ -233,7 +232,7 @@ error:
0
 }
0
 
0
 
0
-void on_readable(struct ev_loop *loop, ev_io *watcher, int revents)
0
+static void on_readable(struct ev_loop *loop, ev_io *watcher, int revents)
0
 {
0
   ebb_client *client = (ebb_client*)(watcher->data);
0
   
0
@@ -285,58 +284,24 @@ error:
0
   ebb_client_close(client);
0
 }
0
 
0
-
0
-void on_request(struct ev_loop *loop, ev_io *watcher, int revents)
0
+static client_init(ebb_server *server, ebb_client *client)
0
 {
0
-  ebb_server *server = (ebb_server*)(watcher->data);
0
-  assert(server->open);
0
-  assert(server->loop == loop);
0
-  assert(&server->request_watcher == watcher);
0
-  
0
-  if(EV_ERROR & revents) {
0
-    g_message("on_request() got error event, closing server.");
0
-    ebb_server_unlisten(server);
0
-    return;
0
-  }
0
-  /* Now we're going to initialize the client 
0
-   * and set up her callbacks for read and write
0
-   * the client won't get passed back to the user, however,
0
-   * until the request is complete and parsed.
0
-   */
0
-  int i;
0
-  ebb_client *client;
0
-  /* Get next availible peer */
0
-  for(i=0; i < EBB_MAX_CLIENTS; i++)
0
-    if(!server->clients[i].open) {
0
-      client = &(server->clients[i]);
0
-      break;
0
-    }
0
-  if(client == NULL) {
0
-    g_message("Too many peers. Refusing connections.");
0
-    return;
0
-  }
0
-  
0
 #ifdef DEBUG
0
-  int count = 0;
0
-  for(i = 0; i < EBB_MAX_CLIENTS; i++)
0
-    if(server->clients[i].open) count += 1;
0
-  g_debug("%d open connections", count);
0
-  
0
   /* does ragel fuck up if request buffer isn't null? */
0
   for(i=0; i< EBB_BUFFERSIZE; i++)
0
     client->request_buffer[i] = 'A';
0
 #endif
0
-  
0
+
0
   client->open = TRUE;
0
   client->server = server;
0
-  
0
+
0
   /* DO SOCKET STUFF */
0
   socklen_t len;
0
   client->fd = accept(server->fd, (struct sockaddr*)&(server->sockaddr), &len);
0
   assert(client->fd >= 0);
0
   int flags = fcntl(client->fd, F_GETFL, 0);
0
   assert(0 <= fcntl(client->fd, F_SETFL, flags | O_NONBLOCK));
0
-  
0
+
0
   /* INITIALIZE http_parser */
0
   http_parser_init(&(client->parser));
0
   client->parser.data = client;
0
@@ -348,29 +313,70 @@ void on_request(struct ev_loop *loop, ev_io *watcher, int revents)
0
   client->parser.query_string   = query_string_cb;
0
   client->parser.http_version   = http_version_cb;
0
   client->parser.content_length = content_length_cb;
0
-  
0
+
0
   /* OTHER */
0
 
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
   client->content_length = 0;
0
-  
0
-  client->status_sent = FALSE;
0
-  client->headers_sent = FALSE;
0
-  client->body_sent = FALSE;
0
-  
0
+
0
+  client->status_written = FALSE;
0
+  client->headers_written = FALSE;
0
+  client->body_written = FALSE;
0
+  client->began_transmission = FALSE;
0
+
0
   /* SETUP READ AND TIMEOUT WATCHERS */
0
   client->read_watcher.data = client;
0
   ev_init(&client->read_watcher, on_readable);
0
   ev_io_set(&client->read_watcher, client->fd, EV_READ | EV_ERROR);
0
   ev_io_start(server->loop, &client->read_watcher);
0
-  
0
+
0
   client->timeout_watcher.data = client;  
0
   ev_timer_init(&client->timeout_watcher, on_timeout, EBB_TIMEOUT, EBB_TIMEOUT);
0
   ev_timer_start(server->loop, &client->timeout_watcher);
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
+  assert(server->open);
0
+  assert(server->loop == loop);
0
+  assert(&server->request_watcher == watcher);
0
+  
0
+  if(EV_ERROR & revents) {
0
+    g_message("on_request() got error event, closing server.");
0
+    ebb_server_unlisten(server);
0
+    return;
0
+  }
0
+  /* Now we're going to initialize the client 
0
+   * and set up her callbacks for read and write
0
+   * the client won't get passed back to the user, however,
0
+   * until the request is complete and parsed.
0
+   */
0
+  int i;
0
+  ebb_client *client;
0
+  /* Get next availible peer */
0
+  for(i=0; i < EBB_MAX_CLIENTS; i++)
0
+    if(!server->clients[i].open) {
0
+      client = &(server->clients[i]);
0
+      break;
0
+    }
0
+  if(client == NULL) {
0
+    g_message("Too many peers. Refusing connections.");
0
+    return;
0
+  }
0
+  
0
+#ifdef DEBUG
0
+  int count = 0;
0
+  for(i = 0; i < EBB_MAX_CLIENTS; i++)
0
+    if(server->clients[i].open) count += 1;
0
+  g_debug("%d open connections", count);
0
+#endif
0
+  
0
+  client_init(server, client);
0
+}
0
+
0
 
0
 ebb_server* ebb_server_alloc()
0
 {
0
@@ -430,7 +436,7 @@ void ebb_server_unlisten(ebb_server *server)
0
 }
0
 
0
 
0
-void ebb_server_listen(ebb_server *server)
0
+static void server_listen(ebb_server *server)
0
 {
0
   int r = listen(server->fd, EBB_MAX_CLIENTS);
0
   assert(r >= 0);
0
@@ -489,7 +495,7 @@ int ebb_server_listen_on_port(ebb_server *server, const int port)
0
   server->fd = sfd;
0
   server->port = malloc(sizeof(char)*8); /* for easy access to the port */
0
   sprintf(server->port, "%d", port);
0
-  ebb_server_listen(server);
0
+  server_listen(server);
0
   return server->fd;
0
 error:
0
   if(sfd > 0) close(sfd);
0
@@ -503,7 +509,7 @@ int ebb_server_listen_on_socket(ebb_server *server, const char *socketpath)
0
   if(fd < 0) return 0;
0
   server->socketpath = strdup(socketpath);
0
   server->fd = fd;
0
-  ebb_server_listen(server);
0
+  server_listen(server);
0
   return fd;
0
 }
0
 
0
@@ -533,13 +539,18 @@ void ebb_client_close(ebb_client *client)
0
 }
0
 
0
 
0
-void on_client_writable(struct ev_loop *loop, ev_io *watcher, int revents)
0
+static void on_client_writable(struct ev_loop *loop, ev_io *watcher, int revents)
0
 {
0
   ebb_client *client = (ebb_client*)(watcher->data);
0
   ssize_t sent;
0
   
0
+  assert(client->status_written);
0
+  assert(client->headers_written);
0
+  assert(client->began_transmission);
0
+  
0
   if(EV_ERROR & revents) {
0
     g_message("on_client_writable() got error event, closing peer");
0
+    ebb_client_close(client);
0
     return;
0
   }
0
   
0
@@ -565,25 +576,28 @@ void on_client_writable(struct ev_loop *loop, ev_io *watcher, int revents)
0
   
0
   ev_timer_again(loop, &(client->timeout_watcher));
0
   
0
-  if(client->written == client->response_buffer->len)
0
-    ebb_client_close(client);
0
+  if(client->written == client->response_buffer->len) {
0
+    ev_io_stop(loop, watcher);
0
+    if(client->body_written)
0
+      ebb_client_close(client);
0
+  }
0
 }
0
 
0
 void ebb_client_write_status(ebb_client *client, int status, const char *human_status)
0
 {
0
-  assert(client->status_sent == FALSE);
0
+  assert(client->status_written == FALSE);
0
   g_string_append_printf( client->response_buffer
0
                         , "HTTP/1.1 %d %s\r\n"
0
                         , status
0
                         , human_status
0
                         );
0
-  client->status_sent = TRUE;
0
+  client->status_written = TRUE;
0
 }
0
 
0
 void ebb_client_write_header(ebb_client *client, const char *field, const char *value)
0
 {
0
-  assert(client->status_sent == TRUE);
0
-  assert(client->headers_sent == FALSE);
0
+  assert(client->status_written == TRUE);
0
+  assert(client->headers_written == FALSE);
0
   g_string_append_printf( client->response_buffer
0
                         , "%s: %s\r\n"
0
                         , field
0
@@ -594,17 +608,19 @@ void ebb_client_write_header(ebb_client *client, const char *field, const char *
0
 void ebb_client_write(ebb_client *client, const char *data, int length)
0
 {
0
   g_string_append_len(client->response_buffer, data, length);
0
+  if(client->began_transmission) {
0
+    /* restart the watcher if we're streaming */
0
+    ev_io_start(client->server->loop, &client->write_watcher);
0
+  }
0
 }
0
 
0
 
0
-void ebb_client_finished(ebb_client *client)
0
+void ebb_client_begin_transmission(ebb_client *client)
0
 {
0
   assert(client->open);
0
-  assert(FALSE == ev_is_active(&(client->write_watcher)));
0
+  assert(FALSE == ev_is_active(&client->write_watcher));
0
   
0
-  /* assure the socket is still in non-blocking mode
0
-   * in the ruby binding, for example, i change this flag 
0
-   */
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()");
0
@@ -612,11 +628,13 @@ void ebb_client_finished(ebb_client *client)
0
     return;
0
   }
0
   
0
+  client->headers_written = TRUE;
0
+  client->began_transmission = TRUE;
0
   client->written = 0;
0
   client->write_watcher.data = client;
0
   ev_init (&(client->write_watcher), on_client_writable);
0
   ev_io_set (&(client->write_watcher), client->fd, EV_WRITE | EV_ERROR);
0
-  ev_io_start(client->server->loop, &(client->write_watcher));
0
+  ev_io_start(client->server->loop, &client->write_watcher);
0
 }
0
 
0
 

0
@@ -33,8 +33,7 @@ int ebb_client_read(ebb_client *client, char *buffer, int length);
0
 void ebb_client_write_status(ebb_client*, int status, const char *human_status);
0
 void ebb_client_write_header(ebb_client*, const char *field, const char *value);
0
 void ebb_client_write(ebb_client*, const char *data, int length);
0
-void ebb_client_finished( ebb_client *client);
0
-
0
+void ebb_client_begin_transmission( ebb_client *client);
0
 
0
 struct ebb_env_item {
0
   enum { EBB_FIELD_VALUE_PAIR
0
@@ -75,9 +74,10 @@ struct ebb_client {
0
   
0
   ev_timer timeout_watcher;
0
   
0
-  int status_sent;
0
-  int headers_sent;
0
-  int body_sent;
0
+  unsigned int status_written : 1;
0
+  unsigned int headers_written : 1;
0
+  unsigned int body_written : 1;
0
+  unsigned int began_transmission : 1;
0
   
0
   /* the ENV structure */
0
   int env_size;

0
@@ -186,14 +186,23 @@ VALUE client_write(VALUE _, VALUE client, VALUE string)
0
 }
0
 
0
 
0
-VALUE client_finished(VALUE _, VALUE client)
0
+VALUE client_begin_transmission(VALUE _, VALUE rb_client)
0
 {
0
-  ebb_client *_client;
0
-  Data_Get_Struct(client, ebb_client, _client);
0
-  ebb_client_finished(_client);
0
+  ebb_client *client;
0
+  Data_Get_Struct(rb_client, ebb_client, client);
0
+  client->status_written = TRUE;
0
+  client->headers_written = TRUE;
0
+  ebb_client_begin_transmission(client);
0
   return Qnil;
0
 }
0
 
0
+VALUE client_set_body_written(VALUE _, VALUE rb_client, VALUE v)
0
+{
0
+  ebb_client *client;
0
+  Data_Get_Struct(rb_client, ebb_client, client);
0
+  client->body_written = RTEST(v);
0
+  return client->body_written ? Qtrue : Qfalse;
0
+}
0
 
0
 void Init_ebb_ext()
0
 {
0
@@ -225,7 +234,8 @@ 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", client_write, 2);
0
-  rb_define_singleton_method(mFFI, "client_finished", client_finished, 1);
0
+  rb_define_singleton_method(mFFI, "client_begin_transmission", client_begin_transmission, 1);
0
+  rb_define_singleton_method(mFFI, "client_set_body_written", client_set_body_written, 2);
0
   rb_define_singleton_method(mFFI, "client_env", client_env, 1);
0
   
0
   /* initialize ebb_server */