aio: nng_aio_defer replaced by nng_aio_start

This represents an API change, and we remove the nng_aio_begin
function as well, introducing the lightweight nng_aio_reset instead.

15 files changed, 97 insertions, 202 deletions

diff --git a/docs/man/libnng.3.adoc b/docs/man/libnng.3.adoc
index 65ecac55..36117614 100644
--- a/docs/man/libnng.3.adoc
+++ b/docs/man/libnng.3.adoc
@@ -160,11 +160,11 @@ The following functions are used in the asynchronous model:
 |===
 // |xref:nng_aio_abort.3.adoc[nng_aio_abort()]|abort asynchronous I/O operation
 // |xref:nng_aio_alloc.3.adoc[nng_aio_alloc()]|allocate asynchronous I/O handle
-|xref:nng_aio_begin.3.adoc[nng_aio_begin()]|begin asynchronous I/O operation
+|xref:nng_aio_reset.3.adoc[nng_aio_reset()]|reset asynchronous I/O operation
 // |xref:nng_aio_busy.3.adoc[nng_aio_busy()]|test if asynchronous I/O is busy
 // |xref:nng_aio_cancel.3.adoc[nng_aio_cancel()]|cancel asynchronous I/O operation
 // |xref:nng_aio_count.3.adoc[nng_aio_count()]|return number of bytes transferred
-|xref:nng_aio_defer.3.adoc[nng_aio_defer()]|defer asynchronous I/O operation
+|xref:nng_aio_start.3.adoc[nng_aio_start()]|start asynchronous I/O operation
 |xref:nng_aio_finish.3.adoc[nng_aio_finish()]|finish asynchronous I/O operation
 // |xref:nng_aio_free.3.adoc[nng_aio_free()]|free asynchronous I/O handle
 |xref:nng_aio_get_input.3.adoc[nng_aio_get_input()]|return input parameter
diff --git a/docs/man/nng_aio_begin.3.adoc b/docs/man/nng_aio_begin.3.adoc
deleted file mode 100644
index 30a73648..00000000
--- a/docs/man/nng_aio_begin.3.adoc
+++ /dev/null
@@ -1,64 +0,0 @@
-= nng_aio_begin(3)
-//
-// Copyright 2018 Staysail Systems, Inc. <info@staysail.tech>
-// Copyright 2018 Capitar IT Group BV <info@capitar.com>
-//
-// This document is supplied under the terms of the MIT License, a
-// copy of which should be located in the distribution where this
-// file was obtained (LICENSE.txt).  A copy of the license may also be
-// found online at https://opensource.org/licenses/MIT.
-//
-
-== NAME
-
-nng_aio_begin - begin asynchronous I/O operation
-
-== SYNOPSIS
-
-[source, c]
-----
-#include <nng/nng.h>
-
-bool nng_aio_begin(nng_aio *aio);
-----
-
-== DESCRIPTION
-
-The `nng_aio_begin()` function is called by the I/O provider to indicate that
-it is going to process the operation.
-
-The function may return `false`, indicating that the _aio_ has been closed
-by the caller asynchronously.
-In this case the provider should abandon the operation and do nothing else.
-
-This operation should be called at the start of any I/O operation, and must
-be called not more than once for a given I/O operation on a given _aio_.
-
-Once this function is called, if `true` is returned, then the provider MUST
-guarantee that xref:nng_aio_finish.3.adoc[`nng_aio_finish()`] is called for the _aio_
-exactly once, when the operation is complete or canceled.
-
-NOTE: This function is only for I/O providers (those actually performing
-the operation such as HTTP handler functions or transport providers); ordinary
-users of the _aio_ should not call this function.
-
-== RETURN VALUES
-
-[horizontal]
-`true`:: The operation has been started.
-`false`:: The operation cannot be started.
-
-== ERRORS
-
-None.
-
-== SEE ALSO
-
-[.text-left]
-xref:nng_aio_alloc.3.adoc[nng_aio_alloc(3)],
-xref:nng_aio_cancel.3.adoc[nng_aio_cancel(3)],
-xref:nng_aio_defer.3.adoc[nng_aio_defer(3)],
-xref:nng_aio_finish.3.adoc[nng_aio_finish(3)],
-xref:nng_aio_result.3.adoc[nng_aio_result(3)],
-xref:nng_aio.5.adoc[nng_aio(5)],
-xref:nng.7.adoc[nng(7)]
diff --git a/docs/man/nng_aio_finish.3.adoc b/docs/man/nng_aio_finish.3.adoc
index 91ad188e..bb0791b8 100644
--- a/docs/man/nng_aio_finish.3.adoc
+++ b/docs/man/nng_aio_finish.3.adoc
@@ -52,9 +52,9 @@ None.
 
 [.text-left]
 xref:nng_aio_alloc.3.adoc[nng_aio_alloc(3)],
-xref:nng_aio_begin.3.adoc[nng_aio_begin(3)],
+xref:nng_aio_reset.3.adoc[nng_aio_reset(3)],
 xref:nng_aio_cancel.3.adoc[nng_aio_cancel(3)],
-xref:nng_aio_defer.3.adoc[nng_aio_defer(3)],
+xref:nng_aio_start.3.adoc[nng_aio_start(3)],
 xref:nng_aio_result.3.adoc[nng_aio_result(3)],
 xref:nng_aio.5.adoc[nng_aio(5)],
 xref:nng.7.adoc[nng(7)]
diff --git a/docs/man/nng_aio_reset.3.adoc b/docs/man/nng_aio_reset.3.adoc
new file mode 100644
index 00000000..cb0feec0
--- /dev/null
+++ b/docs/man/nng_aio_reset.3.adoc
@@ -0,0 +1,42 @@
+= nng_aio_reset(3)
+//
+// Copyright 2024 Staysail Systems, Inc. <info@staysail.tech>
+//
+// This document is supplied under the terms of the MIT License, a
+// copy of which should be located in the distribution where this
+// file was obtained (LICENSE.txt).  A copy of the license may also be
+// found online at https://opensource.org/licenses/MIT.
+//
+
+== NAME
+
+nng_aio_reset - reset asynchronous I/O operation
+
+== SYNOPSIS
+
+[source, c]
+----
+#include <nng/nng.h>
+
+void nng_aio_reset(nng_aio *aio);
+----
+
+== DESCRIPTION
+
+The `nng_aio_reset()` is used by providers to reset certain fields in the _aio_.
+This should be done before performing any operations on _aio_.
+
+NOTE: This function is only for I/O providers (those actually performing
+the operation such as HTTP handler functions or transport providers); ordinary
+users of the _aio_ should not call this function.
+
+== SEE ALSO
+
+[.text-left]
+xref:nng_aio_alloc.3.adoc[nng_aio_alloc(3)],
+xref:nng_aio_cancel.3.adoc[nng_aio_cancel(3)],
+xref:nng_aio_start.3.adoc[nng_aio_start(3)],
+xref:nng_aio_finish.3.adoc[nng_aio_finish(3)],
+xref:nng_aio_result.3.adoc[nng_aio_result(3)],
+xref:nng_aio.5.adoc[nng_aio(5)],
+xref:nng.7.adoc[nng(7)]
diff --git a/docs/man/nng_aio_defer.3.adoc b/docs/man/nng_aio_start.3.adoc
index 5664a6df..82dbde23 100644
--- a/docs/man/nng_aio_defer.3.adoc
+++ b/docs/man/nng_aio_start.3.adoc
@@ -1,6 +1,6 @@
-= nng_aio_defer(3)
+= nng_aio_start(3)
 //
-// Copyright 2018 Staysail Systems, Inc. <info@staysail.tech>
+// Copyright 2024 Staysail Systems, Inc. <info@staysail.tech>
 // Copyright 2018 Capitar IT Group BV <info@capitar.com>
 //
 // This document is supplied under the terms of the MIT License, a
@@ -11,7 +11,7 @@
 
 == NAME
 
-nng_aio_defer - defer asynchronous I/O operation
+nng_aio_start - start asynchronous I/O operation
 
 == SYNOPSIS
 
@@ -21,18 +21,18 @@ nng_aio_defer - defer asynchronous I/O operation
 
 typedef void (*nng_aio_cancelfn)(nng_aio *aio, void *arg, int err);
 
-void nng_aio_defer(nng_aio *aio, nng_aio_cancelfn fn, void *arg);
+void nng_aio_start(nng_aio *aio, nng_aio_cancelfn fn, void *arg);
 ----
 
 == DESCRIPTION
 
-The `nng_aio_defer()` function marks operation associated with _aio_ as
+The `nng_aio_started()` function marks operation associated with _aio_ as
 being deferred for asynchronous completion, and also registers a cancellation
 function _fn_ and associated argument _arg_, thereby
 permitting the operation to be canceled.
 
 If the _aio_ is being canceled, the cancellation routine _fn_ will be called
-with the _aio_, the _arg_ specified by `nng_aio_defer()`, and an error
+with the _aio_, the _arg_ specified by `nng_aio_start()`, and an error
 value in _err_, which is the reason that the operation is being canceled.
 
 The operation may not be cancelable; for example it may have already been
@@ -56,17 +56,9 @@ the routine are multi-thread safe; this will usually involve the use
 of locks or other synchronization primitives.
 
 TIP: For operations that complete synchronously, without any need to be
-deferred, the provider should not bother to call `nng_aio_defer()`,
+deferred, the provider should not bother to call `nng_aio_start()`,
 although it is harmless if it does.
 
-== RETURN VALUES
-
-None.
-
-== ERRORS
-
-None.
-
 == SEE ALSO
 
 [.text-left]
diff --git a/docs/man/nng_http_handler_alloc.3http.adoc b/docs/man/nng_http_handler_alloc.3http.adoc
index 7795bbf5..9291fd56 100644
--- a/docs/man/nng_http_handler_alloc.3http.adoc
+++ b/docs/man/nng_http_handler_alloc.3http.adoc
@@ -98,11 +98,7 @@ then a generic 500 response will be created and
 sent, if possible, and the connection will be closed.
 
 The _aio_ may be scheduled for deferred completion using the
-xref:nng_aio_defer.3.adoc[`nng_aio_defer()`] function.
-
-NOTE: The callback function should *NOT* call
-xref:nng_aio_begin.3.adoc[`nng_aio_begin()`],
-as that will already have been done by the server framework.
+xref:nng_aio_start.3.adoc[`nng_aio_start()`] function.
 
 === Directory Handler
 
@@ -181,7 +177,7 @@ These functions return 0 on success, and non-zero otherwise.
 == SEE ALSO
 
 [.text-left]
-xref:nng_aio_defer.3.adoc[nng_aio_defer(3)],
+xref:nng_aio_start.3.adoc[nng_aio_start(3)],
 xref:nng_aio_finish.3.adoc[nng_aio_finish(3)],
 xref:nng_aio_get_input.3.adoc[nng_aio_get_input(3)],
 xref:nng_aio_set_output.3.adoc[nng_aio_set_output(3)],
diff --git a/docs/ref/api/aio.md b/docs/ref/api/aio.md
index 966201ba..6e084e43 100644
--- a/docs/ref/api/aio.md
+++ b/docs/ref/api/aio.md
@@ -103,7 +103,7 @@ The {{i:`nng_aio_cancel`}} function acts like `nng_aio_abort`, but uses the erro
 The {{i:`nng_aio_stop`}} function aborts the _aio_ operation with [`NNG_ESTOPPED`],
 and then waits the operation and any associated callback to complete.
 This function also marks _aio_ itself permanently stopped, so that any
-new operations scheduled by I/O providers using [`nng_aio_begin`]
+new operations scheduled by I/O providers using [`nng_aio_start`]
 return false. Thus this function should be used to teardown operations.
 
 > [!TIP]
diff --git a/docs/ref/migrate/nng1.md b/docs/ref/migrate/nng1.md
index 99afb4d3..9e9b6188 100644
--- a/docs/ref/migrate/nng1.md
+++ b/docs/ref/migrate/nng1.md
@@ -21,6 +21,18 @@ not resubmit an operation that returns it. This is particularly important for ca
 resubmit operations. Failure to observe this rule will lead to an infinite loop
 as any further operations on the object will fail immediately with `NNG_ESTOPPED`.
 
+## AIO Provider API changes
+
+The API used for providers for asynchronous I/O operations has changed slightly.
+
+- The `nng_aio_begin` function is removed. However a new [`nng_aio_reset`] function should be called
+  instead, before performing any other operations on an _aio_ object. (This simply clears certain fields.)
+- The `nng_aio_defer` function is replaced, with a very [`nng_aio_start`] function. However, this function
+  has slightly different semantics. It will automatically call the callback if the operation cannot be
+  scheduled.
+- Be aware of the new `NNG_ESTOPPED` error code, for operations on a handle that is being torn down by
+  the consumer.
+
 ## Transport Specific Functions
 
 Transports have not needed to be registered for a long time now,
diff --git a/docs/ref/xref.md b/docs/ref/xref.md
index d5b7c9f0..8f161de8 100644
--- a/docs/ref/xref.md
+++ b/docs/ref/xref.md
@@ -90,9 +90,9 @@
 [`nng_stream_listener_set_tls`]: /TODO.md
 [`nng_stream_listener_get_tls`]: /TODO.md
 [`nng_opts_parse`]: /api/cmd_opts.md#parse-command-line-options
-[`nng_aio_begin`]: /TODO.md
-[`nng_aio_defer`]: /TODO.md
 [`nng_aio_finish`]: /TODO.md
+[`nng_aio_reset`]: /TODO.md
+[`nng_aio_start`]: /TODO.md
 [`nng_aio_set_output`]: /TODO.md
 [`nng_send`]: /TODO.md
 [`nng_recv`]: /TODO.md
diff --git a/include/nng/nng.h b/include/nng/nng.h
index 2563bc1e..b3daedb0 100644
--- a/include/nng/nng.h
+++ b/include/nng/nng.h
@@ -618,10 +618,10 @@ NNG_DECL void nng_aio_set_expire(nng_aio *, nng_time);
 // to succeed if n <= 4, otherwise it may fail due to NNG_ENOMEM.
 NNG_DECL int nng_aio_set_iov(nng_aio *, unsigned, const nng_iov *);
 
-// nng_aio_begin is called by the provider to mark the operation as
-// beginning.  If it returns false, then the provider must take no
-// further action on the aio.
-NNG_DECL bool nng_aio_begin(nng_aio *);
+// nng_aio_reset is called by the provider before doing other operations on the
+// aio.  Its purpose is to clear certain output fields, to avoid accidental
+// reuse from prior operations on the aio.
+NNG_DECL void nng_aio_reset(nng_aio *);
 
 // nng_aio_finish is used to "finish" an asynchronous operation.
 // It should only be called by "providers" (such as HTTP server API users).
@@ -630,7 +630,7 @@ NNG_DECL bool nng_aio_begin(nng_aio *);
 // given aio.
 NNG_DECL void nng_aio_finish(nng_aio *, int);
 
-// nng_aio_defer is used to register a cancellation routine, and indicate
+// nng_aio_start is used to register a cancellation routine, and indicate
 // that the operation will be completed asynchronously.  It must only be
 // called once per operation on an aio, and must only be called by providers.
 // If the operation is canceled by the consumer, the cancellation callback
@@ -642,9 +642,9 @@ NNG_DECL void nng_aio_finish(nng_aio *, int);
 // to the reason for cancellation, e.g. NNG_ETIMEDOUT or NNG_ECANCELED.
 // This returns false if the operation cannot be deferred (because the AIO
 // has been stopped with nng_aio_stop.)  If it does so, then the aio's
-// completion callback will fire with a result of NNG_ECLOSED.
+// completion callback will fire with a result of NNG_ESTOPPED.
 typedef void (*nng_aio_cancelfn)(nng_aio *, void *, int);
-NNG_DECL bool nng_aio_defer(nng_aio *, nng_aio_cancelfn, void *);
+NNG_DECL bool nng_aio_start(nng_aio *, nng_aio_cancelfn, void *);
 
 // nng_aio_sleep does a "sleeping" operation, basically does nothing
 // but wait for the specified number of milliseconds to expire, then
diff --git a/src/core/aio.c b/src/core/aio.c
index 87c79d0d..fa6eb76f 100644
--- a/src/core/aio.c
+++ b/src/core/aio.c
@@ -428,69 +428,6 @@ nni_aio_schedule(nni_aio *aio, nni_aio_cancel_fn cancel, void *data)
 }
 
 bool
-nni_aio_defer(nni_aio *aio, nni_aio_cancel_fn cancel, void *data)
-{
-	nni_aio_expire_q *eq      = aio->a_expire_q;
-	bool              timeout = false;
-
-	if (!aio->a_sleep && !aio->a_use_expire) {
-		// Convert the relative timeout to an absolute timeout.
-		switch (aio->a_timeout) {
-		case NNG_DURATION_ZERO:
-			timeout = true;
-			break;
-		case NNG_DURATION_INFINITE:
-		case NNG_DURATION_DEFAULT:
-			aio->a_expire = NNI_TIME_NEVER;
-			break;
-		default:
-			aio->a_expire = nni_clock() + aio->a_timeout;
-			break;
-		}
-	} else if (aio->a_use_expire && aio->a_expire <= nni_clock()) {
-		timeout = true;
-	}
-
-	nni_mtx_lock(&eq->eq_mtx);
-	if (aio->a_stop || eq->eq_stop) {
-		aio->a_stop    = true;
-		aio->a_sleep   = false;
-		aio->a_result  = NNG_ESTOPPED;
-		aio->a_stopped = true;
-		nni_mtx_unlock(&eq->eq_mtx);
-		nni_task_dispatch(&aio->a_task);
-		return (false);
-	}
-	if (aio->a_abort) {
-		aio->a_sleep = false;
-		aio->a_abort = false;
-		nni_mtx_unlock(&eq->eq_mtx);
-		nni_task_dispatch(&aio->a_task);
-		return (false);
-	}
-	if (timeout) {
-		aio->a_sleep  = false;
-		aio->a_result = aio->a_expire_ok ? 0 : NNG_ETIMEDOUT;
-		aio->a_abort  = false;
-		nni_mtx_unlock(&eq->eq_mtx);
-		nni_task_dispatch(&aio->a_task);
-		return (false);
-	}
-
-	NNI_ASSERT(aio->a_cancel_fn == NULL);
-	aio->a_cancel_fn  = cancel;
-	aio->a_cancel_arg = data;
-
-	// We only schedule expiration if we have a way for the expiration
-	// handler to actively cancel it.
-	if ((aio->a_expire != NNI_TIME_NEVER) && (cancel != NULL)) {
-		nni_aio_expire_add(aio);
-	}
-	nni_mtx_unlock(&eq->eq_mtx);
-	return (true);
-}
-
-bool
 nni_aio_start(nni_aio *aio, nni_aio_cancel_fn cancel, void *data)
 {
 	nni_aio_expire_q *eq      = aio->a_expire_q;
diff --git a/src/core/aio.h b/src/core/aio.h
index d909853f..5346d19b 100644
--- a/src/core/aio.h
+++ b/src/core/aio.h
@@ -170,11 +170,6 @@ extern void         nni_aio_bump_count(nni_aio *, size_t);
 // cannot be canceled, which can lead to apparent hangs.
 extern int nni_aio_schedule(nni_aio *, nni_aio_cancel_fn, void *);
 
-// nni_aio_defer is just like nni_io_schedule, but it also calls the callback
-// automatically if the operation cannot be started because the AIO is stopped
-// or was canceled before this call (but after nni_aio_begin).
-extern bool nni_aio_defer(nni_aio *, nni_aio_cancel_fn, void *);
-
 // nni_aio_reset is called by providers before doing any work -- it resets
 // counts other fields to their initial state.  It will not reset the closed
 // state if the aio has been stopped or closed.
diff --git a/src/core/aio_test.c b/src/core/aio_test.c
index 66682460..a41f608d 100644
--- a/src/core/aio_test.c
+++ b/src/core/aio_test.c
@@ -130,8 +130,7 @@ test_provider_cancel(void)
 	int      rv = 0;
 	// We fake an empty provider that does not do anything.
 	NUTS_PASS(nng_aio_alloc(&aio, NULL, NULL));
-	NUTS_TRUE(nng_aio_begin(aio) == true);
-	nng_aio_defer(aio, cancel, &rv);
+	nng_aio_start(aio, cancel, &rv);
 	nng_aio_cancel(aio);
 	nng_aio_wait(aio);
 	NUTS_TRUE(rv == NNG_ECANCELED);
diff --git a/src/nng.c b/src/nng.c
index 5f3c6f12..0c96fd90 100644
--- a/src/nng.c
+++ b/src/nng.c
@@ -212,10 +212,9 @@ nng_recv_aio(nng_socket s, nng_aio *aio)
 	nni_sock *sock;
 	int       rv;
 
+	nni_aio_reset(aio);
 	if ((rv = nni_sock_find(&sock, s.id)) != 0) {
-		if (nni_aio_begin(aio) == 0) {
-			nni_aio_finish_error(aio, rv);
-		}
+		nni_aio_finish_error(aio, rv);
 		return;
 	}
 	nni_sock_recv(sock, aio);
@@ -228,16 +227,13 @@ nng_send_aio(nng_socket s, nng_aio *aio)
 	nni_sock *sock;
 	int       rv;
 
+	nni_aio_reset(aio);
 	if (nni_aio_get_msg(aio) == NULL) {
-		if (nni_aio_begin(aio) == 0) {
-			nni_aio_finish_error(aio, NNG_EINVAL);
-		}
+		nni_aio_finish_error(aio, NNG_EINVAL);
 		return;
 	}
 	if ((rv = nni_sock_find(&sock, s.id)) != 0) {
-		if (nni_aio_begin(aio) == 0) {
-			nni_aio_finish_error(aio, rv);
-		}
+		nni_aio_finish_error(aio, rv);
 		return;
 	}
 	nni_sock_send(sock, aio);
@@ -326,10 +322,9 @@ nng_ctx_recv(nng_ctx cid, nng_aio *aio)
 	int      rv;
 	nni_ctx *ctx;
 
+	nni_aio_reset(aio);
 	if ((rv = nni_ctx_find(&ctx, cid.id)) != 0) {
-		if (nni_aio_begin(aio) == 0) {
-			nni_aio_finish_error(aio, rv);
-		}
+		nni_aio_finish_error(aio, rv);
 		return;
 	}
 	nni_ctx_recv(ctx, aio);
@@ -342,16 +337,13 @@ nng_ctx_send(nng_ctx cid, nng_aio *aio)
 	int      rv;
 	nni_ctx *ctx;
 
+	nni_aio_reset(aio);
 	if (nni_aio_get_msg(aio) == NULL) {
-		if (nni_aio_begin(aio) == 0) {
-			nni_aio_finish_error(aio, NNG_EINVAL);
-		}
+		nni_aio_finish_error(aio, NNG_EINVAL);
 		return;
 	}
 	if ((rv = nni_ctx_find(&ctx, cid.id)) != 0) {
-		if (nni_aio_begin(aio) == 0) {
-			nni_aio_finish_error(aio, rv);
-		}
+		nni_aio_finish_error(aio, rv);
 		return;
 	}
 	nni_ctx_send(ctx, aio);
@@ -1279,20 +1271,17 @@ nng_device_aio(nng_aio *aio, nng_socket s1, nng_socket s2)
 	nni_sock *sock1 = NULL;
 	nni_sock *sock2 = NULL;
 
+	nni_aio_reset(aio);
 	if ((s1.id > 0) && (s1.id != (uint32_t) -1)) {
 		if ((rv = nni_sock_find(&sock1, s1.id)) != 0) {
-			if (nni_aio_begin(aio) == 0) {
-				nni_aio_finish_error(aio, rv);
-			}
+			nni_aio_finish_error(aio, rv);
 			return;
 		}
 	}
 	if (((s2.id > 0) && (s2.id != (uint32_t) -1)) && (s2.id != s1.id)) {
 		if ((rv = nni_sock_find(&sock2, s2.id)) != 0) {
 			nni_sock_rele(sock1);
-			if (nni_aio_begin(aio) == 0) {
-				nni_aio_finish_error(aio, rv);
-			}
+			nni_aio_finish_error(aio, rv);
 			return;
 		}
 	}
@@ -2064,18 +2053,15 @@ nng_aio_finish(nng_aio *aio, int rv)
 }
 
 bool
-nng_aio_defer(nng_aio *aio, nng_aio_cancelfn fn, void *arg)
+nng_aio_start(nng_aio *aio, nng_aio_cancelfn fn, void *arg)
 {
-	return (nni_aio_defer(aio, fn, arg));
+	return (nni_aio_start(aio, fn, arg));
 }
 
-bool
-nng_aio_begin(nng_aio *aio)
+void
+nng_aio_reset(nng_aio *aio)
 {
-	if (nni_aio_begin(aio) != 0) {
-		return (false);
-	}
-	return (true);
+	nni_aio_reset(aio);
 }
 
 #define xstr(a) str(a)
diff --git a/src/testing/streams.c b/src/testing/streams.c
index 92d52f37..9f4b01bc 100644
--- a/src/testing/streams.c
+++ b/src/testing/streams.c
@@ -95,7 +95,7 @@ stream_xfr_alloc(nng_stream *s, void (*submit)(nng_stream *, nng_aio *),
 	nng_aio_set_timeout(x->upper_aio, 30000);
 	nng_aio_set_timeout(x->lower_aio, 5000);
 
-	nng_aio_begin(x->upper_aio);
+	(void) nng_aio_start(x->upper_aio, NULL, NULL);
 
 	x->s      = s;
 	x->rem    = size;