knafeh 1.1.0

QUIC-based RPC library with Python bindings
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127

//! Example: Knafeh RPC server in Rust.
//!
//! Run with:
//!     cargo run --no-default-features --example rust_server
//!
//! TLS certificates are generated at startup via `rcgen`.

use async_trait::async_trait;
use knafeh::codec::JsonCodec;
use knafeh::error::{KnafehError, RpcStatusCode};
use knafeh::rpc::message::{RpcRequest, RpcResponse};
use knafeh::rpc::service::{MethodDescriptor, MethodKind, Service};
use knafeh::rpc::stream::{RpcStreamRequest, RpcStreamResponse};
use knafeh::server::Server;
use knafeh::transport::tls::TlsConfig;

struct GreeterService;

#[async_trait]
impl Service for GreeterService {
    fn name(&self) -> &str {
        "greeter"
    }

    fn methods(&self) -> Vec<MethodDescriptor> {
        vec![
            MethodDescriptor {
                name: "say_hello".to_string(),
                kind: MethodKind::Unary,
            },
            MethodDescriptor {
                name: "say_goodbye".to_string(),
                kind: MethodKind::Unary,
            },
        ]
    }

    async fn call_unary(
        &self,
        method: &str,
        request: RpcRequest,
    ) -> Result<RpcResponse, KnafehError> {
        let body: serde_json::Value = serde_json::from_slice(&request.body)?;
        let name = body.get("name").and_then(|v| v.as_str()).unwrap_or("World");

        let response = match method {
            "say_hello" => {
                serde_json::json!({ "message": format!("Hello, {name}!") })
            }
            "say_goodbye" => {
                serde_json::json!({ "message": format!("Goodbye, {name}!") })
            }
            _ => {
                return Err(KnafehError::Service {
                    code: RpcStatusCode::NotFound,
                    message: format!("unknown method: {method}"),
                });
            }
        };

        Ok(RpcResponse::ok(serde_json::to_vec(&response).unwrap()))
    }

    async fn call_server_stream(
        &self,
        _method: &str,
        _request: RpcRequest,
    ) -> Result<RpcStreamResponse, KnafehError> {
        Err(KnafehError::Service {
            code: RpcStatusCode::Unimplemented,
            message: "not implemented".to_string(),
        })
    }

    async fn call_client_stream(
        &self,
        _method: &str,
        _stream: RpcStreamRequest,
    ) -> Result<RpcResponse, KnafehError> {
        Err(KnafehError::Service {
            code: RpcStatusCode::Unimplemented,
            message: "not implemented".to_string(),
        })
    }

    async fn call_bidi_stream(
        &self,
        _method: &str,
        _stream: RpcStreamRequest,
    ) -> Result<RpcStreamResponse, KnafehError> {
        Err(KnafehError::Service {
            code: RpcStatusCode::Unimplemented,
            message: "not implemented".to_string(),
        })
    }
}

/// Generate ephemeral self-signed TLS cert + key for the example.
fn generate_tls_config() -> (TlsConfig, tempfile::TempDir) {
    let ck = rcgen::generate_simple_self_signed(vec!["localhost".to_string()])
        .expect("failed to generate TLS cert");
    let dir = tempfile::tempdir().expect("failed to create temp dir");
    let cert_path = dir.path().join("cert.pem");
    let key_path = dir.path().join("key.pem");
    std::fs::write(&cert_path, ck.cert.pem()).unwrap();
    std::fs::write(&key_path, ck.key_pair.serialize_pem()).unwrap();
    (TlsConfig::server(&cert_path, &key_path), dir)
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    tracing_subscriber::fmt::init();

    let (tls, _cert_dir) = generate_tls_config();

    let server = Server::builder()
        .bind_str("0.0.0.0:4433")?
        .tls(tls)
        .codec(JsonCodec::new())
        .add_service(GreeterService)
        .build()?;

    println!("Starting Knafeh RPC server on 0.0.0.0:4433...");
    server.serve().await?;

    Ok(())
}