feat(anng): migrate to NNG v2 API

Major update to support NNG v2.0:

API changes:
- Use typed nng_err enum instead of raw u32 error constants
- Rename socket functions: nng_close → nng_socket_close,
  nng_send_aio → nng_socket_send, nng_recv_aio → nng_socket_recv
- Update nng_strerror signature to take nng_err enum
- Add NNG initialization via nng_init() (required by NNG 2.0)

Dependency changes:
- Update nng-sys from 0.3.0 to 0.4.0-v2pre.1
- Remove build.rs version detection (no longer needed)
- Remove #[cfg(nng_110)] gates as NNG 2.0 is now baseline

Error handling:
- Use nng_err::NNG_* variants with pattern guards
- Format transport errors directly (NNG doesn't resolve them)
- Consistent 'err' naming in match patterns

Test updates:
- Adjust error kind expectations for NNG 2 behavior
- Update inproc address format expectations
- Enable pubsub tests unconditionally

Author: Felix Obenhuber

.github/workflows/check.yml

@@ -136,12 +136,17 @@ jobs:
       # --feature-powerset runs for every combination of features
       - name: cargo hack nng
         run: cargo hack -p nng --feature-powerset check
-      - name: cargo hack anng
-        run: cargo hack -p anng --feature-powerset check
       - name: remove NNG v1
         run: sudo apt-get remove -y libnng1 libnng-dev
       - name: build and install NNG v2 for nng-sys
         run: .github/scripts/install-nng.sh
+      - name: cargo hack anng
+        run: cargo hack -p anng --feature-powerset check
+        env:
+          # add /usr/local/include to the C compiler search path for NNG headers
+          CFLAGS: -I/usr/local/include
+          # add /usr/local/lib to the linker search path for NNG libraries
+          RUSTFLAGS: -L native=/usr/local/lib
       - name: cargo hack nng-sys
         run: cargo hack -p nng-sys --feature-powerset check --skip source-update-bindings
         env:

Cargo.lock

@@ -27,9 +27,8 @@ tracing = "0.1.41"
 lazy_static = { version = "1.5.0", optional = true }
 
 [dependencies.nng-sys]
-version = "0.3.0"
-# TODO(flxo): use path dependency once anng is updated to use nng-sys v0.4
-# path = "../nng-sys"
+version = "0.4.0-v2pre.1"
+path = "../nng-sys"
 default-features = false
 
 [features]

anng/Cargo.toml

@@ -3,7 +3,8 @@ use crate::{
     aio::{Aio, AioError, ImplicationOnMessage},
     message::Message,
 };
-use core::{fmt, mem::MaybeUninit, num::NonZeroU32, ptr::NonNull};
+use core::{fmt, mem::MaybeUninit, ptr::NonNull};
+use nng_sys::nng_err;
 
 impl<Protocol: crate::protocols::SupportsContext> Socket<Protocol> {
     /// Creates a new context for concurrent operations on this socket.
@@ -26,10 +27,10 @@ impl<Protocol: crate::protocols::SupportsContext> Socket<Protocol> {
         let errno = unsafe { nng_sys::nng_ctx_open(context.as_mut_ptr(), self.socket) };
         match u32::try_from(errno).expect("errno is never negative") {
             0 => {}
-            nng_sys::NNG_ENOMEM => {
+            x if x == nng_err::NNG_ENOMEM as u32 => {
                 panic!("OOM");
             }
-            nng_sys::NNG_ENOTSUP => {
+            x if x == nng_err::NNG_ENOTSUP as u32 => {
                 // the SupportsContext trait bound ensures this method is only callable
                 // on protocols that support contexts, so this error should be impossible.
                 unreachable!("protocol supports contexts per SupportsContext trait bound");
@@ -131,33 +132,27 @@ impl<Protocol> Context<'_, Protocol> {
                 let msg = unsafe { Message::from_raw_unchecked(msg) };
                 Ok(Some(msg))
             }
-            nng_sys::NNG_EAGAIN => Ok(None),
-            nng_sys::NNG_ECLOSED => {
+            x if x == nng_err::NNG_EAGAIN as u32 => Ok(None),
+            x if x == nng_err::NNG_ECLOSED as u32 => {
                 unreachable!("socket is still open since we have a reference to it");
             }
-            nng_sys::NNG_EINVAL => {
+            x if x == nng_err::NNG_EINVAL as u32 => {
                 unreachable!("flags are valid for the call");
             }
-            nng_sys::NNG_ENOMEM => {
+            x if x == nng_err::NNG_ENOMEM as u32 => {
                 panic!("OOM");
             }
-            err @ nng_sys::NNG_ENOTSUP => {
+            x if x == nng_err::NNG_ENOTSUP as u32 => {
                 // protocol does not support receiving
-                Err(AioError::Operation(
-                    NonZeroU32::try_from(err).expect("statically checked to be >0"),
-                ))
+                Err(AioError::from_nng_err(nng_err::NNG_ENOTSUP))
             }
-            err @ nng_sys::NNG_ESTATE => {
+            x if x == nng_err::NNG_ESTATE as u32 => {
                 // protocol does not support receiving in its current state
-                Err(AioError::Operation(
-                    NonZeroU32::try_from(err).expect("statically checked to be >0"),
-                ))
+                Err(AioError::from_nng_err(nng_err::NNG_ESTATE))
             }
-            err @ nng_sys::NNG_ETIMEDOUT => {
+            x if x == nng_err::NNG_ETIMEDOUT as u32 => {
                 // likely due to a protocol-level timeout (like surveys)
-                Err(AioError::Operation(
-                    NonZeroU32::try_from(err).expect("statically checked to be >0"),
-                ))
+                Err(AioError::from_nng_err(nng_err::NNG_ETIMEDOUT))
             }
             errno => {
                 unreachable!(

anng/build.rs

@@ -100,13 +100,8 @@ use crate::{
     aio::{Aio, ImplicationOnMessage},
     context::Context,
 };
-use core::{
-    ffi::{CStr, c_int},
-    fmt,
-    marker::PhantomData,
-    num::NonZeroU32,
-    ptr::NonNull,
-};
+use core::{ffi::CStr, fmt, marker::PhantomData, ptr::NonNull};
+use nng_sys::nng_err;
 use std::io;
 
 mod aio;
@@ -262,7 +257,7 @@ impl<Protocol> fmt::Debug for ContextfulSocket<'_, Protocol> {
 impl<Protocol> Drop for Socket<Protocol> {
     fn drop(&mut self) {
         // SAFETY: socket is valid and not already closed (socket is live until `self` drops).
-        crate::block_in_place(|| unsafe { nng_sys::nng_close(self.socket) });
+        crate::block_in_place(|| unsafe { nng_sys::nng_socket_close(self.socket) });
     }
 }
 
@@ -288,7 +283,7 @@ impl<Protocol> Socket<Protocol> {
         // SAFETY: socket is valid and not closed (socket is live until `self` drops),
         //         AIO is valid and not busy (per `Aio` busy state invariant), and
         //         message has been set on AIO (just above).
-        unsafe { nng_sys::nng_send_aio(self.socket, self.aio.as_ptr()) };
+        unsafe { nng_sys::nng_socket_send(self.socket, self.aio.as_ptr()) };
         // the above started an async operation (makes AIO busy).
         // we call wait() to preserve the Aio busy invariant.
         match self.aio.wait(ImplicationOnMessage::Sent).await {
@@ -326,33 +321,27 @@ impl<Protocol> Socket<Protocol> {
                 let msg = unsafe { Message::from_raw_unchecked(msg) };
                 Ok(Some(msg))
             }
-            nng_sys::NNG_EAGAIN => Ok(None),
-            nng_sys::NNG_ECLOSED => {
+            x if x == nng_err::NNG_EAGAIN as u32 => Ok(None),
+            x if x == nng_err::NNG_ECLOSED as u32 => {
                 unreachable!("socket is still open since we have a reference to it");
             }
-            nng_sys::NNG_EINVAL => {
+            x if x == nng_err::NNG_EINVAL as u32 => {
                 unreachable!("flags are valid for the call");
             }
-            nng_sys::NNG_ENOMEM => {
+            x if x == nng_err::NNG_ENOMEM as u32 => {
                 panic!("OOM");
             }
-            err @ nng_sys::NNG_ENOTSUP => {
+            x if x == nng_err::NNG_ENOTSUP as u32 => {
                 // protocol does not support receiving
-                Err(AioError::Operation(
-                    NonZeroU32::try_from(err).expect("statically checked to be >0"),
-                ))
+                Err(AioError::from_nng_err(nng_err::NNG_ENOTSUP))
             }
-            err @ nng_sys::NNG_ESTATE => {
+            x if x == nng_err::NNG_ESTATE as u32 => {
                 // protocol does not support receiving in its current state
-                Err(AioError::Operation(
-                    NonZeroU32::try_from(err).expect("statically checked to be >0"),
-                ))
+                Err(AioError::from_nng_err(nng_err::NNG_ESTATE))
             }
-            err @ nng_sys::NNG_ETIMEDOUT => {
+            x if x == nng_err::NNG_ETIMEDOUT as u32 => {
                 // likely due to a protocol-level timeout (like surveys)
-                Err(AioError::Operation(
-                    NonZeroU32::try_from(err).expect("statically checked to be >0"),
-                ))
+                Err(AioError::from_nng_err(nng_err::NNG_ETIMEDOUT))
             }
             errno => {
                 unreachable!("nng_recvmsg documentation claims errno {errno} is never returned");
@@ -370,7 +359,7 @@ impl<Protocol> Socket<Protocol> {
 
         // SAFETY: socket is valid and not closed (socket is live until `self` drops), and
         //         AIO is valid and not busy (per `Aio` busy state invariant).
-        unsafe { nng_sys::nng_recv_aio(self.socket, self.aio.as_ptr()) };
+        unsafe { nng_sys::nng_socket_recv(self.socket, self.aio.as_ptr()) };
         // the above started an async operation (which makes the AIO busy), so we must eventually
         // call wait() later to preserve the Aio busy invariant.
         //
@@ -527,7 +516,7 @@ impl<'socket, Protocol> ContextfulSocket<'socket, Protocol> {
     }
 }
 
-fn nng_strerror(errno: c_int) -> &'static CStr {
+fn nng_strerror(errno: nng_sys::nng_err) -> &'static CStr {
     // SAFETY: nng_strerror has no additional safety requirements.
     let raw = unsafe { nng_sys::nng_strerror(errno) };
     // SAFETY: nng_strerror returns a valid null-terminated string.
@@ -538,8 +527,8 @@ fn nng_strerror(errno: c_int) -> &'static CStr {
     cstr
 }
 
-fn nng_errno_to_string(errno: c_int) -> String {
-    nng_strerror(errno).to_string_lossy().into_owned()
+fn nng_err_to_string(err: nng_sys::nng_err) -> String {
+    nng_strerror(err).to_string_lossy().into_owned()
 }
 
 /// Helper function that calls `tokio::task::block_in_place` when appropriate.