Fix Keep-Alive.

cmd.c

@@ -270,8 +270,9 @@ cmd_select_format(struct cmd *cmd, const char *uri, size_t uri_len, formatting_f
 int
 cmd_is_subscribe(struct cmd *cmd) {
 
-	if(strncasecmp(cmd->argv[0], "SUBSCRIBE", cmd->argv_len[0]) == 0 ||
-		strncasecmp(cmd->argv[0], "PSUBSCRIBE", cmd->argv_len[0]) == 0) {
+	if(cmd->count >= 1 && 
+		(strncasecmp(cmd->argv[0], "SUBSCRIBE", cmd->argv_len[0]) == 0 ||
+		strncasecmp(cmd->argv[0], "PSUBSCRIBE", cmd->argv_len[0]) == 0)) {
 		return 1;
 	}
 	return 0;

formats/common.c

@@ -36,6 +36,7 @@ format_send_reply(struct cmd *cmd, const char *p, size_t sz, const char *content
 
 	if(cmd_is_subscribe(cmd)) {
 		free_cmd = 0;
+		printf("wtf\n");
 
 		/* start streaming */
 		if(cmd->started_responding == 0) {

formats/json.c

@@ -1,6 +1,7 @@
 #include "json.h"
 #include "common.h"
 #include "cmd.h"
+#include "http.h"
 
 #include <string.h>
 #include <hiredis/hiredis.h>

http.c

@@ -17,6 +17,8 @@ http_client_new(int fd, struct server *s) {
 	c->settings.on_path = http_on_path;
 	c->settings.on_body = http_on_body;
 	c->settings.on_message_complete = http_on_complete;
+	c->settings.on_header_field = http_on_header_name;
+	c->settings.on_header_value = http_on_header_value;
 
 	http_parser_init(&c->parser, HTTP_REQUEST);
 	c->parser.data = c;
@@ -37,7 +39,6 @@ http_client_read(int fd, short event, void *ctx) {
 
 	ret = read(c->fd, buffer, sizeof(buffer));
 	if(ret <= 0) { /* broken connection, bye */
-		printf("close client %d\n", c->fd);
 		http_client_free(c);
 		return;
 	}
@@ -52,27 +53,65 @@ http_client_read(int fd, short event, void *ctx) {
 		/* TODO */
 	} else if(nparsed != ret) { /* invalid */
 		http_client_free(c);
+		return;
 	}
-	/*
-	printf("parse %zd bytes: [", c->sz); fflush(stdout);
-	write(1, c->buffer, c->sz);
-	printf("]\n");
-	*/
-
 
-	/* carry on */
 	http_client_serve(c);
 }
 
 void
 http_client_free(struct http_client *c) {
 
+	event_del(&c->ev);
 	close(c->fd);
 	free(c->buffer);
 	free((char*)c->out_content_type.s);
 	free(c);
 }
 
+static int
+http_client_keep_alive(struct http_client *c) {
+
+	/* check disconnection */
+	int disconnect = 0;
+
+	if(c->parser.http_major == 1 && c->parser.http_minor == 0) {
+		disconnect = 1; /* HTTP 1.0: disconnect by default */
+	}
+	if(c->header_connection.s) {
+		if(strncasecmp(c->header_connection.s, "Keep-Alive", 10) == 0) {
+			disconnect = 0;
+		} else if(strncasecmp(c->header_connection.s, "Close", 5) == 0) {
+			disconnect = 1;
+		}
+	}
+	return disconnect ? 0 : 1;
+}
+
+void
+http_client_reset(struct http_client *c) {
+
+	if(!http_client_keep_alive(c)) {
+		http_client_free(c);
+		return;
+	}
+
+	memset(&c->path, 0, sizeof(c->path));
+	memset(&c->body, 0, sizeof(c->body));
+
+	free((char*)c->out_content_type.s);
+	memset(&c->out_content_type, 0, sizeof(c->out_content_type));
+	memset(&c->out_etag, 0, sizeof(c->out_etag));
+
+	free(c->buffer);
+	c->buffer = NULL;
+	c->sz = 0;
+
+	memset(&c->verb, 0, sizeof(c->verb));
+
+	http_parser_init(&c->parser, HTTP_REQUEST);
+}
+
 /**
  * Add read event callback
  */
@@ -81,7 +120,6 @@ http_client_serve(struct http_client *c) {
 
 	event_set(&c->ev, c->fd, EV_READ, http_client_read, c);
 	event_base_set(c->s->base, &c->ev);
-
 	event_add(&c->ev, NULL);
 }
 
@@ -112,7 +150,7 @@ http_on_body(http_parser *p, const char *at, size_t length) {
 int
 http_on_complete(http_parser *p) {
 	struct http_client *c = p->data;
-	int ret;
+	int ret = -1;
 
 	/* check that the command can be executed */
 	switch(c->verb) {
@@ -141,7 +179,7 @@ http_on_complete(http_parser *p) {
 			/* evhttp_send_reply(rq, 405, "Method Not Allowed", NULL); */
 			return 0;
 	}
-	return 0;
+	return ret;
 }
 
 void
@@ -161,23 +199,28 @@ http_send_reply(struct http_client *c, short code, const char *msg,
 		ret = snprintf(out, sz, "HTTP/1.1 %d %s\r\n"
 			"Content-Type: %s\r\n"
 			"Content-Length: %zd\r\n"
+			"Connection: %s\r\n"
 			"Server: Webdis\r\n"
-			"\r\n", code, msg, ct, body_len);
+			"\r\n", code, msg, ct, body_len,
+			(http_client_keep_alive(c) ? "Keep-Alive" : "Close")
+			);
 
-		if(!out) {
+		if(!out) { /* first step: allocate space */
 			sz = ret + body_len;
 			out = malloc(sz);
-		} else {
+		} else { /* second step: copy body and leave loop */
 			if(body && body_len) memcpy(out + ret, body, body_len);
 			break;
 		}
 	}
 
 	ok = write(c->fd, out, sz);
-	if(!ok) {
+	if(ok != (int)sz) {
 		http_client_free(c);
 	}
 	free(out);
+
+	http_client_reset(c);
 }
 
 void
@@ -186,3 +229,32 @@ http_set_header(str_t *h, const char *p) {
 	h->s = strdup(p);
 	h->sz = strlen(p);
 }
+
+/**
+ * Called when a header name is read
+ */
+int
+http_on_header_name(http_parser *p, const char *at, size_t length) {
+
+	struct http_client *c = p->data;
+
+	c->last_header_name.s = at;
+	c->last_header_name.sz = length;
+
+	return 0;
+}
+
+/**
+ * Called when a header value is read
+ */
+int
+http_on_header_value(http_parser *p, const char *at, size_t length) {
+
+	struct http_client *c = p->data;
+
+	if(strncmp("Connection", c->last_header_name.s, length) == 0) {
+		c->header_connection.s = at;
+		c->header_connection.sz = length;
+	}
+	return 0;
+}

http.h

@@ -15,6 +15,7 @@ struct http_client {
 	int fd;
 	struct event ev;
 	struct server *s;
+	int needs_free;
 
 	/* input buffer */
 	char *buffer;
@@ -29,9 +30,13 @@ struct http_client {
 
 	str_t path;
 	str_t body;
+	str_t header_connection;
 
 	str_t out_content_type;
 	str_t out_etag;
+
+	/* private, used in HTTP parser */
+	str_t last_header_name;
 };
 
 struct http_client *
@@ -52,10 +57,20 @@ http_on_path(http_parser*, const char *at, size_t length);
 int
 http_on_body(http_parser*, const char *at, size_t length);
 
+int
+http_on_header_name(http_parser*, const char *at, size_t length);
+
+int
+http_on_header_value(http_parser*, const char *at, size_t length);
+
 int
 http_on_complete(http_parser*);
 
 void
 http_set_header(str_t *h, const char *p);
 
+void
+http_send_reply(struct http_client *c, short code, const char *msg,
+		const char *body, size_t body_len);
+
 #endif

server.c

@@ -193,59 +193,6 @@ on_options(struct evhttp_request *rq) {
 }
 #endif
 
-void
-on_request(struct evhttp_request *rq, void *ctx) {
-	const char *uri = evhttp_request_uri(rq);
-	struct server *s = ctx;
-	int ret;
-
-	if(!s->ac) { /* redis is unavailable */
-		evhttp_send_reply(rq, 503, "Service Unavailable", NULL);
-		return;
-	}
-
-
-	/* check that the command can be executed */
-	switch(rq->type) {
-		case EVHTTP_REQ_GET:
-			slog(s, WEBDIS_DEBUG, uri);
-			ret = cmd_run(s, rq, 1+uri, strlen(uri)-1, NULL, 0);
-			break;
-
-		case EVHTTP_REQ_POST:
-			slog(s, WEBDIS_DEBUG, uri);
-			ret = cmd_run(s, rq,
-				(const char*)EVBUFFER_DATA(rq->input_buffer),
-				EVBUFFER_LENGTH(rq->input_buffer), NULL, 0);
-			break;
-
-#ifdef _EVENT2_HTTP_H_
-		case EVHTTP_REQ_PUT:
-			slog(s, WEBDIS_DEBUG, uri);
-			ret = cmd_run(s, rq, 1+uri, strlen(uri)-1,
-				(const char*)EVBUFFER_DATA(rq->input_buffer),
-				EVBUFFER_LENGTH(rq->input_buffer));
-			break;
-#endif
-
-#ifdef _EVENT2_HTTP_H_
-		case EVHTTP_REQ_OPTIONS:
-			on_options(rq);
-			return;
-#endif
-
-		default:
-			slog(s, WEBDIS_DEBUG, "405");
-			evhttp_send_reply(rq, 405, "Method Not Allowed", NULL);
-			return;
-	}
-
-	if(ret < 0) {
-		evhttp_send_reply(rq, 403, "Forbidden", NULL);
-	}
-}
-
-
 static void
 on_possible_accept(int fd, short event, void *ctx) {