p2p: add rpc server to monitor example

3 files changed, 11 insertions, 2 deletions

diff --git a/net/src/codec/websocket.rs b/net/src/codec/websocket.rs
index b59a55c..8676810 100644
--- a/net/src/codec/websocket.rs
+++ b/net/src/codec/websocket.rs
@@ -19,5 +19,5 @@ pub trait WebSocketEncoder {
 
 pub trait WebSocketDecoder {
     type DeItem;
-    fn decode(&self, src: &Message) -> Result<Self::DeItem>;
+    fn decode(&self, src: &Message) -> Result<Option<Self::DeItem>>;
 }
diff --git a/net/src/endpoint.rs b/net/src/endpoint.rs
index 0c7ecd1..5aebdf9 100644
--- a/net/src/endpoint.rs
+++ b/net/src/endpoint.rs
@@ -238,6 +238,12 @@ impl ToEndpoint for String {
     }
 }
 
+impl ToEndpoint for Endpoint {
+    fn to_endpoint(&self) -> Result<Endpoint> {
+        Ok(self.clone())
+    }
+}
+
 impl ToEndpoint for &str {
     fn to_endpoint(&self) -> Result<Endpoint> {
         Endpoint::from_str(self)
diff --git a/net/src/stream/websocket.rs b/net/src/stream/websocket.rs
index 9d41626..2626d2f 100644
--- a/net/src/stream/websocket.rs
+++ b/net/src/stream/websocket.rs
@@ -47,7 +47,10 @@ where
 
     pub async fn recv(&mut self) -> Result<C::Item> {
         match self.inner.next().await {
-            Some(msg) => self.codec.decode(&msg?),
+            Some(msg) => match self.codec.decode(&msg?)? {
+                Some(m) => Ok(m),
+                None => todo!(),
+            },
             None => Err(Error::IO(std::io::ErrorKind::ConnectionAborted.into())),
         }
     }