feat: axum w/ websockets

Author: prestwich

Cargo.toml

@@ -73,16 +73,20 @@ inherits = "release"
 debug = 2
 strip = false
 
-[profile.bench]
-inherits = "profiling"
-
 [profile.ci-rust]
 inherits = "dev"
 strip = true
 debug = false
 incremental = false
 
-
 [[example]]
 name = "ipc"
-required-features = ["ipc"]
+required-features = ["ipc"]
+
+[[example]]
+name = "axum"
+required-features = ["axum", "pubsub"]
+
+[[example]]
+name = "cors"
+required-features = ["axum"]

examples/axum.rs

@@ -0,0 +1,56 @@
+use ajj::HandlerCtx;
+use std::net::SocketAddr;
+
+#[tokio::main]
+async fn main() -> eyre::Result<()> {
+    let router = make_router();
+
+    // Serve via `POST` on `/` and Websockets on `/ws`
+    let axum = router.clone().into_axum_with_ws("/", "/ws");
+
+    // Now we can serve the router on a TCP listener
+    let addr = SocketAddr::from(([127, 0, 0, 1], 0));
+    let listener = tokio::net::TcpListener::bind(addr).await?;
+
+    println!("Listening for POST on {}/", listener.local_addr()?);
+    println!("Listening for WS on {}/ws", listener.local_addr()?);
+
+    println!("use Ctrl-C to stop");
+    axum::serve(listener, axum).await.map_err(Into::into)
+}
+
+fn make_router() -> ajj::Router<()> {
+    ajj::Router::<()>::new()
+        .route("helloWorld", || async {
+            tracing::info!("serving hello world");
+            Ok::<_, ()>("Hello, world!")
+        })
+        .route("addNumbers", |(a, b): (u32, u32)| async move {
+            tracing::info!("serving addNumbers");
+            Ok::<_, ()>(a + b)
+        })
+        .route("notify", |ctx: HandlerCtx| async move {
+            // Check if notifications are enabled for the connection.
+            if !ctx.notifications_enabled() {
+                // This error will appear in the ResponsePayload's `data` field.
+                return Err("notifications are disabled");
+            }
+
+            let req_id = 15u8;
+
+            // Spawn a task to send the notification after a short delay.
+            ctx.spawn_with_ctx(|ctx| async move {
+                // something expensive goes here
+                let result = 100_000_000;
+                let _ = ctx
+                    .notify(&serde_json::json!({
+                      "req_id": req_id,
+                      "result": result,
+                    }))
+                    .await;
+            });
+
+            // Return the request ID immediately.
+            Ok(req_id)
+        })
+}

examples/cors.rs

@@ -14,7 +14,7 @@
 
 use axum::http::{HeaderValue, Method};
 use eyre::{ensure, Context};
-use std::{future::IntoFuture, net::SocketAddr};
+use std::net::SocketAddr;
 use tower_http::cors::{AllowOrigin, Any, CorsLayer};
 
 #[tokio::main]
@@ -37,10 +37,7 @@ async fn main() -> eyre::Result<()> {
         println!("(specify cors domains as a comma-separated list to restrict origins)");
     }
     println!("use Ctrl-C to stop");
-    axum::serve(listener, router)
-        .into_future()
-        .await
-        .map_err(Into::into)
+    axum::serve(listener, router).await.map_err(Into::into)
 }
 
 fn get_allowed(cors: &str) -> eyre::Result<AllowOrigin> {

examples/ipc.rs

@@ -1,9 +1,19 @@
 //! Basic example of running a JSON-RPC server over IPC using ajj
 //!
-//! This example demonstrates how to set up a simple IPC server using `ajj`.
+//! This example demonstrates how to set up a simple IPC server using `ajj`, and
+//! defines a few basic RPC methods.
+//!
+//! The `make_router` function sets up a router with three routes:
+//! - `helloWorld`: Returns a greeting string.
+//! - `addNumbers`: Takes two numbers as parameters and returns their sum.
+//! - `notify`: Sends a notification after a short delay, demonstrating the use
+//!   of notifications in the RPC context.
 
 use ajj::{
-    pubsub::{ipc::local_socket, Connect},
+    pubsub::{
+        ipc::{to_name, ListenerOptions},
+        Connect,
+    },
     HandlerCtx, Router,
 };
 use tempfile::NamedTempFile;
@@ -20,10 +30,7 @@ async fn main() -> eyre::Result<()> {
     println!("use Ctrl-C to stop");
 
     // The guard keeps the server running until dropped.
-    let guard = ajj::pubsub::ipc::ListenerOptions::new()
-        .name(name)
-        .serve(router)
-        .await?;
+    let guard = ListenerOptions::new().name(name).serve(router).await?;
 
     // Shut down on Ctrl-C
     tokio::signal::ctrl_c().await?;
@@ -32,14 +39,6 @@ async fn main() -> eyre::Result<()> {
     Ok(())
 }
 
-fn to_name(path: &std::ffi::OsStr) -> std::io::Result<local_socket::Name<'_>> {
-    if cfg!(windows) && !path.as_encoded_bytes().starts_with(br"\\.\pipe\") {
-        local_socket::ToNsName::to_ns_name::<local_socket::GenericNamespaced>(path)
-    } else {
-        local_socket::ToFsName::to_fs_name::<local_socket::GenericFilePath>(path)
-    }
-}
-
 // Setting up an AJJ router is easy and fun!
 fn make_router() -> Router<()> {
     Router::<()>::new()

src/pubsub/mod.rs

@@ -98,7 +98,19 @@ pub use ipc_inner::ReadJsonStream;
 /// IPC support via interprocess local sockets.
 #[cfg(feature = "ipc")]
 pub mod ipc {
+    use std::ffi::OsStr;
+
     pub use interprocess::local_socket::{self as local_socket, Listener, ListenerOptions, Name};
+
+    /// Convenience function to convert an [`OsStr`] to a local socket [`Name`]
+    /// in a platform-safe way.
+    pub fn to_name(path: &OsStr) -> std::io::Result<local_socket::Name<'_>> {
+        if cfg!(windows) && !path.as_encoded_bytes().starts_with(br"\\.\pipe\") {
+            local_socket::ToNsName::to_ns_name::<local_socket::GenericNamespaced>(path)
+        } else {
+            local_socket::ToFsName::to_fs_name::<local_socket::GenericFilePath>(path)
+        }
+    }
 }
 
 mod shared;