backend: send decoder packets as json

gamestream-webtransport-proxy/src/gamestream/config.rs

@@ -43,7 +43,7 @@ pub fn stream_config(stream: &crate::backend::Stream) -> _STREAM_CONFIGURATION {
         packetSize: 512,
         streamingRemotely: STREAM_CFG_AUTO,
         audioConfiguration: (0x3 << 16) | (2 << 8) | 0xCA,
-        supportedVideoFormats: VIDEO_FORMAT_H265,
+        supportedVideoFormats: VIDEO_FORMAT_H264,
         clientRefreshRateX100: 0,
         colorSpace: COLORSPACE_REC_601,
         colorRange: COLOR_RANGE_LIMITED,

gamestream-webtransport-proxy/src/gamestream/decoder.rs

@@ -7,9 +7,11 @@ use moonlight_common_c_sys::{
     VIDEO_FORMAT_H265_REXT8_444, VIDEO_FRAME_HANDLE,
 };
 use salvo::{http::body::Frame, hyper::body::Buf};
+use serde::Serialize;
 use tokio::sync::mpsc;
 use tracing::{debug, error};
 
+#[derive(Serialize)]
 enum FrameType {
     PFRAME,
     IDR,
@@ -29,6 +31,7 @@ impl TryFrom<i32> for FrameType {
     }
 }
 
+#[derive(Serialize)]
 enum VideoFormat {
     H264,
     H264_HIGH8_444,
@@ -64,6 +67,7 @@ impl TryFrom<i32> for VideoFormat {
     }
 }
 
+#[derive(Serialize)]
 enum BufferType {
     PICDATA,
     SPS,
@@ -87,6 +91,7 @@ impl TryFrom<i32> for BufferType {
     }
 }
 
+#[derive(Serialize)]
 struct Buffer {
     data: Vec<u8>,
     buffer_type: BufferType,
@@ -105,6 +110,7 @@ impl TryFrom<_LENTRY> for Buffer {
     }
 }
 
+#[derive(Serialize)]
 pub enum RendererMessage {
     Setup {
         video_format: VideoFormat,
@@ -123,7 +129,9 @@ pub enum RendererMessage {
         presentation_time: u64,
 
         full_length: usize,
-        buffers: Vec<Buffer>,
+        //buffers: Vec<Buffer>,
+        buffer: Buffer,
+        index: u64,
 
         hdr_active: bool,
         colorspace: u8,
@@ -147,7 +155,7 @@ impl RendererMessage {
         })
     }
 
-    fn from_decode_unit(decode_unit: _DECODE_UNIT) -> Result<Self> {
+    fn from_decode_unit(decode_unit: _DECODE_UNIT) -> Result<Vec<Self>> {
         let mut buffers = Vec::new();
 
         if decode_unit.bufferList.is_null() {
@@ -155,29 +163,34 @@ impl RendererMessage {
         }
         let mut next = unsafe { *decode_unit.bufferList };
 
+        let mut index = 0;
         loop {
             if next.next.is_null() {
                 break;
             }
             let buffer = Buffer::try_from(next)?;
 
-            buffers.push(buffer);
+            let msg = RendererMessage::DecodeUnit {
+                frame_number: <u64>::try_from(decode_unit.frameNumber)?,
+                frame_type: FrameType::try_from(decode_unit.frameType)?,
+                host_processing_latency: decode_unit.frameHostProcessingLatency,
+                receieve_time_ms: decode_unit.receiveTimeMs,
+                enqueue_time_ms: decode_unit.enqueueTimeMs,
+                presentation_time: decode_unit.presentationTimeMs as u64,
+                full_length: <usize>::try_from(decode_unit.fullLength)?,
+                buffer,
+                index,
+                hdr_active: decode_unit.hdrActive,
+                colorspace: decode_unit.colorspace,
+            };
 
+            buffers.push(msg);
+
+            index = index + 1;
             next = unsafe { *next.next };
         }
 
-        Ok(RendererMessage::DecodeUnit {
+        Ok(buffers)
-            frame_number: <u64>::try_from(decode_unit.frameNumber)?,
-            frame_type: FrameType::try_from(decode_unit.frameType)?,
-            host_processing_latency: decode_unit.frameHostProcessingLatency,
-            receieve_time_ms: decode_unit.receiveTimeMs,
-            enqueue_time_ms: decode_unit.enqueueTimeMs,
-            presentation_time: decode_unit.presentationTimeMs as u64,
-            full_length: <usize>::try_from(decode_unit.fullLength)?,
-            buffers,
-            hdr_active: decode_unit.hdrActive,
-            colorspace: decode_unit.colorspace,
-        })
     }
 }
 
@@ -245,7 +258,7 @@ extern "C" fn submit_decode_unit_cb(decode_unit: PDECODE_UNIT) -> std::os::raw::
     }
     let decode_unit = unsafe { *decode_unit };
 
-    let msg = match RendererMessage::from_decode_unit(decode_unit) {
+    let messages = match RendererMessage::from_decode_unit(decode_unit) {
         Ok(m) => m,
         Err(e) => {
             error!("Cannot construct RendererMessage: {e}");
@@ -253,7 +266,13 @@ extern "C" fn submit_decode_unit_cb(decode_unit: PDECODE_UNIT) -> std::os::raw::
         }
     };
 
-    send_message(msg)
+    for msg in messages {
+        let ret = send_message(msg);
+        if ret != 0 {
+            return ret;
+        }
+    }
+    0
 }
 
 pub fn decoder_callbacks() -> Result<(DECODER_RENDERER_CALLBACKS, mpsc::Receiver<RendererMessage>)>

gamestream-webtransport-proxy/src/proxy/mod.rs

@@ -48,7 +48,11 @@ async fn proxy_main(
                 match gamestream_packet {
                     Some(frame) => {
                         info!("Got decoder packet");
-                        wt_send.write_all(&[0;32]).await?;
+                        let frame_json = serde_json::to_vec(&frame)?;
+                        let frame_json_len: u32 = <u32>::try_from(frame_json.len())?;
+
+                        wt_send.write_all(&frame_json_len.to_le_bytes()).await?;
+                        wt_send.write_all(&frame_json).await?;
                     }
                     None => {
                         error!("Decoder channel is None");
@@ -61,6 +65,9 @@ async fn proxy_main(
             }
         }
     }
+
+    channels.stop_tx.send(());
+
     Ok(())
 }