[Project] Adopt lua http

src/lua/lua_http.c

@@ -60,10 +60,11 @@ static const struct luaL_reg httplib_m[] = {
 struct lua_http_cbdata {
 	struct rspamd_http_connection *conn;
 	struct rspamd_async_session *session;
-	struct rspamd_async_watcher *w;
+	struct rspamd_symcache_item *item;
 	struct rspamd_http_message *msg;
 	struct event_base *ev_base;
 	struct rspamd_config *cfg;
+	struct rspamd_task *task;
 	struct timeval tv;
 	struct rspamd_cryptobox_keypair *local_kp;
 	struct rspamd_cryptobox_pubkey *peer_pk;
@@ -146,15 +147,15 @@ static void
 lua_http_maybe_free (struct lua_http_cbdata *cbd)
 {
 	if (cbd->session) {
-		if (cbd->w) {
-			/* We still need to clear watcher */
-			rspamd_session_watcher_pop (cbd->session, cbd->w);
-		}
 
 		if (cbd->flags & RSPAMD_LUA_HTTP_FLAG_RESOLVED) {
 			/* Event is added merely for resolved events */
 			rspamd_session_remove_event (cbd->session, lua_http_fin, cbd);
 		}
+
+		if (cbd->item) {
+			rspamd_symcache_item_async_dec_check (cbd->task, cbd->item);
+		}
 	}
 	else {
 		lua_http_fin (cbd);
@@ -204,7 +205,6 @@ lua_http_finish_handler (struct rspamd_http_connection *conn,
 {
 	struct lua_http_cbdata *cbd = (struct lua_http_cbdata *)conn->ud;
 	struct rspamd_http_header *h, *htmp;
-	struct rspamd_async_watcher *existing_watcher = NULL;
 	const gchar *body;
 	gsize body_len;
 
@@ -258,9 +258,9 @@ lua_http_finish_handler (struct rspamd_http_connection *conn,
 		lua_settable (L, -3);
 	}
 
-	if (cbd->w) {
+	if (cbd->item) {
 		/* Replace watcher to deal with nested calls */
-		existing_watcher = rspamd_session_replace_watcher (cbd->session, cbd->w);
+		rspamd_symbols_cache_set_cur_item (cbd->task, cbd->item);
 	}
 
 	if (lua_pcall (L, 4, 0, 0) != 0) {
@@ -268,11 +268,6 @@ lua_http_finish_handler (struct rspamd_http_connection *conn,
 		lua_pop (L, 1);
 	}
 
-	if (cbd->w) {
-		/* Restore existing watcher */
-		rspamd_session_replace_watcher (cbd->session, existing_watcher);
-	}
-
 	lua_http_maybe_free (cbd);
 
 	lua_thread_pool_restore_callback (&lcbd);
@@ -292,7 +287,6 @@ lua_http_resume_handler (struct rspamd_http_connection *conn,
 	const gchar *body;
 	gsize body_len;
 	struct rspamd_http_header *h, *htmp;
-	struct rspamd_async_watcher *existing_watcher = NULL;
 
 	if (err) {
 		lua_pushstring (L, err);
@@ -355,17 +349,12 @@ lua_http_resume_handler (struct rspamd_http_connection *conn,
 		lua_settable (L, -3);
 	}
 
-	if (cbd->w) {
+	if (cbd->item) {
 		/* Replace watcher to deal with nested calls */
-		existing_watcher = rspamd_session_replace_watcher (cbd->session, cbd->w);
+		rspamd_symbols_cache_set_cur_item (cbd->task, cbd->item);
 	}
 
 	lua_thread_resume (cbd->thread, 2);
-
-	if (cbd->w) {
-		/* Restore existing watcher */
-		rspamd_session_replace_watcher (cbd->session, existing_watcher);
-	}
 }
 
 static gboolean
@@ -431,12 +420,16 @@ lua_http_make_connection (struct lua_http_cbdata *cbd)
 		cbd->msg = NULL;
 
 		if (cbd->session) {
-			rspamd_session_add_event (cbd->session, cbd->w,
+			rspamd_session_add_event (cbd->session,
 					(event_finalizer_t) lua_http_fin, cbd,
 					g_quark_from_static_string ("lua http"));
 			cbd->flags |= RSPAMD_LUA_HTTP_FLAG_RESOLVED;
 		}
 
+		if (cbd->item) {
+			rspamd_symcache_item_async_inc (cbd->task, cbd->item);
+		}
+
 		return TRUE;
 	}
 
@@ -465,8 +458,14 @@ lua_http_dns_handler (struct rdns_reply *reply, gpointer ud)
 		if (!lua_http_make_connection (cbd)) {
 			lua_http_push_error (cbd, "unable to make connection to the host");
 			lua_http_maybe_free (cbd);
+
+			return;
 		}
 	}
+
+	if (cbd->item) {
+		rspamd_symcache_item_async_dec_check (cbd->task, cbd->item);
+	}
 }
 
 static void
@@ -863,10 +862,13 @@ lua_http_request (lua_State *L)
 		return 1;
 	}
 	if (task == NULL && cfg == NULL) {
-		return luaL_error (L, "Bad params to rspamd_http:request(): either task or config should be set");
+		return luaL_error (L,
+				"Bad params to rspamd_http:request(): either task or config should be set");
 	}
+
 	if (ev_base == NULL) {
-		return luaL_error (L, "Bad params to rspamd_http:request(): ev_base isn't passed");
+		return luaL_error (L,
+				"Bad params to rspamd_http:request(): ev_base isn't passed");
 	}
 
 	cbd = g_malloc0 (sizeof (*cbd));
@@ -883,6 +885,11 @@ lua_http_request (lua_State *L)
 	cbd->max_size = max_size;
 	cbd->url = url;
 	cbd->auth = auth;
+	cbd->task = task;
+
+	if (task) {
+		cbd->item = rspamd_symbols_cache_get_cur_item (task);
+	}
 
 	if (msg->host) {
 		cbd->host = rspamd_fstring_cstr (msg->host);
@@ -900,9 +907,6 @@ lua_http_request (lua_State *L)
 
 	if (session) {
 		cbd->session = session;
-
-		cbd->w = rspamd_session_get_watcher (cbd->session);
-		rspamd_session_watcher_push_specific (cbd->session, cbd->w);
 	}
 
 	if (rspamd_parse_inet_address (&cbd->addr, msg->host->str, msg->host->len)) {
@@ -911,10 +915,6 @@ lua_http_request (lua_State *L)
 			lua_http_maybe_free (cbd);
 			lua_pushboolean (L, FALSE);
 
-			if (cbd->w) {
-				rspamd_session_watcher_pop (cbd->session, cbd->w);
-			}
-
 			return 1;
 		}
 	}
@@ -930,10 +930,6 @@ lua_http_request (lua_State *L)
 				lua_pushboolean (L, FALSE);
 				g_free (to_resolve);
 
-				if (cbd->w) {
-					rspamd_session_watcher_pop (cbd->session, cbd->w);
-				}
-
 				return 1;
 			}
 
@@ -948,12 +944,11 @@ lua_http_request (lua_State *L)
 				lua_http_maybe_free (cbd);
 				lua_pushboolean (L, FALSE);
 
-				if (cbd->w) {
-					rspamd_session_watcher_pop (cbd->session, cbd->w);
-				}
-
 				return 1;
 			}
+			else {
+				rspamd_symcache_item_async_inc (cbd->task, cbd->item);
+			}
 		}
 	}