Repository: sxyazi/yazi
File: yazi-dds/src/client.rs (Rust)

Code before:
					Some(payload) = rx.recv() => {
						if writer.write_all(payload.as_bytes()).await.is_err() {
							(lines, writer) = Self::reconnect(&mut server).await;
							writer.write_all(payload.as_bytes()).await.ok(); // Retry once
						}
					}
					Ok(next) = lines.next_line() => {
						let Some(line) = next else {
							(lines, writer) = Self::reconnect(&mut server).await;
							continue;
						};

						if line.is_empty() {
							continue;
						} else if line.starts_with("hey,") {
							Self::handle_hey(&line);
						} else {
							Payload::from_str(&line).map(|p| p.emit()).ok();
						}
					}
				}
			}
		});
	}

	pub fn echo_events_to_stdout(sender: DDSPeer, kinds: HashSet<String>) {
		let mut rx = QUEUE_RX.drop();
		while rx.try_recv().is_ok() {}

		tokio::spawn(async move {
			let mut server = None;
			let (mut lines, mut writer) = Self::connect(&mut server).await;

			loop {
				select! {
					Some(payload) = rx.recv() => {
						if writer.write_all(payload.as_bytes()).await.is_err() {
							(lines, writer) = Self::reconnect(&mut server).await;
							writer.write_all(payload.as_bytes()).await.ok(); // Retry once
						}
					}
					Ok(next) = lines.next_line() => {
						let Some(line) = next else {
							(lines, writer) = Self::reconnect(&mut server).await;
							continue;
						};

						if line.is_empty() {
							continue;
						}


Review comment (suggestion):
```suggestion
	pub fn echo_events_to_stdout(sender: Option<u64>, kinds: HashSet<String>) {
```

Filter by client ID when it's `Some(u64)`, otherwise receive all.

I think this would be simpler, no longer needing to create a new `DDSPeer` type, as there is already another type named `Peer` in the DDS domain.


Code after:
						if writer.write_all(payload.as_bytes()).await.is_err() {
							(lines, writer) = Self::reconnect(&mut server).await;
							writer.write_all(payload.as_bytes()).await.ok(); // Retry once
						}
					}
					Ok(next) = lines.next_line() => {
						let Some(line) = next else {
							(lines, writer) = Self::reconnect(&mut server).await;
							continue;
						};

						if line.is_empty() {
							continue;
						} else if line.starts_with("hey,") {
							Self::handle_hey(&line);
						} else {
							Payload::from_str(&line).map(|p| p.emit()).ok();
						}
					}
				}
			}
		});
	}

	/// Connect to an existing server to send a single message.
	pub async fn shot(kind: &str, receiver: u64, severity: Option<u16>, body: &str) -> Result<()> {
		Body::validate(kind)?;

		let sender = severity.map(Into::into).unwrap_or(*ID);
		let payload = format!(
			"{}\n{kind},{receiver},{sender},{body}\n{}\n",
			Payload::new(BodyHi::borrowed(Default::default())),
			Payload::new(BodyBye::owned())
		);

		let (mut lines, mut writer) = Stream::connect().await?;
		writer.write_all(payload.as_bytes()).await?;
		writer.flush().await?;
		drop(writer);

		let mut version = None;
		while let Ok(Some(line)) = lines.next_line().await {
			match line.split(',').next() {
				Some("hey") if version.is_none() => {
					if let Ok(Body::Hey(hey)) = Payload::from_str(&line).map(|p| p.body) {
						version = Some(hey.version);
					}
				}
				Some("bye") => break,
				_ => {}
			}