apif_execution/
client.rs

1use anyhow::Result;
2use async_trait::async_trait;
3use serde_json::Value;
4use std::collections::HashMap;
5use std::pin::Pin;
6
7/// How the client communicates with the server.
8#[derive(Debug, Clone, Copy, PartialEq, Eq)]
9pub enum RpcMode {
10    Unary,
11    ServerStream,
12    ClientStream,
13    Bidi,
14}
15
16/// Endpoint metadata resolved by the client.
17#[derive(Debug, Clone)]
18pub struct EndpointMeta {
19    pub rpc_mode: RpcMode,
20    pub input_type: Option<String>,
21    pub output_type: Option<String>,
22}
23
24/// A single item from the response stream.
25#[derive(Debug, Clone)]
26pub enum CallStreamItem {
27    Message(Value),
28    Trailers(HashMap<String, String>),
29}
30
31/// Protocol-agnostic call error.
32#[derive(Debug, Clone)]
33pub struct CallError {
34    pub code: i32,
35    pub message: String,
36}
37
38/// How requests are sent to the server.
39pub enum CallRequest {
40    Unary(Value),
41    Streaming(Pin<Box<dyn futures::Stream<Item = Value> + Send>>),
42}
43
44/// Protocol-agnostic call client trait.
45///
46/// Each protocol (gRPC, HTTP) implements this trait.
47/// The runner uses this trait instead of directly creating a gRPC client.
48#[async_trait]
49pub trait CallClient: Send {
50    /// Resolve endpoint metadata (RPC mode, input/output types).
51    async fn resolve_endpoint(&self, endpoint: &str) -> Result<EndpointMeta>;
52
53    /// Make a call and return a stream of response items.
54    async fn call(
55        &mut self,
56        endpoint: &str,
57        request: CallRequest,
58    ) -> Result<Pin<Box<dyn futures::Stream<Item = Result<CallStreamItem, CallError>> + Send>>>;
59}
60
61/// Factory that creates call clients for specific documents/configs.
62#[async_trait]
63pub trait CallClientFactory: Send + Sync {
64    async fn create_client(
65        &self,
66        config: &crate::config::CallClientConfig,
67    ) -> Result<Box<dyn CallClient>>;
68}
69
70#[cfg(test)]
71mod tests {
72    use super::*;
73    use std::collections::HashMap;
74
75    #[test]
76    fn test_rpc_mode_debug() {
77        assert_eq!(format!("{:?}", RpcMode::Unary), "Unary");
78        assert_eq!(format!("{:?}", RpcMode::ServerStream), "ServerStream");
79        assert_eq!(format!("{:?}", RpcMode::ClientStream), "ClientStream");
80        assert_eq!(format!("{:?}", RpcMode::Bidi), "Bidi");
81    }
82
83    #[test]
84    fn test_call_error_new() {
85        let err = CallError {
86            code: 5,
87            message: "not found".into(),
88        };
89        assert_eq!(err.code, 5);
90        assert_eq!(err.message, "not found");
91    }
92
93    #[test]
94    fn test_call_stream_item_message() {
95        let item = CallStreamItem::Message(serde_json::json!({"key": "value"}));
96        match item {
97            CallStreamItem::Message(v) => assert_eq!(v["key"], "value"),
98            _ => panic!("expected Message"),
99        }
100    }
101
102    #[test]
103    fn test_call_stream_item_trailers() {
104        let mut h = HashMap::new();
105        h.insert("x-status".into(), "ok".into());
106        let item = CallStreamItem::Trailers(h);
107        match item {
108            CallStreamItem::Trailers(t) => assert_eq!(t.get("x-status"), Some(&"ok".into())),
109            _ => panic!("expected Trailers"),
110        }
111    }
112
113    #[test]
114    fn test_endpoint_meta_new() {
115        let meta = EndpointMeta {
116            rpc_mode: RpcMode::Unary,
117            input_type: Some("test.Request".into()),
118            output_type: Some("test.Response".into()),
119        };
120        assert_eq!(meta.rpc_mode, RpcMode::Unary);
121        assert_eq!(meta.input_type.as_deref(), Some("test.Request"));
122    }
123}
apif_execution/client.rs

apif_execution/
client.rs
