use bytes::Bytes;
use futures::{stream, SinkExt, StreamExt};
use mcai_worker_sdk::prelude::*;
use srt_tokio::SrtSocket;
use std::time::{Duration, Instant};
use tokio::time::sleep;

pub struct SrtStreamGenerator;

impl SrtStreamGenerator {
  // pub async fn new_json(port: u16) -> Result<()> {
  //   log::info!("Entered productor stream");
  //   let mut srt_socket = SrtSocketBuilder::new_listen()
  //     .local_port(port)
  //     .connect()
  //     .await
  //     .unwrap();
  //
  //   log::info!("Connected productor stream");
  //
  //   let mut stream = stream::unfold(0, |count| async move {
  //     sleep(Duration::from_millis(10)).await;
  //
  //     let data = r#"{"key": "value"}"#;
  //     let payload = Bytes::from(data);
  //     return Some((Ok((Instant::now(), payload)), count + 1));
  //   })
  //   .boxed();
  //   srt_socket.send_all(&mut stream).await.unwrap();
  //   Ok(())
  // }

  pub async fn new_empty(port: u16) -> Result<()> {
    log::info!("Entered empty stream");
    let mut srt_socket = SrtSocket::builder()
      .listen_on(&*format!(":{}", port))
      .await
      .unwrap();

    log::info!("Connected empty stream");

    let mut stream = stream::unfold(0, |count| async move {
      if count >= 1 {
        sleep(Duration::from_millis(200)).await;
      };
      let mut data: Vec<u8> = vec![0x47];
      data.extend(vec![0x00; 187]);
      let payload = Bytes::from(data);
      Some((Ok((Instant::now(), payload)), count + 1))
    })
    .boxed();
    srt_socket
      .send_all(&mut stream)
      .await
      .expect_err("Unexpected completion");
    Ok(())
  }
}