Fix for pub/sub clients.

client.c

@@ -250,8 +250,14 @@ http_client_read(struct http_client *c) {
 		/* broken link, free buffer and client object */
 
 		/* disconnect pub/sub client if there is one. */
-		if(c->pub_sub) {
+		if(c->pub_sub && c->pub_sub->ac) {
+			/*
+			printf("client disconnected (cmd=%p), ac=%p\n",
+				(void*)c->pub_sub, (void*)c->pub_sub->ac);
+			*/
 			redisAsyncDisconnect(c->pub_sub->ac);
+			if(c->pub_sub) c->pub_sub->ac = NULL;
+			c->pub_sub = NULL;
 		}
 
 		close(c->fd);

cmd.c

@@ -177,6 +177,7 @@ cmd_run(struct worker *w, struct http_client *client,
 
 		/* register with the client, used upon disconnection */
 		client->pub_sub = cmd;
+		cmd->pub_sub_client = client;
 	} else {
 		/* get a connection from the pool */
 		cmd->ac = (redisAsyncContext*)pool_get_context(w->pool);
@@ -224,6 +225,7 @@ cmd_run(struct worker *w, struct http_client *client,
 	}
 	/* failed to find a suitable connection to Redis. */
 	cmd_free(cmd);
+	client->pub_sub = NULL;
 	return CMD_REDIS_UNAVAIL;
 }
 

cmd.h

@@ -38,6 +38,7 @@ struct cmd {
 	int is_websocket;
 	int http_version;
 
+	struct http_client *pub_sub_client;
 	redisAsyncContext *ac;
 };
 

formats/common.c

@@ -36,11 +36,17 @@ format_send_error(struct cmd *cmd, short code, const char *msg) {
 
 	struct http_response resp;
 
-	http_response_init(&resp, code, msg);
-	resp.http_version = cmd->http_version;
-	http_response_set_keep_alive(&resp, cmd->keep_alive);
-	http_response_write(&resp, cmd->fd);
+	if(!cmd->is_websocket && !cmd->pub_sub_client) {
+		http_response_init(&resp, code, msg);
+		resp.http_version = cmd->http_version;
+		http_response_set_keep_alive(&resp, cmd->keep_alive);
+		http_response_write(&resp, cmd->fd);
+	}
 
+	/* for pub/sub, remove command from client */
+	if(cmd->pub_sub_client) {
+		cmd->pub_sub_client->pub_sub = NULL;
+	}
 	cmd_free(cmd);
 }