Merge pull request #88 from osobiehl/feature/jb/add-dtls-support

Support for DTLS and other transport layers, make client async

Cargo.toml

@@ -25,6 +25,23 @@ bytes = "^1.1"
 coap-lite = "0.11.3"
 lru_time_cache = "0.11.11"
 mio = "=0.8.8"               # fix windows broken, remove it after mio updated
+async-trait = "0.1.74"
+
+# dependencies for dtls
+webrtc-dtls = {version = "0.8.0", optional = true}
+webrtc-util = {version = "0.8.0", optional = true}
+rustls  = {version = "^0.21.1", optional = true}
+rustls-pemfile  = {version = "2.0.0", optional = true}
+rcgen  = {version = "^0.11.0", optional = true}
+pkcs8  = {version = "0.10.2", optional = true}
+sec1 = { version = "0.7.3", features = ["pem", "pkcs8", "std"], optional = true}
+
+[features]
+default = ["dtls"]
+dtls = ["dep:webrtc-dtls", "dep:webrtc-util", "dep:rustls", "dep:rustls-pemfile", "dep:rcgen", "dep:pkcs8", "dep:sec1"]
+
 
 [dev-dependencies]
 quickcheck = "0.8.2"
+
+

README.md

@@ -13,6 +13,8 @@ Features:
 - CoAP Observe option [RFC 7641](https://tools.ietf.org/rfc/rfc7641.txt)
 - *Too Many Requests* Response Code [RFC 8516](https://tools.ietf.org/html/rfc8516)
 - Block-Wise Transfers [RFC 7959](https://tools.ietf.org/html/rfc7959)
+- DTLS support via [webrtc-rs](https://github.com/webrtc-rs/webrtc)
+- Option to provide custom transports for client and server
 
 [Documentation](https://docs.rs/coap/)
 
@@ -31,48 +33,50 @@ tokio = { version = "1", features = ["full"] }
 
 ### Server:
 ```rust
-use coap_lite::{RequestType as Method};
-use coap::Server;
-use tokio::runtime::Runtime;
+ use coap_lite::{RequestType as Method, CoapRequest};
+ use coap::Server;
+ use tokio::runtime::Runtime;
+ use std::net::SocketAddr;
+ fn main() {
+     let addr = "127.0.0.1:5683";
+ 	Runtime::new().unwrap().block_on(async move {
+         let mut server = Server::new_udp(addr).unwrap();
+         println!("Server up on {}", addr);
 
-fn main() {
-    let addr = "127.0.0.1:5683";
+         server.run(|mut request: Box<CoapRequest<SocketAddr>>| async {
+             match request.get_method() {
+                 &Method::Get => println!("request by get {}", request.get_path()),
+                 &Method::Post => println!("request by post {}", String::from_utf8(request.message.payload.clone()).unwrap()),
+                 &Method::Put => println!("request by put {}", String::from_utf8(request.message.payload.clone()).unwrap()),
+                 _ => println!("request by other method"),
+             };
 
-    Runtime::new().unwrap().block_on(async move {
-        let mut server = Server::new(addr).unwrap();
-        println!("Server up on {}", addr);
-
-        server.run(|request| async {
-            match request.get_method() {
-                &Method::Get => println!("request by get {}", request.get_path()),
-                &Method::Post => println!("request by post {}", String::from_utf8(request.message.payload).unwrap()),
-                &Method::Put => println!("request by put {}", String::from_utf8(request.message.payload).unwrap()),
-                _ => println!("request by other method"),
-            };
-
-            return match request.response {
-                Some(mut message) => {
-                    message.message.payload = b"OK".to_vec();
-                    Some(message)
-                },
-                _ => None
-            };
-        }).await.unwrap();
-    });
-}
+             match request.response {
+                 Some(ref mut message) => {
+                     message.message.payload = b"OK".to_vec();
+
+                 },
+                 _ => {}
+             };
+             return request
+         }).await.unwrap();
+     });
+ }
 ```
 
 ### Client:
 ```rust
-use coap::CoAPClient;
-
-fn main() {
-    let url = "coap://127.0.0.1:5683/Rust";
-    println!("Client request: {}", url);
+ use coap_lite::{RequestType as Method, CoapRequest};
+ use coap::{UdpCoAPClient};
+ use tokio::main;
+ #[tokio::main]
+ async fn main() {
+     let url = "coap://127.0.0.1:5683/Rust";
+     println!("Client request: {}", url);
 
-    let response = CoAPClient::get(url).unwrap();
-    println!("Server reply: {}", String::from_utf8(response.message.payload).unwrap());
-}
+     let response = UdpCoAPClient::get(url).await.unwrap();
+     println!("Server reply: {}", String::from_utf8(response.message.payload).unwrap());
+ }
 ```
 
 ## Benchmark

appveyor.yml

@@ -1,6 +1,6 @@
 install:
   - appveyor DownloadFile https://win.rustup.rs/ -FileName rustup-init.exe
-  - rustup-init -yv --default-toolchain nightly --default-host x86_64-pc-windows-gnu
+  - rustup-init -yv --default-toolchain nightly --default-host x86_64-pc-windows-msvc
   - set PATH=%PATH%;%USERPROFILE%\.cargo\bin
   - rustc -vV
   - cargo -vV

benches/server.rs

@@ -1,41 +1,47 @@
 #![feature(test, async_closure)]
-
 extern crate test;
 
-use coap::{CoAPClient, Server};
+use std::net::SocketAddr;
+
+use coap::{server::UdpCoapListener, Server, UdpCoAPClient};
 use coap_lite::{CoapOption, CoapRequest, MessageType};
-use std::{sync::mpsc, thread};
-use tokio::runtime::Runtime;
+use tokio::{net::UdpSocket, runtime::Runtime};
 
 #[bench]
 fn bench_server_with_request(b: &mut test::Bencher) {
-    let (tx, rx) = mpsc::channel();
-
-    thread::spawn(move || {
-        Runtime::new().unwrap().block_on(async move {
-            let mut server = Server::new("127.0.0.1:0").unwrap();
-
-            tx.send(server.socket_addr().unwrap().port()).unwrap();
-
-            server
-                .run(async move |request| {
-                    let uri_path = request.get_path().to_string();
-
-                    return match request.response {
-                        Some(mut response) => {
-                            response.message.payload = uri_path.as_bytes().to_vec();
-                            Some(response)
-                        }
-                        _ => None,
-                    };
-                })
-                .await
-                .unwrap();
-        });
+    let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
+
+    let rt = Runtime::new().unwrap();
+    rt.spawn(async move {
+        let sock = UdpSocket::bind("127.0.0.1:0").await.unwrap();
+        let addr = sock.local_addr().unwrap();
+        let listener = Box::new(UdpCoapListener::from_socket(sock));
+        let server = Server::from_listeners(vec![listener]);
+
+        tx.send(addr.port()).unwrap();
+
+        server
+            .run(move |mut request: Box<CoapRequest<SocketAddr>>| async {
+                let uri_path = request.get_path().to_string();
+
+                match request.response {
+                    Some(ref mut response) => {
+                        response.message.payload = uri_path.as_bytes().to_vec();
+                    }
+                    _ => {}
+                };
+                return request;
+            })
+            .await
+            .unwrap();
     });
 
-    let server_port = rx.recv().unwrap();
-    let client = CoAPClient::new(format!("127.0.0.1:{}", server_port)).unwrap();
+    let server_port = rx.blocking_recv().unwrap();
+    let client = rt.block_on(async {
+        UdpCoAPClient::new_udp(format!("127.0.0.1:{}", server_port))
+            .await
+            .unwrap()
+    });
 
     let mut request = CoapRequest::new();
     request.message.header.set_version(1);
@@ -48,8 +54,10 @@ fn bench_server_with_request(b: &mut test::Bencher) {
         .add_option(CoapOption::UriPath, "test".to_string().into_bytes());
 
     b.iter(|| {
-        client.send(&request).unwrap();
-        let recv_packet = client.receive().unwrap();
-        assert_eq!(recv_packet.message.payload, b"test".to_vec());
+        rt.block_on(async {
+            client.send(&request).await.unwrap();
+            let recv_packet = client.receive().await.unwrap();
+            assert_eq!(recv_packet.message.payload, b"test".to_vec());
+        });
     });
 }

examples/client.rs

@@ -1,31 +1,33 @@
 extern crate coap;
 
-use coap::CoAPClient;
+use coap::client::ObserveMessage;
+use coap::UdpCoAPClient;
 use std::io;
 use std::io::ErrorKind;
 
-fn main() {
+#[tokio::main]
+async fn main() {
     println!("GET url:");
-    example_get();
+    example_get().await;
 
     println!("POST data to url:");
-    example_post();
+    example_post().await;
 
     println!("PUT data to url:");
-    example_put();
+    example_put().await;
 
     println!("DELETE url:");
-    example_delete();
+    example_delete().await;
 
     println!("Observing:");
-    example_observe();
+    example_observe().await;
 }
 
-fn example_get() {
+async fn example_get() {
     let url = "coap://127.0.0.1:5683/hello/get";
     println!("Client request: {}", url);
 
-    match CoAPClient::get(url) {
+    match UdpCoAPClient::get(url).await {
         Ok(response) => {
             println!(
                 "Server reply: {}",
@@ -42,12 +44,12 @@ fn example_get() {
     }
 }
 
-fn example_post() {
+async fn example_post() {
     let url = "coap://127.0.0.1:5683/hello/post";
     let data = b"data".to_vec();
     println!("Client request: {}", url);
 
-    match CoAPClient::post(url, data) {
+    match UdpCoAPClient::post(url, data).await {
         Ok(response) => {
             println!(
                 "Server reply: {}",
@@ -64,12 +66,12 @@ fn example_post() {
     }
 }
 
-fn example_put() {
+async fn example_put() {
     let url = "coap://127.0.0.1:5683/hello/put";
     let data = b"data".to_vec();
     println!("Client request: {}", url);
 
-    match CoAPClient::put(url, data) {
+    match UdpCoAPClient::put(url, data).await {
         Ok(response) => {
             println!(
                 "Server reply: {}",
@@ -86,11 +88,11 @@ fn example_put() {
     }
 }
 
-fn example_delete() {
+async fn example_delete() {
     let url = "coap://127.0.0.1:5683/hello/delete";
     println!("Client request: {}", url);
 
-    match CoAPClient::delete(url) {
+    match UdpCoAPClient::delete(url).await {
         Ok(response) => {
             println!(
                 "Server reply: {}",
@@ -107,18 +109,20 @@ fn example_delete() {
     }
 }
 
-fn example_observe() {
-    let mut client = CoAPClient::new("127.0.0.1:5683").unwrap();
-    client
+async fn example_observe() {
+    let client = UdpCoAPClient::new_udp("127.0.0.1:5683").await.unwrap();
+    let observe_channel = client
         .observe("/hello/put", |msg| {
             println!(
                 "resource changed {}",
                 String::from_utf8(msg.payload).unwrap()
             );
         })
+        .await
         .unwrap();
 
-    println!("Press any key to stop...");
+    println!("Enter any key to stop...");
 
     io::stdin().read_line(&mut String::new()).unwrap();
+    observe_channel.send(ObserveMessage::Terminate).unwrap();
 }

examples/echo.rs

@@ -1,36 +1,34 @@
 extern crate coap;
 
-use coap::{CoAPClient, Server};
-use std::thread;
-use tokio::runtime::Runtime;
+use std::net::SocketAddr;
 
-fn main() {
-    thread::spawn(move || {
-        Runtime::new().unwrap().block_on(async move {
-            let mut server = Server::new("127.0.0.1:5683").unwrap();
+use coap::{Server, UdpCoAPClient};
+use coap_lite::CoapRequest;
+#[tokio::main]
+async fn main() {
+    let _server_task = tokio::spawn(async move {
+        let server = Server::new_udp("127.0.0.1:5683").unwrap();
+        server
+            .run(|mut request: Box<CoapRequest<SocketAddr>>| async {
+                let uri_path = request.get_path().to_string();
 
-            server
-                .run(|request| async {
-                    let uri_path = request.get_path().to_string();
-
-                    return match request.response {
-                        Some(mut response) => {
-                            response.message.payload = uri_path.as_bytes().to_vec();
-                            Some(response)
-                        }
-                        _ => None,
-                    };
-                })
-                .await
-                .unwrap();
-        });
+                match request.response {
+                    Some(ref mut response) => {
+                        response.message.payload = uri_path.as_bytes().to_vec();
+                    }
+                    _ => {}
+                };
+                return request;
+            })
+            .await
+            .unwrap();
     });
 
     let url = "coap://127.0.0.1:5683/Rust";
     println!("Client request: {}", url);
 
     // Maybe need sleep seconds before start client on some OS: https://github.com/Covertness/coap-rs/issues/75
-    let response = CoAPClient::get(url).unwrap();
+    let response = UdpCoAPClient::get(url).await.unwrap();
     println!(
         "Server reply: {}",
         String::from_utf8(response.message.payload).unwrap()