platform_core/event_stream.rs
1//
2// Copyright 2018-2026 Accenture Technology
3//
4// Licensed under the Apache License, Version 2.0 (the "License");
5// you may not use this file except in compliance with the License.
6// You may obtain a copy of the License at
7//
8// http://www.apache.org/licenses/LICENSE-2.0
9//
10// Unless required by applicable law or agreed to in writing, software
11// distributed under the License is distributed on an "AS IS" BASIS,
12// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13// See the License for the specific language governing permissions and
14// limitations under the License.
15//
16
17//! HTTP response streaming — the producer side of the multi-shot reply route
18//! (Java `EventStreamWriter`).
19//!
20//! Streaming is native to the event system: a caller provides a `reply_to`
21//! address and the callee may send it as many events as it likes. A streaming
22//! HTTP response is a *sequence* of events to the caller's reply route, each
23//! marked with the reserved envelope header `x-event-stream: data | eof |
24//! exception`, until end of transmission. The marker is internal protocol
25//! consumed by the REST automation edge — it never appears on the HTTP wire.
26//! The first data event commits the HTTP head (status, content type, optional
27//! idle-allowance override); writes after close are dropped.
28
29use crate::envelope::EventEnvelope;
30use crate::function::AppError;
31use crate::platform::Platform;
32use crate::post_office::PostOffice;
33
34/// Reserved envelope header marking one event of a streaming HTTP response
35/// (Java `EventStreamWriter.X_EVENT_STREAM`). Values: [`DATA`], [`EOF`],
36/// [`EXCEPTION`]. Absence of the header = single-shot response.
37pub const X_EVENT_STREAM: &str = "x-event-stream";
38
39/// Optional companion header naming a data segment — maps to the SSE
40/// `event:` field (Java `EventStreamWriter.X_EVENT_NAME`).
41pub const X_EVENT_NAME: &str = "x-event-name";
42
43/// One segment of the stream.
44pub const DATA: &str = "data";
45/// End of transmission (optional body = trailing metadata).
46pub const EOF: &str = "eof";
47/// In-band failure (body = the standard error key-values
48/// `'{"type": "error", "status": n, "message": text}'`).
49pub const EXCEPTION: &str = "exception";
50/// Reserved SSE event name of the Event-over-HTTP envelope-mode wire dialect:
51/// a frame with this name carries one base64-encoded serialized EventEnvelope
52/// (Java `EventStreamWriter.ENVELOPE`).
53pub const ENVELOPE: &str = "envelope";
54
55/// Producer helper for a streaming HTTP response — thin sugar over plain
56/// event sends to the caller's reply route (Java `EventStreamWriter`).
57pub struct EventStreamWriter {
58 po: PostOffice,
59 reply_to: String,
60 correlation_id: Option<String>,
61 first_status: i32,
62 first_content_type: Option<String>,
63 first_ttl_seconds: u64,
64 head_sent: bool,
65 closed: bool,
66}
67
68impl EventStreamWriter {
69 /// Create a writer for a reply route and correlation id.
70 ///
71 /// Returns HTTP-400 when the reply route is empty — a streaming producer
72 /// without a reply address has nowhere to stream.
73 pub fn new(
74 platform: &Platform,
75 reply_to: &str,
76 correlation_id: Option<&str>,
77 ) -> Result<Self, AppError> {
78 if reply_to.is_empty() {
79 return Err(AppError::new(
80 400,
81 "Streaming producer requires a reply_to address",
82 ));
83 }
84 Ok(Self {
85 po: PostOffice::new(platform),
86 reply_to: reply_to.to_string(),
87 correlation_id: correlation_id.map(str::to_string),
88 first_status: 200,
89 first_content_type: None,
90 first_ttl_seconds: 0,
91 head_sent: false,
92 closed: false,
93 })
94 }
95
96 /// Create a writer from the incoming request envelope (the usual form for
97 /// an interceptor function).
98 pub fn from_request(platform: &Platform, request: &EventEnvelope) -> Result<Self, AppError> {
99 Self::new(
100 platform,
101 request.reply_to().unwrap_or(""),
102 request.correlation_id(),
103 )
104 }
105
106 /// Optional head control carried by the first outgoing event: response
107 /// status and content type. Later events cannot change the head.
108 pub fn first(&mut self, status: i32, content_type: &str) -> &mut Self {
109 self.first_status = status;
110 self.first_content_type = Some(content_type.to_string());
111 self
112 }
113
114 /// Head control plus an idle-allowance override in seconds between
115 /// segments (rides the first event as the `x-ttl` envelope header).
116 pub fn first_with_ttl(
117 &mut self,
118 status: i32,
119 content_type: &str,
120 ttl_seconds: u64,
121 ) -> &mut Self {
122 self.first_ttl_seconds = ttl_seconds;
123 self.first(status, content_type)
124 }
125
126 /// Send one `data` segment (String, bytes or map — any serializable body).
127 pub async fn write<T: serde::Serialize>(&mut self, segment: T) -> Result<(), AppError> {
128 self.send(segment, None).await
129 }
130
131 /// Send one named segment — the name maps to the SSE `event:` field.
132 pub async fn write_named<T: serde::Serialize>(
133 &mut self,
134 event_name: &str,
135 segment: T,
136 ) -> Result<(), AppError> {
137 self.send(segment, Some(event_name)).await
138 }
139
140 /// Declare end of transmission.
141 pub async fn close(&mut self) -> Result<(), AppError> {
142 self.close_with(serde_json::Value::Null).await
143 }
144
145 /// Declare end of transmission with trailing metadata (rendered as the
146 /// terminal SSE event's data; ignored in chunked mode).
147 pub async fn close_with<T: serde::Serialize>(&mut self, metadata: T) -> Result<(), AppError> {
148 if self.closed {
149 return Ok(());
150 }
151 self.closed = true;
152 let event = self.envelope(EOF, metadata, None)?;
153 self.po.send(event).await
154 }
155
156 /// Declare an in-band failure and end the stream.
157 pub async fn fail(&mut self, error: &AppError) -> Result<(), AppError> {
158 if self.closed {
159 return Ok(());
160 }
161 self.closed = true;
162 let status = if error.status() >= 400 {
163 error.status()
164 } else {
165 500
166 };
167 // the standard error key-values: '{"type": "error", "status": n, "message": text}'
168 let body =
169 serde_json::json!({"type": "error", "status": status, "message": error.message()});
170 let event = self.envelope(EXCEPTION, body, None)?.set_status(status);
171 self.po.send(event).await
172 }
173
174 /// True when the stream has been closed or failed.
175 pub fn is_closed(&self) -> bool {
176 self.closed
177 }
178
179 async fn send<T: serde::Serialize>(
180 &mut self,
181 body: T,
182 event_name: Option<&str>,
183 ) -> Result<(), AppError> {
184 if self.closed {
185 log::debug!(
186 "Segment to {} dropped - stream already closed",
187 self.reply_to
188 );
189 return Ok(());
190 }
191 let event = self.envelope(DATA, body, event_name)?;
192 self.po.send(event).await
193 }
194
195 fn envelope<T: serde::Serialize>(
196 &mut self,
197 marker: &str,
198 body: T,
199 event_name: Option<&str>,
200 ) -> Result<EventEnvelope, AppError> {
201 let mut event = EventEnvelope::new()
202 .set_to(&self.reply_to)
203 .set_header(X_EVENT_STREAM, marker)
204 .set_body(body)?;
205 if let Some(cid) = &self.correlation_id {
206 event = event.set_correlation_id(cid);
207 }
208 if let Some(name) = event_name.filter(|n| !n.is_empty()) {
209 event = event.set_header(X_EVENT_NAME, name);
210 }
211 if !self.head_sent {
212 self.head_sent = true;
213 event = event.set_status(self.first_status);
214 if let Some(content_type) = &self.first_content_type {
215 event = event.set_header("content-type", content_type);
216 }
217 if self.first_ttl_seconds > 0 {
218 event = event.set_header("x-ttl", &self.first_ttl_seconds.to_string());
219 }
220 }
221 Ok(event)
222 }
223}