blob: 960fb079d360841537dc2ad15c8f65d01cc4bc01 [file]
use crate::binary::command::{BinaryServerCommand, ServerCommand, ServerCommandHandler};
use crate::binary::handlers::utils::receive_and_validate;
use crate::binary::mapper;
use crate::binary::sender::SenderKind;
use crate::streaming::session::Session;
use crate::streaming::systems::system::SharedSystem;
use anyhow::Result;
use iggy::error::IggyError;
use iggy::streams::get_stream::GetStream;
use tracing::debug;
impl ServerCommandHandler for GetStream {
fn code(&self) -> u32 {
iggy::command::GET_STREAM_CODE
}
async fn handle(
self,
sender: &mut SenderKind,
_length: u32,
session: &Session,
system: &SharedSystem,
) -> Result<(), IggyError> {
debug!("session: {session}, command: {self}");
let system = system.read().await;
let Ok(stream) = system.try_find_stream(session, &self.stream_id) else {
sender.send_empty_ok_response().await?;
return Ok(());
};
let Some(stream) = stream else {
sender.send_empty_ok_response().await?;
return Ok(());
};
let response = mapper::map_stream(stream);
sender.send_ok_response(&response).await?;
Ok(())
}
}
impl BinaryServerCommand for GetStream {
async fn from_sender(
sender: &mut SenderKind,
code: u32,
length: u32,
) -> Result<Self, IggyError> {
match receive_and_validate(sender, code, length).await? {
ServerCommand::GetStream(get_stream) => Ok(get_stream),
_ => Err(IggyError::InvalidCommand),
}
}
}