restapi: refactor client

Signed-off-by: He Xian <[email protected]>
hexian000 · Jan 14, 2025 · 7bd5b27 · 7bd5b27
1 parent 16cf51e
commit 7bd5b27
Show file tree

Hide file tree

Showing 10 changed files with 59 additions and 52 deletions.
diff --git a/src/api_client.c b/src/api_client.c
@@ -40,6 +40,7 @@ struct api_client_ctx {
 	struct session ss;
 	enum api_client_state state;
 	struct api_client_cb cb;
+	struct ev_idle w_start;
 	struct ev_timer w_timeout;
 	struct ev_idle w_ruleset;
 	struct dialreq *dialreq;
@@ -57,6 +58,7 @@ ASSERT_SUPER(struct session, struct api_client_ctx, ss);
 static void api_client_close(
 	struct ev_loop *restrict loop, struct api_client_ctx *restrict ctx)
 {
+	ev_idle_stop(loop, &ctx->w_start);
 	ev_timer_stop(loop, &ctx->w_timeout);
 	ev_idle_stop(loop, &ctx->w_ruleset);
 	if (ctx->state == STATE_CLIENT_CONNECT) {
@@ -320,9 +322,18 @@ static bool parse_header(void *ctx, const char *key, char *value)
 	return true;
 }
 
-static struct api_client_ctx *api_client_do(
-	struct ev_loop *loop, struct dialreq *req, const char *uri,
-	const char *payload, const size_t len,
+static void start_cb(struct ev_loop *loop, struct ev_idle *watcher, int revents)
+{
+	CHECK_REVENTS(revents, EV_IDLE);
+	ev_idle_stop(loop, watcher);
+	struct api_client_ctx *restrict ctx = watcher->data;
+	ev_timer_start(loop, &ctx->w_timeout);
+	dialer_do(&ctx->dialer, loop, ctx->dialreq);
+}
+
+static bool api_client_do(
+	struct ev_loop *loop, struct api_client_ctx **pctx, struct dialreq *req,
+	const char *uri, const char *payload, const size_t len,
 	const struct api_client_cb *in_cb)
 {
 	CHECK(len <= INT_MAX);
@@ -331,7 +342,7 @@ static struct api_client_ctx *api_client_do(
 	if (ctx == NULL) {
 		LOGOOM();
 		dialreq_free(req);
-		return NULL;
+		return false;
 	}
 	ctx->state = STATE_CLIENT_CONNECT;
 	ctx->dialreq = req;
@@ -340,9 +351,11 @@ static struct api_client_ctx *api_client_do(
 	if (!make_request(&ctx->parser, uri, payload, len)) {
 		LOGOOM();
 		api_client_close(loop, ctx);
-		return NULL;
+		return false;
 	}
 	ctx->cb = *in_cb;
+	ev_idle_init(&ctx->w_start, start_cb);
+	ctx->w_start.data = ctx;
 	ev_timer_init(&ctx->w_timeout, timeout_cb, G.conf->timeout, 0.0);
 	ctx->w_timeout.data = ctx;
 	ev_idle_init(&ctx->w_ruleset, idle_cb);
@@ -354,7 +367,7 @@ static struct api_client_ctx *api_client_do(
 		.func = dialer_cb,
 		.data = ctx,
 	};
-	dialer_init(&ctx->dialer, cb);
+	dialer_init(&ctx->dialer, &cb);
 	if (ctx->cb.func != NULL) {
 		/* managed by ruleset */
 		ctx->ss.close = NULL;
@@ -364,29 +377,29 @@ static struct api_client_ctx *api_client_do(
 		session_add(&ctx->ss);
 	}
 
-	ev_timer_start(loop, &ctx->w_timeout);
-	dialer_start(&ctx->dialer, loop, req);
-	if (in_cb->func == NULL) {
-		return NULL;
+	ev_idle_start(loop, &ctx->w_start);
+	if (pctx != NULL) {
+		*pctx = ctx;
 	}
-	return ctx;
+	return true;
 }
 
 void api_client_invoke(
 	struct ev_loop *loop, struct dialreq *req, const char *payload,
 	const size_t len)
 {
 	(void)api_client_do(
-		loop, req, "/ruleset/invoke", payload, len,
+		loop, NULL, req, "/ruleset/invoke", payload, len,
 		&(struct api_client_cb){ NULL, NULL });
 }
 
-struct api_client_ctx *api_client_rpcall(
-	struct ev_loop *loop, struct dialreq *req, const char *payload,
-	const size_t len, const struct api_client_cb *cb)
+bool api_client_rpcall(
+	struct ev_loop *loop, struct api_client_ctx **pctx, struct dialreq *req,
+	const char *payload, const size_t len, const struct api_client_cb *cb)
 {
 	ASSERT(cb->func != NULL);
-	return api_client_do(loop, req, "/ruleset/rpcall", payload, len, cb);
+	return api_client_do(
+		loop, pctx, req, "/ruleset/rpcall", payload, len, cb);
 }
 
 void api_client_cancel(struct ev_loop *loop, struct api_client_ctx *ctx)

diff --git a/src/api_client.h b/src/api_client.h
@@ -4,6 +4,7 @@
 #ifndef API_CLIENT_H
 #define API_CLIENT_H
 
+#include <stdbool.h>
 #include <stddef.h>
 
 struct ev_loop;
@@ -22,9 +23,9 @@ void api_client_invoke(
 	struct ev_loop *loop, struct dialreq *req, const char *payload,
 	size_t len);
 
-struct api_client_ctx *api_client_rpcall(
-	struct ev_loop *loop, struct dialreq *req, const char *payload,
-	size_t len, const struct api_client_cb *cb);
+bool api_client_rpcall(
+	struct ev_loop *loop, struct api_client_ctx **pctx, struct dialreq *req,
+	const char *payload, size_t len, const struct api_client_cb *cb);
 
 void api_client_cancel(struct ev_loop *loop, struct api_client_ctx *ctx);
 

diff --git a/src/dialer.c b/src/dialer.c
@@ -372,7 +372,6 @@ dialer_stop(struct dialer *restrict d, struct ev_loop *loop, const bool ok)
 {
 	switch (d->state) {
 	case STATE_INIT:
-		ev_clear_pending(loop, &d->w_start);
 		break;
 	case STATE_RESOLVE:
 		if (d->resolve_query != NULL) {
@@ -1253,11 +1252,8 @@ static void resolve_cb(
 	}
 }
 
-static void
-start_cb(struct ev_loop *loop, struct ev_watcher *watcher, int revents)
+static void dialer_start(struct dialer *restrict d, struct ev_loop *loop)
 {
-	CHECK_REVENTS(revents, EV_CUSTOM);
-	struct dialer *restrict d = watcher->data;
 	const struct dialreq *restrict req = d->req;
 	const struct dialaddr *restrict addr =
 		req->num_proxy > 0 ? &req->proxy[0].addr : &req->addr;
@@ -1304,18 +1300,15 @@ start_cb(struct ev_loop *loop, struct ev_watcher *watcher, int revents)
 	}
 }
 
-void dialer_init(struct dialer *restrict d, const struct event_cb cb)
+void dialer_init(struct dialer *restrict d, const struct event_cb *cb)
 {
-	d->done_cb = cb;
+	d->done_cb = *cb;
 	d->req = NULL;
 	d->resolve_query = NULL;
 	d->jump = 0;
 	d->state = STATE_INIT;
 	d->syserr = 0;
 	{
-		struct ev_watcher *restrict w_start = &d->w_start;
-		ev_init(w_start, start_cb);
-		w_start->data = d;
 		struct ev_io *restrict w_socket = &d->w_socket;
 		ev_io_init(w_socket, socket_cb, -1, EV_NONE);
 		w_socket->data = d;
@@ -1324,7 +1317,7 @@ void dialer_init(struct dialer *restrict d, const struct event_cb cb)
 	BUF_INIT(d->rbuf, 0);
 }
 
-void dialer_start(
+void dialer_do(
 	struct dialer *restrict d, struct ev_loop *restrict loop,
 	const struct dialreq *restrict req)
 {
@@ -1336,7 +1329,7 @@ void dialer_start(
 	}
 	d->req = req;
 	d->syserr = 0;
-	ev_feed_event(loop, &d->w_start, EV_CUSTOM);
+	dialer_start(d, loop);
 }
 
 int dialer_get(struct dialer *d)

diff --git a/src/dialer.h b/src/dialer.h
@@ -78,7 +78,6 @@ struct dialer {
 	size_t jump;
 	int state;
 	int syserr;
-	struct ev_watcher w_start;
 	struct ev_io w_socket;
 	unsigned char *next;
 	struct {
@@ -87,9 +86,9 @@ struct dialer {
 	} rbuf;
 };
 
-void dialer_init(struct dialer *d, struct event_cb cb);
+void dialer_init(struct dialer *d, const struct event_cb *cb);
 
-void dialer_start(
+void dialer_do(
 	struct dialer *d, struct ev_loop *loop, const struct dialreq *req);
 
 void dialer_cancel(struct dialer *d, struct ev_loop *loop);

diff --git a/src/forward.c b/src/forward.c
@@ -226,10 +226,10 @@ static void dialer_cb(struct ev_loop *loop, void *data)
 	};
 	struct server_stats *restrict stats = &ctx->s->stats;
 	transfer_init(
-		&ctx->uplink, cb, ctx->accepted_fd, ctx->dialed_fd,
+		&ctx->uplink, &cb, ctx->accepted_fd, ctx->dialed_fd,
 		&stats->byt_up);
 	transfer_init(
-		&ctx->downlink, cb, ctx->dialed_fd, ctx->accepted_fd,
+		&ctx->downlink, &cb, ctx->dialed_fd, ctx->accepted_fd,
 		&stats->byt_down);
 	transfer_start(loop, &ctx->uplink);
 	transfer_start(loop, &ctx->downlink);
@@ -239,14 +239,14 @@ static void forward_ctx_start(
 	struct ev_loop *loop, struct forward_ctx *restrict ctx,
 	const struct dialreq *req)
 {
-	dialer_start(&ctx->dialer, loop, req);
 	ev_timer_start(loop, &ctx->w_timeout);
 
 	FW_CTX_LOG(VERBOSE, ctx, "connect");
 	ctx->state = STATE_CONNECT;
 	struct server_stats *restrict stats = &ctx->s->stats;
 	stats->num_request++;
 	stats->num_halfopen++;
+	dialer_do(&ctx->dialer, loop, req);
 }
 
 #if WITH_RULESET
@@ -341,7 +341,7 @@ forward_ctx_new(struct server *restrict s, const int accepted_fd)
 		.data = ctx,
 	};
 	ctx->dialreq = NULL;
-	dialer_init(&ctx->dialer, cb);
+	dialer_init(&ctx->dialer, &cb);
 	ctx->ss.close = forward_ss_close;
 	session_add(&ctx->ss);
 	return ctx;

diff --git a/src/http_proxy.c b/src/http_proxy.c
@@ -244,7 +244,7 @@ static void http_connect(struct ev_loop *loop, struct http_ctx *restrict ctx)
 	}
 	HTTP_CTX_LOG(VERBOSE, ctx, "connect");
 	ctx->state = STATE_CONNECT;
-	dialer_start(&ctx->dialer, loop, ctx->dialreq);
+	dialer_do(&ctx->dialer, loop, ctx->dialreq);
 }
 
 #if WITH_RULESET
@@ -379,10 +379,10 @@ static void http_ctx_hijack(struct ev_loop *loop, struct http_ctx *restrict ctx)
 	};
 	struct server_stats *restrict stats = &ctx->s->stats;
 	transfer_init(
-		&ctx->uplink, cb, ctx->accepted_fd, ctx->dialed_fd,
+		&ctx->uplink, &cb, ctx->accepted_fd, ctx->dialed_fd,
 		&stats->byt_up);
 	transfer_init(
-		&ctx->downlink, cb, ctx->dialed_fd, ctx->accepted_fd,
+		&ctx->downlink, &cb, ctx->dialed_fd, ctx->accepted_fd,
 		&stats->byt_down);
 	transfer_start(loop, &ctx->uplink);
 	transfer_start(loop, &ctx->downlink);
@@ -553,7 +553,7 @@ static struct http_ctx *http_ctx_new(struct server *restrict s, const int fd)
 		.func = dialer_cb,
 		.data = ctx,
 	};
-	dialer_init(&ctx->dialer, cb);
+	dialer_init(&ctx->dialer, &cb);
 
 	ctx->ss.close = http_ss_close;
 	session_add(&ctx->ss);

diff --git a/src/ruleset/await.c b/src/ruleset/await.c
@@ -338,8 +338,9 @@ static int await_invoke(lua_State *restrict L)
 		.func = invoke_cb,
 		.data = ud,
 	};
-	ud->ctx = api_client_rpcall(r->loop, req, code, len, &cb);
-	if (ud->ctx == NULL) {
+	const bool ok =
+		api_client_rpcall(r->loop, &ud->ctx, req, code, len, &cb);
+	if (!ok) {
 		lua_pushliteral(L, ERR_MEMORY);
 		return lua_error(L);
 	}

diff --git a/src/socks.c b/src/socks.c
@@ -415,10 +415,10 @@ static void dialer_cb(struct ev_loop *loop, void *data)
 	};
 	struct server_stats *restrict stats = &ctx->s->stats;
 	transfer_init(
-		&ctx->uplink, cb, ctx->accepted_fd, ctx->dialed_fd,
+		&ctx->uplink, &cb, ctx->accepted_fd, ctx->dialed_fd,
 		&stats->byt_up);
 	transfer_init(
-		&ctx->downlink, cb, ctx->dialed_fd, ctx->accepted_fd,
+		&ctx->downlink, &cb, ctx->dialed_fd, ctx->accepted_fd,
 		&stats->byt_down);
 	transfer_start(loop, &ctx->uplink);
 	transfer_start(loop, &ctx->downlink);
@@ -784,7 +784,7 @@ static void socks_connect(struct ev_loop *loop, struct socks_ctx *restrict ctx)
 
 	SOCKS_CTX_LOG(VERBOSE, ctx, "connect");
 	ctx->state = STATE_CONNECT;
-	dialer_start(&ctx->dialer, loop, ctx->dialreq);
+	dialer_do(&ctx->dialer, loop, ctx->dialreq);
 }
 
 #if WITH_RULESET
@@ -932,7 +932,7 @@ socks_ctx_new(struct server *restrict s, const int accepted_fd)
 		.func = dialer_cb,
 		.data = ctx,
 	};
-	dialer_init(&ctx->dialer, cb);
+	dialer_init(&ctx->dialer, &cb);
 	ctx->ss.close = socks_ss_close;
 	session_add(&ctx->ss);
 	return ctx;

diff --git a/src/transfer.c b/src/transfer.c
@@ -296,16 +296,16 @@ static void pipe_cb(struct ev_loop *loop, struct ev_io *watcher, int revents)
 #endif
 
 void transfer_init(
-	struct transfer *restrict t, const struct event_cb cb, const int src_fd,
-	const int dst_fd, uintmax_t *byt_transferred)
+	struct transfer *restrict t, const struct event_cb *cb,
+	const int src_fd, const int dst_fd, uintmax_t *byt_transferred)
 {
 	t->state = XFER_INIT;
 	t->src_fd = src_fd;
 	t->dst_fd = dst_fd;
 	struct ev_io *restrict w_socket = &t->w_socket;
 	ev_io_init(w_socket, transfer_cb, src_fd, EV_READ);
 	w_socket->data = t;
-	t->state_cb = cb;
+	t->state_cb = *cb;
 	t->byt_transferred = byt_transferred;
 
 #if WITH_SPLICE

diff --git a/src/transfer.h b/src/transfer.h
@@ -38,7 +38,7 @@ struct transfer {
 };
 
 void transfer_init(
-	struct transfer *t, struct event_cb cb, int src_fd, int dst_fd,
+	struct transfer *t, const struct event_cb *cb, int src_fd, int dst_fd,
 	uintmax_t *byt_transferred);
 
 void transfer_start(struct ev_loop *loop, struct transfer *t);