1use std::{fmt, net::SocketAddr, str::FromStr, sync::Arc};
2
3use async_trait::async_trait;
4use futures_util::{SinkExt, StreamExt};
5use nnrp_core::{CommonHeader, TransportId, COMMON_HEADER_LEN};
6use nnrp_runtime::{
7 BoxedFramedTransport, FramedListener, FramedTransport, NnrpClient, NnrpClientConfig,
8 NnrpServer, NnrpServerConfig, RuntimeError, RuntimeFrameLimits, RuntimePacket,
9 RuntimeTransportKind,
10};
11use nnrp_transport_provider::{
12 TransportProviderDescriptor, TransportProviderKind, TransportProviderRegistry,
13};
14use rustls::{
15 pki_types::{CertificateDer, PrivatePkcs8KeyDer},
16 ClientConfig, RootCertStore, ServerConfig,
17};
18use tokio::net::{TcpListener, TcpStream, ToSocketAddrs};
19use tokio_rustls::{server::TlsStream, TlsAcceptor};
20use tokio_tungstenite::{
21 accept_async, connect_async, connect_async_tls_with_config,
22 tungstenite::{http::Uri, protocol::CloseFrame, Error as WebSocketError, Message},
23 Connector, MaybeTlsStream, WebSocketStream,
24};
25
26#[derive(Debug, Clone, PartialEq, Eq)]
27pub enum WebSocketEndpoint {
28 Ws(String),
29 Wss(String),
30}
31
32impl WebSocketEndpoint {
33 pub fn ws(uri: impl Into<String>) -> Result<Self, RuntimeError> {
34 parse_endpoint(uri.into(), "ws://", Self::Ws)
35 }
36
37 pub fn wss(uri: impl Into<String>) -> Result<Self, RuntimeError> {
38 parse_endpoint(uri.into(), "wss://", Self::Wss)
39 }
40
41 pub fn as_str(&self) -> &str {
42 match self {
43 Self::Ws(uri) | Self::Wss(uri) => uri,
44 }
45 }
46
47 pub fn is_secure(&self) -> bool {
48 matches!(self, Self::Wss(_))
49 }
50}
51
52impl FromStr for WebSocketEndpoint {
53 type Err = RuntimeError;
54
55 fn from_str(value: &str) -> Result<Self, Self::Err> {
56 if value.starts_with("ws://") {
57 Self::ws(value)
58 } else if value.starts_with("wss://") {
59 Self::wss(value)
60 } else {
61 Err(RuntimeError::UnsupportedTransport(
62 "WebSocket endpoint must use ws:// or wss://",
63 ))
64 }
65 }
66}
67
68impl fmt::Display for WebSocketEndpoint {
69 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
70 formatter.write_str(self.as_str())
71 }
72}
73
74#[derive(Debug)]
75pub struct WebSocketTransport {
76 stream: WebSocketStreamKind,
77 limits: RuntimeFrameLimits,
78}
79
80impl WebSocketTransport {
81 pub async fn connect(endpoint: &WebSocketEndpoint) -> Result<Self, RuntimeError> {
82 Self::connect_with_limits(endpoint, RuntimeFrameLimits::default()).await
83 }
84
85 pub async fn connect_with_limits(
86 endpoint: &WebSocketEndpoint,
87 limits: RuntimeFrameLimits,
88 ) -> Result<Self, RuntimeError> {
89 let (stream, _) = connect_async(endpoint.as_str()).await.map_err(runtime_ws)?;
90 Ok(Self {
91 stream: WebSocketStreamKind::Client(stream),
92 limits,
93 })
94 }
95
96 pub async fn connect_secure_with_limits(
97 endpoint: &WebSocketEndpoint,
98 server_name: &str,
99 trusted_certificate_der: impl Into<Vec<u8>>,
100 limits: RuntimeFrameLimits,
101 ) -> Result<Self, RuntimeError> {
102 if !endpoint.is_secure() {
103 return Err(RuntimeError::UnsupportedTransport(
104 "secure WebSocket connect requires wss://",
105 ));
106 }
107 let uri = endpoint.as_str().parse::<Uri>().map_err(|_| {
108 RuntimeError::UnsupportedTransport("secure WebSocket endpoint is invalid")
109 })?;
110 if uri.host() != Some(server_name) {
111 return Err(RuntimeError::UnsupportedTransport(
112 "secure WebSocket server name must match endpoint host",
113 ));
114 }
115 let mut roots = RootCertStore::empty();
116 roots
117 .add(CertificateDer::from(trusted_certificate_der.into()))
118 .map_err(runtime_io)?;
119 let client_config = ClientConfig::builder()
120 .with_root_certificates(roots)
121 .with_no_client_auth();
122 let connector = Connector::Rustls(Arc::new(client_config));
123 let (stream, _) =
124 connect_async_tls_with_config(endpoint.as_str(), None, false, Some(connector))
125 .await
126 .map_err(runtime_ws)?;
127 Ok(Self {
128 stream: WebSocketStreamKind::Client(stream),
129 limits,
130 })
131 }
132
133 fn server(stream: WebSocketStream<TcpStream>, limits: RuntimeFrameLimits) -> Self {
134 Self {
135 stream: WebSocketStreamKind::Server(stream),
136 limits,
137 }
138 }
139
140 fn secure_server(
141 stream: WebSocketStream<TlsStream<TcpStream>>,
142 limits: RuntimeFrameLimits,
143 ) -> Self {
144 Self {
145 stream: WebSocketStreamKind::SecureServer(stream),
146 limits,
147 }
148 }
149}
150
151#[async_trait]
152impl FramedTransport for WebSocketTransport {
153 fn transport_kind(&self) -> RuntimeTransportKind {
154 RuntimeTransportKind::WebSocket
155 }
156
157 async fn read_packet(&mut self) -> Result<RuntimePacket, RuntimeError> {
158 loop {
159 let message =
160 self.stream
161 .next_message()
162 .await?
163 .ok_or(RuntimeError::UnexpectedMessage(
164 "websocket stream closed before an NNRP binary frame",
165 ))?;
166 match message {
167 Message::Binary(bytes) => {
168 self.limits.validate_packet_len(bytes.len())?;
169 return packet_from_binary(bytes.as_ref(), self.limits);
170 }
171 Message::Text(_) => {
172 return Err(RuntimeError::UnexpectedMessage(
173 "websocket text messages are not valid NNRP data frames",
174 ));
175 }
176 Message::Close(close) => return Err(websocket_close_error(close)),
177 Message::Ping(_) | Message::Pong(_) | Message::Frame(_) => continue,
178 }
179 }
180 }
181
182 async fn write_packet(&mut self, packet: &RuntimePacket) -> Result<(), RuntimeError> {
183 let bytes = packet.to_bytes()?;
184 self.limits.validate_packet_len(bytes.len())?;
185 self.stream
186 .send_message(Message::Binary(bytes.into()))
187 .await
188 .map_err(runtime_ws)
189 }
190
191 async fn close(&mut self) -> Result<(), RuntimeError> {
192 self.stream.close().await.map_err(runtime_ws)
193 }
194}
195
196pub struct WebSocketFramedListener {
197 listener: TcpListener,
198 limits: RuntimeFrameLimits,
199 tls_acceptor: Option<TlsAcceptor>,
200}
201
202impl fmt::Debug for WebSocketFramedListener {
203 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
204 formatter
205 .debug_struct("WebSocketFramedListener")
206 .field("listener", &self.listener)
207 .field("limits", &self.limits)
208 .field("secure", &self.tls_acceptor.is_some())
209 .finish()
210 }
211}
212
213impl WebSocketFramedListener {
214 pub async fn bind(addr: impl ToSocketAddrs) -> Result<Self, RuntimeError> {
215 Self::bind_with_limits(addr, RuntimeFrameLimits::default()).await
216 }
217
218 pub async fn bind_with_limits(
219 addr: impl ToSocketAddrs,
220 limits: RuntimeFrameLimits,
221 ) -> Result<Self, RuntimeError> {
222 Ok(Self {
223 listener: TcpListener::bind(addr).await?,
224 limits,
225 tls_acceptor: None,
226 })
227 }
228
229 pub async fn bind_secure_with_limits(
230 addr: impl ToSocketAddrs,
231 certificate_der: impl Into<Vec<u8>>,
232 private_key_pkcs8_der: impl Into<Vec<u8>>,
233 limits: RuntimeFrameLimits,
234 ) -> Result<Self, RuntimeError> {
235 let certificate_chain = vec![CertificateDer::from(certificate_der.into())];
236 let private_key = PrivatePkcs8KeyDer::from(private_key_pkcs8_der.into());
237 let server_config = ServerConfig::builder()
238 .with_no_client_auth()
239 .with_single_cert(certificate_chain, private_key.into())
240 .map_err(runtime_io)?;
241 Ok(Self {
242 listener: TcpListener::bind(addr).await?,
243 limits,
244 tls_acceptor: Some(TlsAcceptor::from(Arc::new(server_config))),
245 })
246 }
247}
248
249#[async_trait]
250impl FramedListener for WebSocketFramedListener {
251 fn transport_kind(&self) -> RuntimeTransportKind {
252 RuntimeTransportKind::WebSocket
253 }
254
255 fn local_addr(&self) -> Result<SocketAddr, RuntimeError> {
256 Ok(self.listener.local_addr()?)
257 }
258
259 async fn accept(&self) -> Result<BoxedFramedTransport, RuntimeError> {
260 let (stream, _) = self.listener.accept().await?;
261 if let Some(acceptor) = &self.tls_acceptor {
262 let stream = acceptor.accept(stream).await.map_err(runtime_io)?;
263 let websocket = accept_async(stream).await.map_err(runtime_ws)?;
264 return Ok(Box::new(WebSocketTransport::secure_server(
265 websocket,
266 self.limits,
267 )));
268 }
269 let websocket = accept_async(stream).await.map_err(runtime_ws)?;
270 Ok(Box::new(WebSocketTransport::server(websocket, self.limits)))
271 }
272}
273
274#[derive(Debug, Clone, Copy, Default)]
275pub struct WebSocketProvider;
276
277impl WebSocketProvider {
278 pub const NAME: &'static str = "nnrp-transport-websocket";
279
280 pub fn descriptor() -> TransportProviderDescriptor {
281 TransportProviderDescriptor::available(
282 Self::NAME,
283 env!("CARGO_PKG_VERSION"),
284 TransportId::WebSocket,
285 TransportProviderKind::PureRust,
286 )
287 }
288
289 pub fn register(registry: &mut TransportProviderRegistry) {
290 registry.register(Self::descriptor());
291 }
292
293 pub async fn connect_transport(
294 endpoint: &WebSocketEndpoint,
295 ) -> Result<WebSocketTransport, RuntimeError> {
296 WebSocketTransport::connect(endpoint).await
297 }
298
299 pub async fn bind_listener(
300 addr: impl ToSocketAddrs,
301 ) -> Result<WebSocketFramedListener, RuntimeError> {
302 WebSocketFramedListener::bind(addr).await
303 }
304
305 pub async fn connect(
306 endpoint: &WebSocketEndpoint,
307 config: NnrpClientConfig,
308 ) -> Result<NnrpClient, RuntimeError> {
309 NnrpClient::from_transport(
310 Self::connect_transport(endpoint).await?,
311 config.with_transport(RuntimeTransportKind::WebSocket),
312 )
313 }
314
315 pub async fn bind(
316 addr: impl ToSocketAddrs,
317 config: NnrpServerConfig,
318 ) -> Result<NnrpServer, RuntimeError> {
319 NnrpServer::from_listener(
320 Self::bind_listener(addr).await?,
321 config.with_transport(RuntimeTransportKind::WebSocket),
322 )
323 }
324}
325
326pub fn register_websocket_provider(registry: &mut TransportProviderRegistry) {
327 WebSocketProvider::register(registry);
328}
329
330#[derive(Debug)]
331enum WebSocketStreamKind {
332 Client(WebSocketStream<MaybeTlsStream<TcpStream>>),
333 Server(WebSocketStream<TcpStream>),
334 SecureServer(WebSocketStream<TlsStream<TcpStream>>),
335}
336
337impl WebSocketStreamKind {
338 async fn next_message(&mut self) -> Result<Option<Message>, RuntimeError> {
339 match self {
340 Self::Client(stream) => stream.next().await.transpose().map_err(runtime_ws),
341 Self::Server(stream) => stream.next().await.transpose().map_err(runtime_ws),
342 Self::SecureServer(stream) => stream.next().await.transpose().map_err(runtime_ws),
343 }
344 }
345
346 async fn send_message(&mut self, message: Message) -> Result<(), WebSocketError> {
347 match self {
348 Self::Client(stream) => stream.send(message).await,
349 Self::Server(stream) => stream.send(message).await,
350 Self::SecureServer(stream) => stream.send(message).await,
351 }
352 }
353
354 async fn close(&mut self) -> Result<(), WebSocketError> {
355 match self {
356 Self::Client(stream) => stream.close(None).await,
357 Self::Server(stream) => stream.close(None).await,
358 Self::SecureServer(stream) => stream.close(None).await,
359 }
360 }
361}
362
363fn parse_endpoint(
364 value: String,
365 expected_prefix: &'static str,
366 constructor: impl FnOnce(String) -> WebSocketEndpoint,
367) -> Result<WebSocketEndpoint, RuntimeError> {
368 if !value.starts_with(expected_prefix) {
369 return Err(RuntimeError::UnsupportedTransport(
370 "WebSocket endpoint scheme does not match constructor",
371 ));
372 }
373 if value[expected_prefix.len()..].is_empty() {
374 return Err(RuntimeError::UnsupportedTransport(
375 "WebSocket endpoint authority cannot be empty",
376 ));
377 }
378 Ok(constructor(value))
379}
380
381fn packet_from_binary(
382 bytes: &[u8],
383 limits: RuntimeFrameLimits,
384) -> Result<RuntimePacket, RuntimeError> {
385 if bytes.len() < COMMON_HEADER_LEN {
386 return Err(RuntimeError::UnexpectedMessage(
387 "websocket binary message is shorter than an NNRP header",
388 ));
389 }
390 let header = CommonHeader::parse(&bytes[..COMMON_HEADER_LEN])?;
391 limits.validate_packet_len(header.packet_len()?)?;
392 if header.packet_len()? != bytes.len() {
393 return Err(nnrp_core::NnrpError::PacketLengthMismatch {
394 declared: header.packet_len()?,
395 actual: bytes.len(),
396 }
397 .into());
398 }
399 let meta_start = COMMON_HEADER_LEN;
400 let meta_end = meta_start + header.meta_len as usize;
401 let body_end = meta_end + header.body_len as usize;
402 RuntimePacket::from_parts(
403 header,
404 bytes[meta_start..meta_end].to_vec(),
405 bytes[meta_end..body_end].to_vec(),
406 )
407 .map_err(Into::into)
408}
409
410fn runtime_ws(error: WebSocketError) -> RuntimeError {
411 RuntimeError::Io(std::io::Error::other(error))
412}
413
414fn runtime_io(error: impl Into<Box<dyn std::error::Error + Send + Sync>>) -> RuntimeError {
415 RuntimeError::Io(std::io::Error::other(error))
416}
417
418fn websocket_close_error(close: Option<CloseFrame>) -> RuntimeError {
419 let detail = match close {
420 Some(frame) => {
421 let code = u16::from(frame.code);
422 format!(
423 "websocket close frame received before an NNRP data frame: code={code}, reason={}",
424 frame.reason
425 )
426 }
427 None => "websocket close frame received before an NNRP data frame without code".to_string(),
428 };
429 RuntimeError::TransportClosed {
430 transport: RuntimeTransportKind::WebSocket,
431 detail,
432 }
433}
434
435#[cfg(test)]
436mod tests {
437 use super::*;
438 use nnrp_core::{
439 BackpressureLevel, FrameSubmitMetadata, InputProfile, PartialResultMetadata,
440 PayloadKindBitmap, PressureMetadata, ProgressMetadata, ResultClass, ResultPushMetadata,
441 SubmitMode, TileIndexMode, STANDARD_PROFILE_TOKEN,
442 };
443 use nnrp_runtime::{NnrpClientEvent, NnrpResult};
444 use nnrp_transport_provider::{RemoteTransportSupport, TransportPolicy};
445 use tokio_tungstenite::tungstenite::protocol::frame::coding::CloseCode;
446
447 #[test]
448 fn websocket_endpoint_parses_ws_and_wss_schemes() {
449 let ws = "ws://127.0.0.1:8080/nnrp"
450 .parse::<WebSocketEndpoint>()
451 .unwrap();
452 assert_eq!(ws.as_str(), "ws://127.0.0.1:8080/nnrp");
453 assert!(!ws.is_secure());
454 assert_eq!(ws.to_string(), ws.as_str());
455
456 let wss = WebSocketEndpoint::wss("wss://example.test/nnrp").unwrap();
457 assert_eq!(wss.as_str(), "wss://example.test/nnrp");
458 assert!(wss.is_secure());
459
460 assert!("http://example.test".parse::<WebSocketEndpoint>().is_err());
461 assert!(WebSocketEndpoint::ws("wss://example.test").is_err());
462 assert!(WebSocketEndpoint::ws("ws://").is_err());
463 }
464
465 #[test]
466 fn websocket_provider_registers_and_selects_websocket() {
467 let mut registry = TransportProviderRegistry::new();
468 register_websocket_provider(&mut registry);
469 assert_eq!(registry.providers().len(), 1);
470 assert_eq!(registry.providers()[0].name, WebSocketProvider::NAME);
471 assert_eq!(registry.providers()[0].transport_id, TransportId::WebSocket);
472
473 let remote = RemoteTransportSupport::new([TransportId::WebSocket]);
474 let selection = registry
475 .select(&remote, TransportPolicy::ForceWebSocket, None)
476 .expect("websocket provider should satisfy force websocket");
477 assert_eq!(selection.selected.name, WebSocketProvider::NAME);
478 }
479
480 #[tokio::test]
481 async fn websocket_loopback_submits_frame_and_receives_result() -> Result<(), RuntimeError> {
482 let server = WebSocketProvider::bind("127.0.0.1:0", NnrpServerConfig::default()).await?;
483 let endpoint = WebSocketEndpoint::ws(format!("ws://{}", server.local_addr()?))?;
484
485 let server_task = tokio::spawn(async move {
486 let mut session = server.accept().await?;
487 let submit = session.receive_submit().await?;
488 session
489 .send_result(submit.frame_id, token_result(), b"ws-ok".to_vec())
490 .await
491 });
492
493 let client = WebSocketProvider::connect(&endpoint, NnrpClientConfig::default()).await?;
494 let mut session = client.open_session().await?;
495 session.submit(token_submit(), b"hello".to_vec()).await?;
496 let NnrpResult { body, .. } = session.await_result().await?;
497 assert_eq!(body, b"ws-ok");
498
499 server_task
500 .await
501 .map_err(|_| RuntimeError::Internal("websocket server task panicked"))??;
502 Ok(())
503 }
504
505 #[tokio::test]
506 async fn websocket_rejects_text_messages_as_data_frames() -> Result<(), RuntimeError> {
507 let listener = WebSocketFramedListener::bind("127.0.0.1:0").await?;
508 let endpoint = WebSocketEndpoint::ws(format!("ws://{}", listener.local_addr()?))?;
509 let server_task = tokio::spawn(async move {
510 let mut accepted = listener.accept().await?;
511 accepted.read_packet().await
512 });
513
514 let (mut client, _) = connect_async(endpoint.as_str()).await.map_err(runtime_ws)?;
515 client
516 .send(Message::Text("not-nnrp".into()))
517 .await
518 .map_err(runtime_ws)?;
519
520 let error = server_task
521 .await
522 .map_err(|_| RuntimeError::Internal("websocket text server task panicked"))?
523 .expect_err("text messages should be rejected");
524 assert!(matches!(error, RuntimeError::UnexpectedMessage(_)));
525 Ok(())
526 }
527
528 #[tokio::test]
529 async fn websocket_rejects_oversized_binary_messages_before_header_parse(
530 ) -> Result<(), RuntimeError> {
531 let listener =
532 WebSocketFramedListener::bind_with_limits("127.0.0.1:0", RuntimeFrameLimits::new(8))
533 .await?;
534 let endpoint = WebSocketEndpoint::ws(format!("ws://{}", listener.local_addr()?))?;
535 let server_task = tokio::spawn(async move {
536 let mut accepted = listener.accept().await?;
537 accepted.read_packet().await
538 });
539
540 let (mut client, _) = connect_async(endpoint.as_str()).await.map_err(runtime_ws)?;
541 client
542 .send(Message::Binary(vec![0; 9].into()))
543 .await
544 .map_err(runtime_ws)?;
545
546 let error = server_task
547 .await
548 .map_err(|_| RuntimeError::Internal("websocket oversized server task panicked"))?
549 .expect_err("oversized binary messages should be rejected before parsing");
550 assert!(matches!(
551 error,
552 RuntimeError::FrameTooLarge {
553 declared: 9,
554 max: 8
555 }
556 ));
557 Ok(())
558 }
559
560 #[tokio::test]
561 async fn websocket_maps_close_frame_to_transport_diagnostics() -> Result<(), RuntimeError> {
562 let listener = WebSocketFramedListener::bind("127.0.0.1:0").await?;
563 let endpoint = WebSocketEndpoint::ws(format!("ws://{}", listener.local_addr()?))?;
564 let server_task = tokio::spawn(async move {
565 let mut accepted = listener.accept().await?;
566 accepted.read_packet().await
567 });
568
569 let (mut client, _) = connect_async(endpoint.as_str()).await.map_err(runtime_ws)?;
570 client
571 .close(Some(CloseFrame {
572 code: CloseCode::Policy,
573 reason: "policy-close".into(),
574 }))
575 .await
576 .map_err(runtime_ws)?;
577
578 let error = server_task
579 .await
580 .map_err(|_| RuntimeError::Internal("websocket close server task panicked"))?
581 .expect_err("close frames should map to transport diagnostics");
582 match error {
583 RuntimeError::TransportClosed { transport, detail } => {
584 assert_eq!(transport, RuntimeTransportKind::WebSocket);
585 assert!(detail.contains("code=1008"));
586 assert!(detail.contains("policy-close"));
587 }
588 error => panic!("expected transport close diagnostic, got {error:?}"),
589 }
590 Ok(())
591 }
592
593 #[tokio::test]
594 async fn websocket_loopback_routes_progress_partial_result_and_pressure(
595 ) -> Result<(), RuntimeError> {
596 let server = WebSocketProvider::bind("127.0.0.1:0", NnrpServerConfig::default()).await?;
597 let endpoint = WebSocketEndpoint::ws(format!("ws://{}", server.local_addr()?))?;
598
599 let server_task = tokio::spawn(async move {
600 let mut session = server.accept().await?;
601 let submit = session.receive_submit().await?;
602 let credit = session.receive_pressure_update().await?;
603 assert_eq!(credit.metadata.credit_window, 9);
604 session.send_backpressure(soft_backpressure()).await?;
605 session
606 .send_progress(progress(submit.frame_id as u64), b"stage".to_vec())
607 .await?;
608 session
609 .send_partial_result(partial_result(submit.frame_id as u64), b"partial".to_vec())
610 .await
611 });
612
613 let client = WebSocketProvider::connect(&endpoint, NnrpClientConfig::default()).await?;
614 let mut session = client.open_session().await?;
615 let frame_id = session
616 .submit_nowait(token_submit(), b"partial-request".to_vec())
617 .await?;
618 session.send_credit_update(credit_update()).await?;
619
620 match session.await_event().await? {
621 NnrpClientEvent::Backpressure(pressure) => {
622 assert_eq!(pressure.pressure_level, BackpressureLevel::Soft as u16);
623 assert_eq!(pressure.credit_window, 2);
624 }
625 event => panic!("expected backpressure event, got {event:?}"),
626 }
627 match session.await_event().await? {
628 NnrpClientEvent::Progress { metadata, body } => {
629 assert_eq!(metadata.operation_id, frame_id as u64);
630 assert_eq!(metadata.progress_sequence, 1);
631 assert_eq!(metadata.percent_x100, 2_500);
632 assert_eq!(body, b"stage");
633 }
634 event => panic!("expected progress event, got {event:?}"),
635 }
636 match session.await_event().await? {
637 NnrpClientEvent::PartialResult { metadata, body } => {
638 assert_eq!(metadata.operation_id, frame_id as u64);
639 assert_eq!(metadata.result_sequence, 1);
640 assert_eq!(body, b"partial");
641 }
642 event => panic!("expected partial result event, got {event:?}"),
643 }
644
645 server_task
646 .await
647 .map_err(|_| RuntimeError::Internal("websocket control server task panicked"))??;
648 Ok(())
649 }
650
651 fn token_submit() -> FrameSubmitMetadata {
652 FrameSubmitMetadata {
653 src_width: 0,
654 src_height: 0,
655 tile_width: 0,
656 tile_height: 0,
657 tile_count: 0,
658 section_count: 0,
659 frame_class: 0,
660 input_profile: InputProfile::Unspecified,
661 tile_index_mode: TileIndexMode::DenseRange,
662 latency_budget_ms: 25,
663 target_fps_x100: 0,
664 retry_of_frame: 0,
665 tile_base_id: 0,
666 camera_bytes: 0,
667 tile_index_bytes: 0,
668 operation_id: 1,
669 submit_mode: SubmitMode::Inline,
670 budget_policy: 0,
671 loss_tolerance_policy: 0,
672 object_ref_mask: 0,
673 dependency_frame_id: 0,
674 payload_kind_bitmap: PayloadKindBitmap(PayloadKindBitmap::TOKEN_CHUNK),
675 payload_frame_count: 1,
676 }
677 }
678
679 fn token_result() -> ResultPushMetadata {
680 ResultPushMetadata {
681 status_code: 200,
682 result_flags: 0,
683 section_count: 0,
684 tile_count: 0,
685 active_profile_id: STANDARD_PROFILE_TOKEN,
686 inference_ms: 1,
687 queue_ms: 0,
688 server_total_ms: 1,
689 tile_base_id: 0,
690 tile_index_bytes: 0,
691 result_class: ResultClass::Complete,
692 applied_budget_policy: 0,
693 reused_frame_id: 0,
694 covered_tile_count: 0,
695 dropped_tile_count: 0,
696 payload_kind_bitmap: PayloadKindBitmap(PayloadKindBitmap::TOKEN_CHUNK),
697 payload_frame_count: 1,
698 }
699 }
700
701 fn credit_update() -> PressureMetadata {
702 PressureMetadata {
703 scope_id: 1,
704 credit_window: 9,
705 pressure_level: BackpressureLevel::None as u16,
706 pressure_reason: 0,
707 retry_after_ms: 0,
708 flags: 0,
709 }
710 }
711
712 fn soft_backpressure() -> PressureMetadata {
713 PressureMetadata {
714 scope_id: 1,
715 credit_window: 2,
716 pressure_level: BackpressureLevel::Soft as u16,
717 pressure_reason: 1,
718 retry_after_ms: 25,
719 flags: 0,
720 }
721 }
722
723 fn progress(operation_id: u64) -> ProgressMetadata {
724 ProgressMetadata {
725 operation_id,
726 progress_sequence: 1,
727 stage_code: 2,
728 percent_x100: 2_500,
729 object_id: 0,
730 body_bytes: 5,
731 }
732 }
733
734 fn partial_result(operation_id: u64) -> PartialResultMetadata {
735 PartialResultMetadata {
736 operation_id,
737 result_sequence: 1,
738 object_id: 0,
739 delta_sequence: 0,
740 body_bytes: 7,
741 flags: 0,
742 }
743 }
744}