gproxy_transform/envelope/
synthesize.rs

1use bytes::Bytes;
2use gproxy_protocol::{ContentGenerationKind as Kind, StreamFraming};
3
4use super::SseFrame;
5use crate::TransformError;
6
7/// Convert one complete content-generation response into a strict stream.
8pub fn synthesize_response(
9    kind: Kind,
10    body: Bytes,
11    framing: StreamFraming,
12) -> Result<Vec<Bytes>, TransformError> {
13    match kind {
14        Kind::OpenAiChat => {
15            require_framing(framing, &[StreamFraming::Sse])?;
16            let events = crate::typed::synthesize::openai_chat(serde_json::from_slice(&body)?);
17            encode(events, framing, None, true)
18        }
19        Kind::OpenAiResponses | Kind::OpenAiResponsesWebSocket => {
20            require_framing(framing, &[StreamFraming::Sse, StreamFraming::WebSocket])?;
21            let events = crate::typed::synthesize::openai_responses(serde_json::from_slice(&body)?);
22            encode(events, framing, Some(response_name), false)
23        }
24        Kind::ClaudeMessages => {
25            require_framing(framing, &[StreamFraming::Sse])?;
26            let events = crate::typed::synthesize::claude(serde_json::from_slice(&body)?);
27            encode(events, framing, Some(claude_name), false)
28        }
29        Kind::GeminiGenerateContent => {
30            require_framing(framing, &[StreamFraming::Sse, StreamFraming::JsonArray])?;
31            let events = crate::typed::synthesize::gemini(serde_json::from_slice(&body)?);
32            encode(events, framing, None, false)
33        }
34        #[cfg(not(feature = "exhaustive"))]
35        _ => Err(TransformError::unsupported(
36            "content generation kind",
37            "unrecognized external variant",
38        )),
39    }
40}
41
42fn require_framing(
43    framing: StreamFraming,
44    supported: &[StreamFraming],
45) -> Result<(), TransformError> {
46    if supported.contains(&framing) {
47        Ok(())
48    } else {
49        Err(TransformError::shape(
50            "synthetic stream",
51            "framing is not valid for the target protocol",
52        ))
53    }
54}
55
56fn encode<T: serde::Serialize>(
57    events: Vec<T>,
58    framing: StreamFraming,
59    event_name: Option<fn(&T) -> Option<&str>>,
60    done: bool,
61) -> Result<Vec<Bytes>, TransformError> {
62    match framing {
63        StreamFraming::Sse => {
64            let mut output = events
65                .iter()
66                .map(|event| SseFrame::typed(event_name.and_then(|name| name(event)), event))
67                .collect::<Result<Vec<_>, _>>()?;
68            if done {
69                output.push(SseFrame::encode(None, "[DONE]"));
70            }
71            Ok(output)
72        }
73        StreamFraming::JsonArray => Ok(vec![Bytes::from(serde_json::to_vec(&events)?)]),
74        StreamFraming::WebSocket => events
75            .iter()
76            .map(|event| {
77                serde_json::to_vec(event)
78                    .map(Bytes::from)
79                    .map_err(Into::into)
80            })
81            .collect(),
82    }
83}
84
85fn response_name(event: &gproxy_protocol::openai::ResponseStreamEvent) -> Option<&str> {
86    event.event_name()
87}
88
89fn claude_name(event: &gproxy_protocol::claude::StreamEvent) -> Option<&str> {
90    event.event_name()
91}
gproxy_transform/envelope/synthesize.rs

gproxy_transform/envelope/
synthesize.rs
