xds_server/
streaming.rs

1//! Shared streaming helpers for xDS discovery services.
2//!
3//! This module provides common streaming logic that is shared across
4//! all discovery services (CDS, EDS, LDS, RDS, SDS) to reduce code
5//! duplication and ensure consistent behavior.
6
7use std::sync::Arc;
8
9use tokio::sync::mpsc;
10use tonic::{Status, Streaming};
11use tracing::{error, info};
12
13use xds_core::{NodeHash, TypeUrl};
14
15use crate::sotw::{SotwHandler, SotwResponse};
16use crate::stream::StreamContext;
17
18pub use xds_types::envoy::service::discovery::v3::{DiscoveryRequest, DiscoveryResponse};
19
20/// Configuration for a discovery stream.
21#[derive(Debug, Clone, Copy)]
22pub struct StreamConfig {
23    /// The type URL this stream handles (as a static string).
24    pub type_url: &'static str,
25    /// Service name for logging.
26    pub service_name: &'static str,
27}
28
29impl StreamConfig {
30    /// Create a new stream configuration.
31    pub const fn new(type_url: &'static str, service_name: &'static str) -> Self {
32        Self {
33            type_url,
34            service_name,
35        }
36    }
37}
38
39/// Common stream configurations for each discovery service.
40pub mod configs {
41    use super::*;
42
43    /// Configuration for Cluster Discovery Service.
44    pub const CDS: StreamConfig = StreamConfig::new(TypeUrl::CLUSTER, "CDS");
45    /// Configuration for Endpoint Discovery Service.
46    pub const EDS: StreamConfig = StreamConfig::new(TypeUrl::ENDPOINT, "EDS");
47    /// Configuration for Listener Discovery Service.
48    pub const LDS: StreamConfig = StreamConfig::new(TypeUrl::LISTENER, "LDS");
49    /// Configuration for Route Discovery Service.
50    pub const RDS: StreamConfig = StreamConfig::new(TypeUrl::ROUTE, "RDS");
51    /// Configuration for Secret Discovery Service.
52    pub const SDS: StreamConfig = StreamConfig::new(TypeUrl::SECRET, "SDS");
53}
54
55/// Result of processing a stream request.
56pub enum StreamAction {
57    /// Send a response to the client.
58    SendResponse(DiscoveryResponse),
59    /// No response needed (client is up to date).
60    NoResponse,
61    /// An error occurred, send error and break.
62    Error(Status),
63    /// Stream should break (channel closed, etc).
64    Break,
65}
66
67/// Handle the common discovery stream logic.
68///
69/// This function processes incoming discovery requests and sends responses
70/// through the provided channel. It handles:
71/// - Type URL validation
72/// - Node extraction from first request
73/// - Request processing via SotwHandler
74/// - Response conversion and sending
75///
76/// # Arguments
77///
78/// * `stream` - The incoming request stream
79/// * `tx` - Channel sender for responses
80/// * `handler` - The SotW handler for processing requests
81/// * `config` - Stream configuration (type URL, service name)
82/// * `convert_response` - Function to convert SotwResponse to DiscoveryResponse
83pub async fn handle_discovery_stream<F>(
84    mut stream: Streaming<DiscoveryRequest>,
85    tx: mpsc::Sender<Result<DiscoveryResponse, Status>>,
86    handler: Arc<SotwHandler>,
87    config: StreamConfig,
88    convert_response: F,
89) where
90    F: Fn(SotwResponse) -> Result<DiscoveryResponse, Status> + Send + 'static,
91{
92    let mut ctx = StreamContext::new();
93    let mut node_hash: Option<NodeHash> = None;
94
95    info!(
96        stream = %ctx.id(),
97        service = config.service_name,
98        "{} stream started",
99        config.service_name
100    );
101
102    while let Some(result) = tokio_stream::StreamExt::next(&mut stream).await {
103        match result {
104            Ok(request) => {
105                // Validate type URL
106                if !request.type_url.is_empty() && request.type_url != config.type_url {
107                    error!(
108                        stream = %ctx.id(),
109                        expected = config.type_url,
110                        got = %request.type_url,
111                        "invalid type URL for {}",
112                        config.service_name
113                    );
114                    continue;
115                }
116
117                // Extract node info from first request
118                if node_hash.is_none() {
119                    if let Some(ref node) = request.node {
120                        let hash = NodeHash::from_id(&node.id);
121                        ctx.set_node(node.id.clone(), hash);
122                        node_hash = Some(hash);
123                    }
124                }
125
126                let hash = match node_hash {
127                    Some(h) => h,
128                    None => {
129                        // First request must include node information
130                        error!(
131                            stream = %ctx.id(),
132                            service = config.service_name,
133                            "first request missing required node information"
134                        );
135                        let _ = tx
136                            .send(Err(Status::invalid_argument(
137                                "first request must include node information",
138                            )))
139                            .await;
140                        break;
141                    }
142                };
143
144                // Process request
145                match handler.process_request(
146                    &ctx,
147                    config.type_url.into(),
148                    &request.version_info,
149                    &request.resource_names,
150                    hash,
151                ) {
152                    Ok(Some(response)) => match convert_response(response) {
153                        Ok(discovery_response) => {
154                            if tx.send(Ok(discovery_response)).await.is_err() {
155                                break;
156                            }
157                        }
158                        Err(e) => {
159                            error!(
160                                stream = %ctx.id(),
161                                error = %e,
162                                "failed to convert response"
163                            );
164                            let _ = tx.send(Err(e)).await;
165                            break;
166                        }
167                    },
168                    Ok(None) => {
169                        // Client is up to date, no response needed
170                    }
171                    Err(e) => {
172                        error!(
173                            stream = %ctx.id(),
174                            error = %e,
175                            "{} request failed",
176                            config.service_name
177                        );
178                        break;
179                    }
180                }
181            }
182            Err(e) => {
183                error!(
184                    stream = %ctx.id(),
185                    error = %e,
186                    "stream error"
187                );
188                break;
189            }
190        }
191    }
192
193    info!(
194        stream = %ctx.id(),
195        service = config.service_name,
196        "{} stream ended",
197        config.service_name
198    );
199}
200
201/// Convert a SotW response to a DiscoveryResponse.
202///
203/// This is a helper function that handles the common conversion logic
204/// for all discovery services.
205pub fn convert_sotw_response(
206    response: SotwResponse,
207    type_url: &str,
208) -> Result<DiscoveryResponse, Status> {
209    use xds_types::google::protobuf::Any;
210
211    let resources: Vec<Any> = response
212        .resources
213        .iter()
214        .map(|r| {
215            r.encode().map(|encoded| Any {
216                type_url: encoded.type_url.clone(),
217                value: encoded.value.clone(),
218            })
219        })
220        .collect::<Result<Vec<_>, _>>()
221        .map_err(|e| Status::internal(format!("failed to encode resource: {}", e)))?;
222
223    Ok(DiscoveryResponse {
224        version_info: response.version_info,
225        resources,
226        type_url: type_url.to_string(),
227        nonce: response.nonce,
228        canary: false,
229        control_plane: None,
230        resource_errors: vec![],
231    })
232}
233
234#[cfg(test)]
235mod tests {
236    use super::*;
237
238    #[test]
239    fn stream_config_creation() {
240        let config = StreamConfig::new(TypeUrl::CLUSTER, "CDS");
241        assert_eq!(config.type_url, TypeUrl::CLUSTER);
242        assert_eq!(config.service_name, "CDS");
243    }
244
245    #[test]
246    fn predefined_configs() {
247        assert_eq!(configs::CDS.type_url, TypeUrl::CLUSTER);
248        assert_eq!(configs::EDS.type_url, TypeUrl::ENDPOINT);
249        assert_eq!(configs::LDS.type_url, TypeUrl::LISTENER);
250        assert_eq!(configs::RDS.type_url, TypeUrl::ROUTE);
251        assert_eq!(configs::SDS.type_url, TypeUrl::SECRET);
252    }
253}
xds_server/streaming.rs

xds_server/
streaming.rs
