Middleware for metrics

examples/Cargo.toml

@@ -18,6 +18,10 @@ tokio = { version = "1", features = ["full"] }
 name = "http"
 path = "http.rs"
 
+[[example]]
+name = "middleware"
+path = "middleware.rs"
+
 [[example]]
 name = "ws"
 path = "ws.rs"

examples/middleware.rs

@@ -0,0 +1,94 @@
+// Copyright 2019-2021 Parity Technologies (UK) Ltd.
+//
+// Permission is hereby granted, free of charge, to any
+// person obtaining a copy of this software and associated
+// documentation files (the "Software"), to deal in the
+// Software without restriction, including without
+// limitation the rights to use, copy, modify, merge,
+// publish, distribute, sublicense, and/or sell copies of
+// the Software, and to permit persons to whom the Software
+// is furnished to do so, subject to the following
+// conditions:
+//
+// The above copyright notice and this permission notice
+// shall be included in all copies or substantial portions
+// of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF
+// ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED
+// TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A
+// PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT
+// SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY
+// CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
+// OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR
+// IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
+// DEALINGS IN THE SOFTWARE.
+
+use jsonrpsee::{
+	types::traits::Client,
+	utils::server::middleware,
+	ws_client::WsClientBuilder,
+	ws_server::{RpcModule, WsServerBuilder},
+};
+use std::net::SocketAddr;
+use std::sync::atomic;
+
+#[derive(Default)]
+struct ManInTheMiddle {
+	when: atomic::AtomicU64,
+}
+
+impl Clone for ManInTheMiddle {
+	fn clone(&self) -> Self {
+		ManInTheMiddle { when: atomic::AtomicU64::new(self.when.load(atomic::Ordering::SeqCst)) }
+	}
+}
+
+impl middleware::Middleware for ManInTheMiddle {
+	type Instant = u64;
+	fn on_request(&self) -> Self::Instant {
+		self.when.fetch_add(1, atomic::Ordering::SeqCst)
+	}
+
+	fn on_call(&self, name: &str) {
+		println!("They called '{}'", name);
+	}
+
+	fn on_result(&self, name: &str, succeess: bool, started_at: Self::Instant) {
+		println!("call={}, worked? {}, when? {}", name, succeess, started_at);
+	}
+
+	fn on_response(&self, started_at: Self::Instant) {
+		println!("Response started_at={}", started_at);
+	}
+}
+
+#[tokio::main]
+async fn main() -> anyhow::Result<()> {
+	tracing_subscriber::FmtSubscriber::builder()
+		.with_env_filter(tracing_subscriber::EnvFilter::from_default_env())
+		.try_init()
+		.expect("setting default subscriber failed");
+
+	let addr = run_server().await?;
+	let url = format!("ws://{}", addr);
+
+	let client = WsClientBuilder::default().build(&url).await?;
+	let response: String = client.request("say_hello", None).await?;
+	tracing::info!("response: {:?}", response);
+	// TODO: This prints `They called 'blabla'` but nothing more. I expected the `on_response` callback to be called too?
+	let _response: Result<String, _> = client.request("blabla", None).await;
+	let _ = client.request::<String>("say_hello", None).await?;
+
+	Ok(())
+}
+
+async fn run_server() -> anyhow::Result<SocketAddr> {
+	let m = ManInTheMiddle::default();
+	let server = WsServerBuilder::with_middleware(m).build("127.0.0.1:0").await?;
+	let mut module = RpcModule::new(());
+	module.register_method("say_hello", |_, _| Ok("lo"))?;
+	let addr = server.local_addr()?;
+	server.start(module)?;
+	Ok(addr)
+}

jsonrpsee/src/lib.rs

@@ -76,6 +76,10 @@ pub use jsonrpsee_types as types;
 #[cfg(any(feature = "http-server", feature = "ws-server"))]
 pub use jsonrpsee_utils::server::rpc_module::{RpcModule, SubscriptionSink};
 
+/// TODO: (dp) any reason not to export this? narrow the scope to `jsonrpsee_utils::server`?
+#[cfg(any(feature = "http-server", feature = "ws-server"))]
+pub use jsonrpsee_utils as utils;
+
 #[cfg(feature = "http-server")]
 pub use http_server::tracing;
 

proc-macros/tests/ui/incorrect/method/method_unexpected_field.stderr

@@ -1,5 +1,5 @@
 error: Unknown argument `magic`, expected one of: `aliases`, `blocking`, `name`, `param_kind`, `resources`
- --> tests/ui/incorrect/method/method_unexpected_field.rs:6:25
+ --> $DIR/method_unexpected_field.rs:6:25
   |
 6 |     #[method(name = "foo", magic = false)]
   |                            ^^^^^

proc-macros/tests/ui/incorrect/sub/sub_dup_name_override.stderr

@@ -1,5 +1,5 @@
 error: "override" is already defined
- --> tests/ui/incorrect/sub/sub_dup_name_override.rs:9:5
+ --> $DIR/sub_dup_name_override.rs:9:5
   |
 9 |     fn two(&self) -> RpcResult<()>;
   |        ^^^

proc-macros/tests/ui/incorrect/sub/sub_name_override.stderr

@@ -1,5 +1,5 @@
 error: "one" is already defined
- --> tests/ui/incorrect/sub/sub_name_override.rs:7:5
+ --> $DIR/sub_name_override.rs:7:5
   |
 7 |     fn one(&self) -> RpcResult<()>;
   |        ^^^

proc-macros/tests/ui/incorrect/sub/sub_unsupported_field.stderr

@@ -1,5 +1,5 @@
 error: Unknown argument `magic`, expected one of: `aliases`, `item`, `name`, `param_kind`, `unsubscribe_aliases`
- --> tests/ui/incorrect/sub/sub_unsupported_field.rs:6:42
+ --> $DIR/sub_unsupported_field.rs:6:42
   |
 6 |     #[subscription(name = "sub", item = u8, magic = true)]
   |                                             ^^^^^

utils/src/server/middleware.rs

@@ -27,7 +27,7 @@
 //! TODO
 
 /// TODO
-pub trait Middleware: Send + Sync + Clone + 'static {
+pub trait Middleware: Default + Send + Sync + Clone + 'static {
 	/// Intended to carry timestamp of a request, for example `std::time::Instant`. How the middleware
 	/// measures time, if at all, is entirely up to the implementation.
 	type Instant: Send + Copy;

ws-server/src/server.rs

@@ -56,7 +56,7 @@ const MAX_CONNECTIONS: u64 = 100;
 
 /// A WebSocket JSON RPC server.
 #[derive(Debug)]
-pub struct Server<M = ()> {
+pub struct Server<M> {
 	listener: TcpListener,
 	cfg: Settings,
 	stop_monitor: StopMonitor,
@@ -489,13 +489,32 @@ impl Default for Settings {
 }
 
 /// Builder to configure and create a JSON-RPC Websocket server
-#[derive(Debug, Default)]
-pub struct Builder {
+#[derive(Debug)]
+pub struct Builder<M = ()> {
 	settings: Settings,
 	resources: Resources,
+	middleware: M,
+}
+
+impl Default for Builder<()> {
+	fn default() -> Self {
+		Self { settings: Default::default(), resources: Default::default(), middleware: () }
+	}
 }
 
 impl Builder {
+	/// Build a default server.
+	pub fn new() -> Self {
+		Default::default()
+	}
+}
+
+impl<M> Builder<M> {
+	/// Build a server with the specified [`Middleware`].
+	pub fn with_middleware(middleware: M) -> Self {
+		Builder { settings: Default::default(), resources: Default::default(), middleware }
+	}
+
 	/// Set the maximum size of a request body in bytes. Default is 10 MiB.
 	pub fn max_request_body_size(mut self, size: u32) -> Self {
 		self.settings.max_request_body_size = size;
@@ -614,11 +633,11 @@ impl Builder {
 	/// }
 	/// ```
 	///
-	pub async fn build(self, addrs: impl ToSocketAddrs) -> Result<Server, Error> {
+	pub async fn build(self, addrs: impl ToSocketAddrs) -> Result<Server<M>, Error> {
 		let listener = TcpListener::bind(addrs).await?;
 		let stop_monitor = StopMonitor::new();
 		let resources = self.resources;
-		Ok(Server { listener, cfg: self.settings, stop_monitor, resources, middleware: () })
+		Ok(Server { listener, cfg: self.settings, stop_monitor, resources, middleware: self.middleware })
 	}
 }