forward an actual stream to the client

Cargo.lock

@@ -111,6 +111,7 @@ dependencies = [
  "anyhow",
  "futures-util",
  "tokio",
+ "tokio-stream",
  "tokio-tungstenite",
 ]
 
@@ -515,6 +516,17 @@ dependencies = [
  "syn",
 ]
 
+[[package]]
+name = "tokio-stream"
+version = "0.1.12"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8fb52b74f05dbf495a8fba459fdc331812b96aa086d9eb78101fa0d4569c3313"
+dependencies = [
+ "futures-core",
+ "pin-project-lite",
+ "tokio",
+]
+
 [[package]]
 name = "tokio-tungstenite"
 version = "0.18.0"

Cargo.toml

@@ -7,4 +7,5 @@ edition = "2021"
 anyhow = "1.0.69"
 futures-util = { version = "0.3.26", features = ["sink", "std"] }
 tokio = { version = "1.25.0", features = ["macros", "rt-multi-thread", "time"] }
+tokio-stream = "0.1.12"
 tokio-tungstenite = { version = "0.18.0", features = ["rustls-native-certs"] }

src/main.rs

@@ -1,25 +1,21 @@
-use std::io;
-
 use futures_util::{
-    SinkExt,
     StreamExt,
     stream::{
         SplitSink,
         SplitStream,
+        Stream,
     },
 };
-
 use tokio::{
     time::{
-        sleep,
         Duration,
+        interval,
     },
     net::{
         TcpListener,
         TcpStream,
     },
 };
-
 use tokio_tungstenite::{
     accept_async_with_config,
     WebSocketStream,
@@ -30,9 +26,10 @@ use tokio_tungstenite::{
         protocol::WebSocketConfig,
     },
 };
+use tokio_stream::wrappers::IntervalStream;
 
-type Tx = SplitSink<WebSocketStream<tokio::net::TcpStream>, Message>;
+type WsTx = SplitSink<WebSocketStream<tokio::net::TcpStream>, Message>;
-type Rx = SplitStream<WebSocketStream<tokio::net::TcpStream>>;
+type WsRx = SplitStream<WebSocketStream<tokio::net::TcpStream>>;
 
 async fn accept(tcp_stream: TcpStream) -> Result<WebSocketStream<TcpStream>, WsError> {
     let wsconfig = WebSocketConfig {
@@ -44,38 +41,43 @@ async fn accept(tcp_stream: TcpStream) -> Result<WebSocketStream<TcpStream>, WsE
     accept_async_with_config(tcp_stream, Some(wsconfig)).await
 }
 
-async fn handle(mut rx: Rx) -> anyhow::Result<()> {
+async fn handle(mut rx: WsRx) -> anyhow::Result<()> {
     println!("enter incoming loop");
     while let Some(msg) = rx.next().await {
-        msg.map(|x| { println!("Got: {x:?}"); })?;
+        match msg {
+            Ok(Message::Text(x)) => { println!("Got Text Message: {x:?}"); },
+            Ok(x) => { println!("Got another Message: {x:?}"); },
+            Err(x) => { println!("Got ERROR: {x:?}"); },
+        }
     }
     Ok(())
 }
 
-async fn serve(mut tx: Tx) -> anyhow::Result<()> {
+async fn serve<S: Stream<Item = String>>(tx: WsTx, input: S) -> Result<(), WsError> {
-    println!("enter send loop");
+    input
-    let mut c = 0;
+        .map(Message::Text)
-    loop {
+        .map(Ok)
-        sleep(Duration::from_millis(1000)).await;
+        .forward(tx).await
-        c += 1;
-        let txt = format!("bla bla {c}");
-        println!("{txt}");
-        tx.send(Message::Text(txt)).await?;
-    }
 }
 
 #[tokio::main]
-async fn main() -> io::Result<()> {
+async fn main() -> anyhow::Result<()> {
     let listener = TcpListener::bind("127.0.0.1:8080").await?;
 
     while let Ok((tcp_stream, addr)) = listener.accept().await {
         println!("Got a connection from: {addr}");
 
         let ws_result = accept(tcp_stream).await;
-        let (tx, rx) = ws_result.map(|ws| ws.split()).unwrap();
+        let (tx, rx) = ws_result.map(|ws| ws.split())?;
+
+        // in reality this is an mqtt stream
+        let interval_stream = IntervalStream::new(interval(Duration::from_secs(1)));
+        let stream = interval_stream.zip(futures_util::stream::repeat("djshjh".to_string()))
+            .map(|(i, txt)| format!("{i:?}: {txt}"));
+
         tokio::select! {
             res = handle(rx) => println!("Done waiting for messages. Reason: {res:?}"),
-            _ = serve(tx) => println!("Done serving locations.")
+            _ = serve(tx, stream) => println!("Done serving locations.")
         }
     }