gproxy_protocol/adapt/generate/stream/fanout/
mod.rs

1//! Ordered incremental multi-candidate calls. Child streams are polled one at a
2//! time; their first content is exposed before EOF. No started POST is replayed.
3
4mod bridge;
5mod driver;
6mod event;
7mod images;
8mod prepare;
9mod run;
10
11use super::{
12    StreamChunk, StreamInvocation, StreamSettings, StreamStart, bridge::StreamBridge,
13    event::NativeEvent, invoke::ClientFull, output::Encoder,
14};
15use super::{codec_error, conflict, invalid, limit};
16use crate::{
17    adapt::generate::{
18        GenerationStateAccess,
19        fanout::{FanoutOptions, group_id},
20    },
21    capability::StateStore,
22    transform::{
23        Report, TransformError,
24        identity::{IdentityFlow, IdentityRole},
25    },
26};
27use std::collections::{BTreeMap, BTreeSet};
28
29pub use bridge::FanoutBridge;
30pub use event::FanoutEvent;
31
32/// Caller-owned progress retains every child receipt and the active stream
33/// across cancellation, in memory only. Dropping it cancels the active body.
34pub struct FanoutStream<B: FanoutBridge>
35where
36    B::ClientEvent: FanoutEvent,
37{
38    children: Vec<StreamInvocation<B>>,
39    id: String,
40    settings: StreamSettings,
41    index: usize,
42    seeded: bool,
43    seen: BTreeMap<String, usize>,
44    response_ids: BTreeSet<String>,
45    completed: Vec<ClientFull<B>>,
46    encoder: Encoder,
47    collector: Option<<B::ClientEvent as NativeEvent>::Collector>,
48    created: Option<i64>,
49    report: Report,
50    aggregate: Option<ClientFull<B>>,
51    tail: Option<B::ClientEvent>,
52    tail_emitted: bool,
53    emit_usage: bool,
54    finished: bool,
55    failed: bool,
56
57    retained_bytes: usize,
58}
59
60impl<B: FanoutBridge> FanoutStream<B>
61where
62    B::ClientEvent: FanoutEvent,
63{
64    async fn new<S: StateStore>(
65        mut children: Vec<StreamInvocation<B>>,
66        options: FanoutOptions,
67        state: &GenerationStateAccess<'_, S>,
68    ) -> Result<Self, TransformError> {
69        let ids: Vec<_> = children
70            .iter()
71            .map(|v| v.selected.identities.clone())
72            .collect();
73        let id = group_id(&options)?;
74        let settings = children[0].settings;
75
76        for child in &children {
77            child.binding.check(state)?;
78        }
79        let collector = B::ClientEvent::collector(
80            IdentityFlow::new(options.namespace),
81            ids[0].response_policy.clone(),
82            settings.events,
83        );
84        let emit_usage = children[0].emit_usage;
85        // Child collection always retains measured usage. Only the aggregate's
86        // final wire chunk follows the actual client's include_usage option.
87        for child in &mut children {
88            child.emit_usage = true;
89        }
90        let response_ids = BTreeSet::from([id.clone()]);
91        Ok(Self {
92            children,
93            id,
94            settings,
95            index: 0,
96            seeded: false,
97            seen: Default::default(),
98            response_ids,
99            completed: Vec::new(),
100            encoder: Encoder::new(settings.client_framing, settings.codec),
101            collector: Some(collector),
102            created: None,
103            report: Report::default(),
104            aggregate: None,
105            tail: None,
106            tail_emitted: false,
107            emit_usage,
108            finished: false,
109            failed: false,
110
111            retained_bytes: 0,
112        })
113    }
114    pub fn children(&self) -> &[StreamInvocation<B>] {
115        &self.children
116    }
117    pub fn response_id(&self) -> &str {
118        &self.id
119    }
120    pub fn report(&self) -> &Report {
121        &self.report
122    }
123    pub fn client_result(&self) -> Option<&ClientFull<B>> {
124        self.aggregate.as_ref()
125    }
126    fn fail(&mut self) {
127        self.failed = true;
128        for child in &mut self.children {
129            child.reader = None;
130        }
131    }
132    fn emit(
133        &mut self,
134        event: B::ClientEvent,
135    ) -> Result<StreamChunk<B::ClientEvent>, TransformError> {
136        let bytes = self.encoder.event(&event, self.settings.codec)?;
137        Ok(StreamChunk {
138            event: Some(event),
139            bytes,
140            finished: false,
141        })
142    }
143}
gproxy_protocol/adapt/generate/stream/fanout/mod.rs

gproxy_protocol/adapt/generate/stream/fanout/
mod.rs
