websocket: use nni_aio_start

2 files changed, 18 insertions, 47 deletions

diff --git a/src/sp/transport/ws/websocket.c b/src/sp/transport/ws/websocket.c
index ab139a48..ec71af0a 100644
--- a/src/sp/transport/ws/websocket.c
+++ b/src/sp/transport/ws/websocket.c
@@ -127,11 +127,9 @@ wstran_pipe_recv(void *arg, nni_aio *aio)
 {
 	ws_pipe *p = arg;
 
-	if (nni_aio_begin(aio) != 0) {
-		return;
-	}
+	nni_aio_reset(aio);
 	nni_mtx_lock(&p->mtx);
-	if (!nni_aio_defer(aio, wstran_pipe_recv_cancel, p)) {
+	if (!nni_aio_start(aio, wstran_pipe_recv_cancel, p)) {
 		nni_mtx_unlock(&p->mtx);
 		return;
 	}
@@ -160,15 +158,9 @@ wstran_pipe_send(void *arg, nni_aio *aio)
 {
 	ws_pipe *p = arg;
 
-	if (nni_aio_begin(aio) != 0) {
-		// No way to give the message back to the protocol, so
-		// we just discard it silently to prevent it from leaking.
-		nni_msg_free(nni_aio_get_msg(aio));
-		nni_aio_set_msg(aio, NULL);
-		return;
-	}
+	nni_aio_reset(aio);
 	nni_mtx_lock(&p->mtx);
-	if (!nni_aio_defer(aio, wstran_pipe_send_cancel, p)) {
+	if (!nni_aio_start(aio, wstran_pipe_send_cancel, p)) {
 		nni_mtx_unlock(&p->mtx);
 		return;
 	}
@@ -271,11 +263,9 @@ wstran_listener_accept(void *arg, nni_aio *aio)
 	// We already bound, so we just need to look for an available
 	// pipe (created by the handler), and match it.
 	// Otherwise we stick the AIO in the accept list.
-	if (nni_aio_begin(aio) != 0) {
-		return;
-	}
+	nni_aio_reset(aio);
 	nni_mtx_lock(&l->mtx);
-	if (!nni_aio_defer(aio, wstran_listener_cancel, l)) {
+	if (!nni_aio_start(aio, wstran_listener_cancel, l)) {
 		nni_mtx_unlock(&l->mtx);
 		return;
 	}
@@ -306,12 +296,10 @@ wstran_dialer_connect(void *arg, nni_aio *aio)
 {
 	ws_dialer *d = arg;
 
-	if (nni_aio_begin(aio) != 0) {
-		return;
-	}
+	nni_aio_reset(aio);
 
 	nni_mtx_lock(&d->mtx);
-	if (!nni_aio_defer(aio, wstran_dialer_cancel, d)) {
+	if (!nni_aio_start(aio, wstran_dialer_cancel, d)) {
 		nni_mtx_unlock(&d->mtx);
 		return;
 	}
diff --git a/src/supplemental/websocket/websocket.c b/src/supplemental/websocket/websocket.c
index b7afc938..6053c5c2 100644
--- a/src/supplemental/websocket/websocket.c
+++ b/src/supplemental/websocket/websocket.c
@@ -746,9 +746,7 @@ ws_send_close(nni_ws *ws, uint16_t code)
 	}
 	ws->closed = true;
 
-	if (nni_aio_begin(aio) != 0) {
-		return;
-	}
+	nni_aio_reset(aio);
 	ws->wclose = true;
 	rv = ws_msg_init_control(&frame, ws, WS_CLOSE, buf, sizeof(buf));
 	if (rv != 0) {
@@ -756,9 +754,8 @@ ws_send_close(nni_ws *ws, uint16_t code)
 		nni_aio_finish_error(aio, rv);
 		return;
 	}
-	if ((rv = nni_aio_schedule(aio, ws_cancel_close, ws)) != 0) {
+	if (!nni_aio_start(aio, ws_cancel_close, ws)) {
 		ws->wclose = false;
-		nni_aio_finish_error(aio, rv);
 		ws_frame_fini(frame);
 		return;
 	}
@@ -1719,11 +1716,8 @@ ws_listener_accept(void *arg, nni_aio *aio)
 {
 	nni_ws_listener *l = arg;
 	nni_ws          *ws;
-	int              rv;
 
-	if (nni_aio_begin(aio) != 0) {
-		return;
-	}
+	nni_aio_reset(aio);
 	nni_mtx_lock(&l->mtx);
 	if (l->closed) {
 		nni_aio_finish_error(aio, NNG_ECLOSED);
@@ -1742,8 +1736,7 @@ ws_listener_accept(void *arg, nni_aio *aio)
 		nni_aio_finish(aio, 0, 0);
 		return;
 	}
-	if ((rv = nni_aio_schedule(aio, ws_accept_cancel, l)) != 0) {
-		nni_aio_finish_error(aio, rv);
+	if (!nni_aio_start(aio, ws_accept_cancel, l)) {
 		nni_mtx_unlock(&l->mtx);
 		return;
 	}
@@ -2350,9 +2343,7 @@ ws_dialer_dial(void *arg, nni_aio *aio)
 	nni_ws        *ws;
 	int            rv;
 
-	if (nni_aio_begin(aio) != 0) {
-		return;
-	}
+	nni_aio_reset(aio);
 	if ((rv = ws_init(&ws)) != 0) {
 		nni_aio_finish_error(aio, rv);
 		return;
@@ -2364,9 +2355,8 @@ ws_dialer_dial(void *arg, nni_aio *aio)
 		ws_reap(ws);
 		return;
 	}
-	if ((rv = nni_aio_schedule(aio, ws_dial_cancel, ws)) != 0) {
+	if (!nni_aio_start(aio, ws_dial_cancel, ws)) {
 		nni_mtx_unlock(&d->mtx);
-		nni_aio_finish_error(aio, rv);
 		ws_reap(ws);
 		return;
 	}
@@ -2741,9 +2731,7 @@ ws_str_send(void *arg, nni_aio *aio)
 	int       rv;
 	ws_frame *frame;
 
-	if (nni_aio_begin(aio) != 0) {
-		return;
-	}
+	nni_aio_reset(aio);
 
 	if (!ws->isstream) {
 		nni_msg *msg;
@@ -2786,9 +2774,8 @@ ws_str_send(void *arg, nni_aio *aio)
 		ws_frame_fini(frame);
 		return;
 	}
-	if ((rv = nni_aio_schedule(aio, ws_write_cancel, ws)) != 0) {
+	if (!nni_aio_start(aio, ws_write_cancel, ws)) {
 		nni_mtx_unlock(&ws->mtx);
-		nni_aio_finish_error(aio, rv);
 		ws_frame_fini(frame);
 		return;
 	}
@@ -2803,15 +2790,11 @@ static void
 ws_str_recv(void *arg, nng_aio *aio)
 {
 	nni_ws *ws = arg;
-	int     rv;
 
-	if (nni_aio_begin(aio) != 0) {
-		return;
-	}
+	nni_aio_reset(aio);
 	nni_mtx_lock(&ws->mtx);
-	if ((rv = nni_aio_schedule(aio, ws_read_cancel, ws)) != 0) {
+	if (!nni_aio_start(aio, ws_read_cancel, ws)) {
 		nni_mtx_unlock(&ws->mtx);
-		nni_aio_finish_error(aio, rv);
 		return;
 	}
 	nni_list_append(&ws->recvq, aio);