feat: ipc serving example

Author: prestwich

Cargo.toml

@@ -81,3 +81,8 @@ inherits = "dev"
 strip = true
 debug = false
 incremental = false
+
+
+[[example]]
+name = "ipc"
+required-features = ["ipc"]

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.
@@ -17,6 +17,32 @@ use eyre::{ensure, Context};
 use std::{future::IntoFuture, 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)
+        .into_future()
+        .await
+        .map_err(Into::into)
+}
+
 fn get_allowed(cors: &str) -> eyre::Result<AllowOrigin> {
     // Wildcard `*` means any origin is allowed.
     if cors == "*" {
@@ -53,12 +79,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 +91,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,78 @@
+//! 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`.
+
+use ajj::{
+    pubsub::{ipc::local_socket, 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 = ajj::pubsub::ipc::ListenerOptions::new()
+        .name(name)
+        .serve(router)
+        .await?;
+
+    // Shut down on Ctrl-C
+    tokio::signal::ctrl_c().await?;
+    drop(guard);
+
+    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()
+        .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)
 //!   })

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

@@ -89,10 +89,17 @@
 //! [`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 {
+    pub use interprocess::local_socket::{self as local_socket, Listener, ListenerOptions, Name};
+}
 
 mod shared;
 pub(crate) use shared::WriteItem;