Simple subscribe mechanism

Cargo.lock

@@ -14,6 +14,12 @@ version = "0.3.31"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "05f29059c0c2090612e8d742178b0580d2dc940c837851ad723096f87af6663e"
 
+[[package]]
+name = "futures-sink"
+version = "0.3.31"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e575fab7d1e0dcb8d0c7bcf9a63ee213816ab51902e6d244a95819acacf1d4f7"
+
 [[package]]
 name = "libc"
 version = "0.2.178"
@@ -37,7 +43,7 @@ checksum = "69d83b0086dc8ecf3ce9ae2874b2d1290252e2a30720bea58a5c6639b0092873"
 dependencies = [
  "libc",
  "wasi",
- "windows-sys",
+ "windows-sys 0.61.2",
 ]
 
 [[package]]
@@ -73,6 +79,16 @@ dependencies = [
  "libc",
 ]
 
+[[package]]
+name = "socket2"
+version = "0.6.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "17129e116933cf371d018bb80ae557e889637989d8638274fb25622827b03881"
+dependencies = [
+ "libc",
+ "windows-sys 0.60.2",
+]
+
 [[package]]
 name = "syn"
 version = "2.0.111"
@@ -115,8 +131,9 @@ dependencies = [
  "mio",
  "pin-project-lite",
  "signal-hook-registry",
+ "socket2",
  "tokio-macros",
- "windows-sys",
+ "windows-sys 0.61.2",
 ]
 
 [[package]]
@@ -139,6 +156,20 @@ dependencies = [
  "futures-core",
  "pin-project-lite",
  "tokio",
+ "tokio-util",
+]
+
+[[package]]
+name = "tokio-util"
+version = "0.7.17"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2efa149fe76073d6e8fd97ef4f4eca7b67f599660115591483572e406e165594"
+dependencies = [
+ "bytes",
+ "futures-core",
+ "futures-sink",
+ "pin-project-lite",
+ "tokio",
 ]
 
 [[package]]
@@ -159,6 +190,15 @@ version = "0.2.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "f0805222e57f7521d6a62e36fa9163bc891acd422f971defe97d64e70d0a4fe5"
 
+[[package]]
+name = "windows-sys"
+version = "0.60.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f2f500e4d28234f72040990ec9d39e3a6b950f9f22d3dba18416c35882612bcb"
+dependencies = [
+ "windows-targets",
+]
+
 [[package]]
 name = "windows-sys"
 version = "0.61.2"
@@ -167,3 +207,68 @@ checksum = "ae137229bcbd6cdf0f7b80a31df61766145077ddf49416a728b02cb3921ff3fc"
 dependencies = [
  "windows-link",
 ]
+
+[[package]]
+name = "windows-targets"
+version = "0.53.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4945f9f551b88e0d65f3db0bc25c33b8acea4d9e41163edf90dcd0b19f9069f3"
+dependencies = [
+ "windows-link",
+ "windows_aarch64_gnullvm",
+ "windows_aarch64_msvc",
+ "windows_i686_gnu",
+ "windows_i686_gnullvm",
+ "windows_i686_msvc",
+ "windows_x86_64_gnu",
+ "windows_x86_64_gnullvm",
+ "windows_x86_64_msvc",
+]
+
+[[package]]
+name = "windows_aarch64_gnullvm"
+version = "0.53.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a9d8416fa8b42f5c947f8482c43e7d89e73a173cead56d044f6a56104a6d1b53"
+
+[[package]]
+name = "windows_aarch64_msvc"
+version = "0.53.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b9d782e804c2f632e395708e99a94275910eb9100b2114651e04744e9b125006"
+
+[[package]]
+name = "windows_i686_gnu"
+version = "0.53.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "960e6da069d81e09becb0ca57a65220ddff016ff2d6af6a223cf372a506593a3"
+
+[[package]]
+name = "windows_i686_gnullvm"
+version = "0.53.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "fa7359d10048f68ab8b09fa71c3daccfb0e9b559aed648a8f95469c27057180c"
+
+[[package]]
+name = "windows_i686_msvc"
+version = "0.53.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1e7ac75179f18232fe9c285163565a57ef8d3c89254a30685b57d83a38d326c2"
+
+[[package]]
+name = "windows_x86_64_gnu"
+version = "0.53.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9c3842cdd74a865a8066ab39c8a7a473c0778a3f29370b5fd6b4b9aa7df4a499"
+
+[[package]]
+name = "windows_x86_64_gnullvm"
+version = "0.53.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0ffa179e2d07eee8ad8f57493436566c7cc30ac536a3379fdf008f47f6bb7ae1"
+
+[[package]]
+name = "windows_x86_64_msvc"
+version = "0.53.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d6bbff5f0aada427a1e5a6da5f1f98158182f26556f345ac9e04d36d0ebed650"

Cargo.toml

@@ -13,4 +13,4 @@ publish = false
 [dependencies]
 thiserror = "2.0.17"
 tokio = { version = "1.48.0", features = ["process", "rt-multi-thread", "macros"] }
-tokio-stream = "0.1.17"
+tokio-stream = { version = "0.1.17", features = ["full", "io-util", "signal", "tokio-util"] }

src/config.rs

@@ -30,6 +30,18 @@ pub enum MinecraftVersion {
     Snapshot(Snapshot),
 }
 
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum StreamType {
+    Stdout,
+    Stderr,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct StreamLine {
+    source: StreamType,
+    val: String,
+}
+
 impl Display for Version {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         write!(f, "{}.{}.{}", self.major, self.minor, self.patch)

src/error.rs

@@ -57,4 +57,10 @@ pub enum HandleError {
     InvalidPathJAR(String),
 }
 
+#[derive(Debug, Clone, Error)]
+pub enum SubscribeError {
+    #[error("No stdout found")]
+    NoStdout,
+}
+
 type Result<T> = std::result::Result<T, Error>;

src/instance.rs

@@ -1,8 +1,11 @@
-use std::path::PathBuf;
+use std::{path::PathBuf, sync::Arc};
+
+use tokio::sync::broadcast;
+use tokio_stream::wrappers::BroadcastStream;
 
 use crate::{
-    config::{MinecraftType, MinecraftVersion},
-    error::HandleError,
+    config::{MinecraftType, MinecraftVersion, StreamLine, StreamType},
+    error::{HandleError, SubscribeError},
 };
 
 #[derive(Debug, Clone)]
@@ -26,6 +29,8 @@ pub enum InstanceStatus {
 pub struct InstanceHandle {
     pub data: InstanceData,
     pub status: InstanceStatus,
+    stdout_tx: Option<broadcast::Sender<StreamLine>>,
+    stderr_tx: Option<broadcast::Sender<StreamLine>>,
 }
 
 impl InstanceHandle {
@@ -58,6 +63,33 @@ impl InstanceHandle {
         };
 
         let status = InstanceStatus::Stopped;
-        Ok(Self { data, status })
+        Ok(Self {
+            data,
+            status,
+            stdout_tx: None,
+            stderr_tx: None,
+        })
+    }
+
+    pub fn subscribe(
+        &self,
+        stream: StreamType,
+    ) -> Result<BroadcastStream<StreamLine>, SubscribeError> {
+        match stream {
+            StreamType::Stdout => {
+                let rx = match &self.stdout_tx {
+                    Some(value) => value.subscribe(),
+                    None => return Err(SubscribeError::NoStdout),
+                };
+                Ok(BroadcastStream::new(rx))
+            }
+            StreamType::Stderr => {
+                let rx = match &self.stderr_tx {
+                    Some(value) => value.subscribe(),
+                    None => return Err(SubscribeError::NoStdout),
+                };
+                Ok(BroadcastStream::new(rx))
+            }
+        }
     }
 }