Merge pull request #15 from karyontech/codec_api

Update karyon net codec API

Author: hozan23

Cargo.lock

@@ -28,7 +28,7 @@ impl Executor {
 
     #[cfg(feature = "tokio")]
     pub fn handle(&self) -> &tokio::runtime::Handle {
-        return self.inner.handle();
+        self.inner.handle()
     }
 }
 

core/src/async_runtime/executor.rs

@@ -5,12 +5,12 @@ A fast and lightweight async implementation of [JSON-RPC
 
 features: 
 - Supports TCP, TLS, WebSocket, and Unix protocols.
-- Uses `smol`(async-std) as the async runtime, but also supports `tokio` via the 
+- Uses `smol`(async-std) as the async runtime, with support for `tokio` via the 
   `tokio` feature.
-- Allows registration of multiple services (structs) of different types on a
-  single server.
-- Supports pub/sub  
-- Allows passing an `async_executors::Executor` or tokio's `Runtime` when building
+- Enables the registration of multiple services (structs) on a single server.
+- Offers support for custom JSON codec.
+- Includes support for pub/sub.  
+- Allows the use of an `async_executors::Executor` or `tokio::Runtime` when building
   the server.
 
 
@@ -31,7 +31,8 @@ use serde_json::Value;
 use smol::stream::StreamExt;
 
 use karyon_jsonrpc::{
-    RPCError, Server, Client, rpc_impl, rpc_pubsub_impl, SubscriptionID, Channel
+    error::RPCError, server::{Server, ServerBuilder, Channel}, client::ClientBuilder,
+    rpc_impl, rpc_pubsub_impl, message::SubscriptionID,
 };
 
 struct HelloWorld {}
@@ -84,7 +85,7 @@ async {
     let service = Arc::new(HelloWorld {});
     // Creates a new server
 
-    let server = Server::builder("tcp://127.0.0.1:60000")
+    let server = ServerBuilder::new("tcp://127.0.0.1:60000")
         .expect("create new server builder")
         .service(service.clone())
         .pubsub_service(service)
@@ -101,7 +102,7 @@ async {
 // Client
 async {
     // Creates a new client
-    let client = Client::builder("tcp://127.0.0.1:60000")
+    let client = ClientBuilder::new("tcp://127.0.0.1:60000")
         .expect("create new client builder")
         .build()
         .await

jsonrpc/README.md

@@ -4,7 +4,7 @@ use log::info;
 use serde::{Deserialize, Serialize};
 use smol::Timer;
 
-use karyon_jsonrpc::Client;
+use karyon_jsonrpc::client::ClientBuilder;
 
 #[derive(Deserialize, Serialize)]
 struct Req {
@@ -18,7 +18,7 @@ struct Pong {}
 fn main() {
     env_logger::init();
     smol::future::block_on(async {
-        let client = Client::builder("tcp://127.0.0.1:6000")
+        let client = ClientBuilder::new("tcp://127.0.0.1:6000")
             .expect("Create client builder")
             .build()
             .await

jsonrpc/examples/client.rs

@@ -0,0 +1,87 @@
+use std::time::Duration;
+
+use log::info;
+use serde::{Deserialize, Serialize};
+use smol::Timer;
+
+use karyon_jsonrpc::{
+    client::ClientBuilder,
+    codec::{Codec, Decoder, Encoder},
+    error::Error,
+};
+
+#[derive(Deserialize, Serialize)]
+struct Req {
+    x: u32,
+    y: u32,
+}
+
+#[derive(Deserialize, Serialize, Debug)]
+struct Pong {}
+
+#[derive(Clone)]
+pub struct CustomJsonCodec {}
+
+impl Codec for CustomJsonCodec {
+    type Message = serde_json::Value;
+    type Error = Error;
+}
+
+impl Encoder for CustomJsonCodec {
+    type EnMessage = serde_json::Value;
+    type EnError = Error;
+    fn encode(
+        &self,
+        src: &Self::EnMessage,
+        dst: &mut [u8],
+    ) -> std::result::Result<usize, Self::EnError> {
+        let msg = match serde_json::to_string(src) {
+            Ok(m) => m,
+            Err(err) => return Err(Error::Encode(err.to_string())),
+        };
+        let buf = msg.as_bytes();
+        dst[..buf.len()].copy_from_slice(buf);
+        Ok(buf.len())
+    }
+}
+
+impl Decoder for CustomJsonCodec {
+    type DeMessage = serde_json::Value;
+    type DeError = Error;
+    fn decode(
+        &self,
+        src: &mut [u8],
+    ) -> std::result::Result<Option<(usize, Self::DeMessage)>, Self::DeError> {
+        let de = serde_json::Deserializer::from_slice(src);
+        let mut iter = de.into_iter::<serde_json::Value>();
+
+        let item = match iter.next() {
+            Some(Ok(item)) => item,
+            Some(Err(ref e)) if e.is_eof() => return Ok(None),
+            Some(Err(e)) => return Err(Error::Decode(e.to_string())),
+            None => return Ok(None),
+        };
+
+        Ok(Some((iter.byte_offset(), item)))
+    }
+}
+
+fn main() {
+    env_logger::init();
+    smol::future::block_on(async {
+        let client = ClientBuilder::new_with_codec("tcp://127.0.0.1:6000", CustomJsonCodec {})
+            .expect("Create client builder")
+            .build()
+            .await
+            .expect("Create rpc client");
+
+        loop {
+            Timer::after(Duration::from_millis(100)).await;
+            let result: Pong = client
+                .call("Calc.ping", ())
+                .await
+                .expect("Call Calc.ping method");
+            info!("Ping result:  {:?}", result);
+        }
+    });
+}

jsonrpc/examples/client_custom_codec.rs

@@ -4,13 +4,13 @@ use log::info;
 use serde::{Deserialize, Serialize};
 use smol::Timer;
 
-use karyon_jsonrpc::Client;
+use karyon_jsonrpc::client::ClientBuilder;
 
 #[derive(Deserialize, Serialize, Debug)]
 struct Pong {}
 
 async fn run_client() {
-    let client = Client::builder("tcp://127.0.0.1:6000")
+    let client = ClientBuilder::new("tcp://127.0.0.1:6000")
         .expect("Create client builder")
         .build()
         .await

jsonrpc/examples/pubsub_client.rs

@@ -6,7 +6,10 @@ use serde_json::Value;
 
 use karyon_core::async_util::sleep;
 use karyon_jsonrpc::{
-    message::SubscriptionID, rpc_impl, rpc_pubsub_impl, Channel, RPCError, Server,
+    error::RPCError,
+    message::SubscriptionID,
+    rpc_impl, rpc_pubsub_impl,
+    server::{channel::Channel, ServerBuilder},
 };
 
 struct Calc {}
@@ -69,7 +72,7 @@ fn main() {
         let calc = Arc::new(Calc {});
 
         // Creates a new server
-        let server = Server::builder("tcp://127.0.0.1:6000")
+        let server = ServerBuilder::new("tcp://127.0.0.1:6000")
             .expect("Create a new server builder")
             .service(calc.clone())
             .pubsub_service(calc)

jsonrpc/examples/pubsub_server.rs

@@ -4,7 +4,7 @@ use serde::{Deserialize, Serialize};
 use serde_json::Value;
 
 use karyon_core::async_util::sleep;
-use karyon_jsonrpc::{rpc_impl, RPCError, Server};
+use karyon_jsonrpc::{error::RPCError, rpc_impl, server::ServerBuilder};
 
 struct Calc {
     version: String,
@@ -49,7 +49,7 @@ fn main() {
         };
 
         // Creates a new server
-        let server = Server::builder("tcp://127.0.0.1:6000")
+        let server = ServerBuilder::new("tcp://127.0.0.1:6000")
             .expect("Create a new server builder")
             .service(Arc::new(calc))
             .build()

jsonrpc/examples/server.rs

@@ -0,0 +1,99 @@
+use std::{sync::Arc, time::Duration};
+
+use serde::{Deserialize, Serialize};
+use serde_json::Value;
+
+use karyon_core::async_util::sleep;
+
+use karyon_jsonrpc::{
+    codec::{Codec, Decoder, Encoder},
+    error::{Error, RPCError},
+    rpc_impl,
+    server::ServerBuilder,
+};
+
+struct Calc {}
+
+#[derive(Deserialize, Serialize)]
+struct Req {
+    x: u32,
+    y: u32,
+}
+
+#[derive(Deserialize, Serialize)]
+struct Pong {}
+
+#[rpc_impl]
+impl Calc {
+    async fn ping(&self, _params: Value) -> Result<Value, RPCError> {
+        Ok(serde_json::json!(Pong {}))
+    }
+}
+
+#[derive(Clone)]
+pub struct CustomJsonCodec {}
+
+impl Codec for CustomJsonCodec {
+    type Message = serde_json::Value;
+    type Error = Error;
+}
+
+impl Encoder for CustomJsonCodec {
+    type EnMessage = serde_json::Value;
+    type EnError = Error;
+    fn encode(
+        &self,
+        src: &Self::EnMessage,
+        dst: &mut [u8],
+    ) -> std::result::Result<usize, Self::EnError> {
+        let msg = match serde_json::to_string(src) {
+            Ok(m) => m,
+            Err(err) => return Err(Error::Encode(err.to_string())),
+        };
+        let buf = msg.as_bytes();
+        dst[..buf.len()].copy_from_slice(buf);
+        Ok(buf.len())
+    }
+}
+
+impl Decoder for CustomJsonCodec {
+    type DeMessage = serde_json::Value;
+    type DeError = Error;
+    fn decode(
+        &self,
+        src: &mut [u8],
+    ) -> std::result::Result<Option<(usize, Self::DeMessage)>, Self::DeError> {
+        let de = serde_json::Deserializer::from_slice(src);
+        let mut iter = de.into_iter::<serde_json::Value>();
+
+        let item = match iter.next() {
+            Some(Ok(item)) => item,
+            Some(Err(ref e)) if e.is_eof() => return Ok(None),
+            Some(Err(e)) => return Err(Error::Decode(e.to_string())),
+            None => return Ok(None),
+        };
+
+        Ok(Some((iter.byte_offset(), item)))
+    }
+}
+
+fn main() {
+    env_logger::init();
+    smol::block_on(async {
+        // Register the Calc service
+        let calc = Calc {};
+
+        // Creates a new server
+        let server = ServerBuilder::new_with_codec("tcp://127.0.0.1:6000", CustomJsonCodec {})
+            .expect("Create a new server builder")
+            .service(Arc::new(calc))
+            .build()
+            .await
+            .expect("start a new server");
+
+        // Start the server
+        server.start();
+
+        sleep(Duration::MAX).await;
+    });
+}

jsonrpc/examples/server_custom_codec.rs

@@ -4,7 +4,10 @@ use serde::{Deserialize, Serialize};
 use serde_json::Value;
 
 use karyon_jsonrpc::{
-    message::SubscriptionID, rpc_impl, rpc_pubsub_impl, Channel, RPCError, Server,
+    error::RPCError,
+    message::SubscriptionID,
+    rpc_impl, rpc_pubsub_impl,
+    server::{Channel, ServerBuilder},
 };
 
 struct Calc {
@@ -84,7 +87,7 @@ async fn main() {
     });
 
     // Creates a new server
-    let server = Server::builder("ws://127.0.0.1:6000")
+    let server = ServerBuilder::new("ws://127.0.0.1:6000")
         .expect("Create a new server builder")
         .service(calc.clone())
         .pubsub_service(calc)