feat: ipc and axum serving example (#33)

* feat: ipc serving example

* feat: axum w/ websockets

* chore: bump version

Author: prestwich

Cargo.toml

@@ -5,7 +5,7 @@ description = "Simple, modern, ergonomic JSON-RPC 2.0 router built with tower an
 keywords = ["json-rpc", "jsonrpc", "json"]
 categories = ["web-programming::http-server", "web-programming::websocket"]
 
-version = "0.4.0"
+version = "0.4.1"
 edition = "2021"
 rust-version = "1.81"
 authors = ["init4", "James Prestwich"]
@@ -73,11 +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"]
+
+[[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

@@ -1,4 +1,4 @@
-//! Basic example of using ajj with CORS.
+//! Basic example of using ajj with CORS via axum.
 //!
 //! This example demonstrates how to set up a simple HTTP server using `axum`
 //! and `tower_http` for CORS support.
@@ -14,9 +14,32 @@
 
 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]
+async fn main() -> eyre::Result<()> {
+    let cors = std::env::args().nth(1).unwrap_or("*".to_string());
+
+    let router = make_router()
+        // Convert to an axum router
+        .into_axum("/")
+        // And then layer on your CORS settings
+        .layer(make_cors(&cors)?);
+
+    // 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 on {}", listener.local_addr()?);
+    println!("CORS allowed for: {}", cors);
+    if cors == "*" {
+        println!("(specify cors domains as a comma-separated list to restrict origins)");
+    }
+    println!("use Ctrl-C to stop");
+    axum::serve(listener, router).await.map_err(Into::into)
+}
+
 fn get_allowed(cors: &str) -> eyre::Result<AllowOrigin> {
     // Wildcard `*` means any origin is allowed.
     if cors == "*" {
@@ -53,12 +76,10 @@ fn make_cors(cors: &str) -> eyre::Result<CorsLayer> {
         .allow_headers(Any))
 }
 
-#[tokio::main]
-async fn main() -> eyre::Result<()> {
-    let cors = std::env::args().nth(1).unwrap_or("*".to_string());
-
+// Setting up an AJJ router is easy and fun!
+fn make_router() -> ajj::Router<()> {
     // Setting up an AJJ router is easy and fun!
-    let router = ajj::Router::<()>::new()
+    ajj::Router::<()>::new()
         .route("helloWorld", || async {
             tracing::info!("serving hello world");
             Ok::<_, ()>("Hello, world!")
@@ -67,15 +88,4 @@ async fn main() -> eyre::Result<()> {
             tracing::info!("serving addNumbers");
             Ok::<_, ()>(a + b)
         })
-        // Convert to an axum router
-        .into_axum("/")
-        // And then layer on your CORS settings
-        .layer(make_cors(&cors)?);
-
-    // 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?;
-
-    axum::serve(listener, router).into_future().await?;
-    Ok(())
 }

examples/ipc.rs

@@ -0,0 +1,77 @@
+//! 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`, 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::{to_name, ListenerOptions},
+        Connect,
+    },
+    HandlerCtx, Router,
+};
+use tempfile::NamedTempFile;
+
+#[tokio::main]
+async fn main() -> eyre::Result<()> {
+    let router = make_router();
+
+    // Create a temporary file for the IPC socket.
+    let tempfile = NamedTempFile::new()?;
+    let name = to_name(tempfile.path().as_os_str()).expect("invalid name");
+
+    println!("Serving IPC on socket: {:?}", tempfile.path());
+    println!("use Ctrl-C to stop");
+
+    // The guard keeps the server running until dropped.
+    let guard = ListenerOptions::new().name(name).serve(router).await?;
+
+    // Shut down on Ctrl-C
+    tokio::signal::ctrl_c().await?;
+    drop(guard);
+
+    Ok(())
+}
+
+// Setting up an AJJ router is easy and fun!
+fn make_router() -> Router<()> {
+    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)
+        })
+}

src/lib.rs

@@ -31,13 +31,13 @@
 //!
 //!     let req_id = 15u8;
 //!
-//!     tokio::task::spawn_blocking(move || {
+//!     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;
 //!     });
 //!     Ok(req_id)
 //!   })
@@ -148,7 +148,7 @@
 )]
 #![cfg_attr(not(test), warn(unused_crate_dependencies))]
 #![deny(unused_must_use, rust_2018_idioms)]
-#![cfg_attr(docsrs, feature(doc_cfg, doc_auto_cfg))]
+#![cfg_attr(docsrs, feature(doc_cfg))]
 
 #[macro_use]
 pub(crate) mod macros;

src/pubsub/ipc.rs renamed to src/pubsub/ipc_inner.rs

@@ -89,10 +89,29 @@
 //! [`HandlerCtx`]: crate::HandlerCtx
 
 #[cfg(feature = "ipc")]
-mod ipc;
+mod ipc_inner;
 #[cfg(feature = "ipc")]
 #[doc(hidden)]
-pub use ipc::ReadJsonStream;
+// Re-exported for use in tests
+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;
 pub(crate) use shared::WriteItem;