Fix states logic broken by the previous commit.

5 files changed, 27 insertions, 22 deletions

diff --git a/src/libmime/filter.c b/src/libmime/filter.c
index a2c214d31..3a76232b9 100644
--- a/src/libmime/filter.c
+++ b/src/libmime/filter.c
@@ -431,7 +431,9 @@ rspamd_process_filters (struct rspamd_task *task)
 		}
 	}
 
-	task->state = WAIT_FILTER;
+	if (rspamd_session_events_pending (task->s) != 0) {
+		task->state = WAIT_FILTER;
+	}
 
 	return 1;
 }
diff --git a/src/libserver/events.c b/src/libserver/events.c
index 4925b8c69..a453cf459 100644
--- a/src/libserver/events.c
+++ b/src/libserver/events.c
@@ -284,3 +284,12 @@ rspamd_session_watch_stop (struct rspamd_async_session *s)
 
 	return remain;
 }
+
+
+guint
+rspamd_session_events_pending (struct rspamd_async_session *session)
+{
+	g_assert (session != NULL);
+
+	return g_hash_table_size (session->events);
+}
diff --git a/src/libserver/events.h b/src/libserver/events.h
index 46f7ae0ef..b2f463b5b 100644
--- a/src/libserver/events.h
+++ b/src/libserver/events.h
@@ -82,6 +82,13 @@ gboolean rspamd_session_destroy (struct rspamd_async_session *session);
 gboolean rspamd_session_pending (struct rspamd_async_session *session);
 
 /**
+ * Returns number of events pending
+ * @param session
+ * @return
+ */
+guint rspamd_session_events_pending (struct rspamd_async_session *session);
+
+/**
  * Start watching for events in the session, so the specified watcher will be added
  * to all subsequent events until `rspamd_session_watch_stop` is called
  * @param s session object
diff --git a/src/libserver/task.c b/src/libserver/task.c
index 502fdfb8d..4833c0835 100644
--- a/src/libserver/task.c
+++ b/src/libserver/task.c
@@ -355,11 +355,11 @@ rspamd_task_process (struct rspamd_task *task,
 	else {
 		rspamd_lua_call_pre_filters (task);
 		/* We want fin_task after pre filters are processed */
-		task->state = WAIT_PRE_FILTER;
+		if (rspamd_session_events_pending (task->s) != 0) {
+			task->state = WAIT_PRE_FILTER;
+		}
 	}
 
-	rspamd_session_pending (task->s);
-
 	return TRUE;
 }
 
diff --git a/src/worker.c b/src/worker.c
index e59054414..f142da22c 100644
--- a/src/worker.c
+++ b/src/worker.c
@@ -106,13 +106,13 @@ rspamd_worker_body_handler (struct rspamd_http_connection *conn,
 
 	ctx = task->worker->ctx;
 
+	task->state = WRITE_REPLY;
+
 	if (!rspamd_protocol_handle_request (task, msg)) {
-		task->state = WRITE_REPLY;
 		return 0;
 	}
 
 	if (task->cmd == CMD_PING) {
-		task->state = WRITE_REPLY;
 		return 0;
 	}
 
@@ -120,14 +120,11 @@ rspamd_worker_body_handler (struct rspamd_http_connection *conn,
 		msg_err ("got zero length body, cannot continue");
 		task->last_error = "message's body is empty";
 		task->error_code = RSPAMD_LENGTH_ERROR;
-		task->state = WRITE_REPLY;
+
 		return 0;
 	}
 
-
-	if (!rspamd_task_process (task, msg, chunk, len, TRUE)) {
-		task->state = WRITE_REPLY;
-	}
+	rspamd_task_process (task, msg, chunk, len, TRUE);
 
 	return 0;
 }
@@ -149,7 +146,7 @@ rspamd_worker_finish_handler (struct rspamd_http_connection *conn,
 {
 	struct rspamd_task *task = (struct rspamd_task *) conn->ud;
 
-	if (task->state == CLOSING_CONNECTION) {
+	if (task->state == CLOSING_CONNECTION || task->state == WRITING_REPLY) {
 		/* We are done here */
 		msg_debug ("normally closing connection from: %s",
 			rspamd_inet_address_to_string (task->client_addr));
@@ -163,16 +160,6 @@ rspamd_worker_finish_handler (struct rspamd_http_connection *conn,
 		 */
 		msg_debug ("want write message to the wire: %s",
 			rspamd_inet_address_to_string (task->client_addr));
-		rspamd_protocol_write_reply (task);
-		/* Forcefully set the state */
-		task->state = CLOSING_CONNECTION;
-	}
-	else if (task->state == WRITING_REPLY) {
-		msg_debug ("still writing reply to: %s",
-			rspamd_inet_address_to_string (task->client_addr));
-		task->state = CLOSING_CONNECTION;
-	}
-	else {
 		/*
 		 * If all filters have finished their tasks, this function will trigger
 		 * writing a reply.