use std::cell::RefCell;
use std::io::{Read, Write};
use cast::cast_channel;
use utils;
use errors::Error;
#[derive(Debug, Clone)]
pub enum CastMessagePayload {
String(String),
Binary(Vec<u8>),
}
#[derive(Debug, Clone)]
pub struct CastMessage {
pub namespace: String,
pub source: String,
pub destination: String,
pub payload: CastMessagePayload,
}
pub struct MessageManager<S>
where
S: Write + Read,
{
message_buffer: RefCell<Vec<CastMessage>>,
stream: RefCell<S>,
request_counter: RefCell<i32>,
}
impl<S> MessageManager<S>
where
S: Write + Read,
{
pub fn new(stream: S) -> Self {
MessageManager {
stream: RefCell::new(stream),
message_buffer: RefCell::new(vec![]),
request_counter: RefCell::new(1),
}
}
pub fn send(&self, message: CastMessage) -> Result<(), Error> {
let mut raw_message = cast_channel::CastMessage::new();
raw_message.set_protocol_version(cast_channel::CastMessage_ProtocolVersion::CASTV2_1_0);
raw_message.set_namespace(message.namespace);
raw_message.set_source_id(message.source);
raw_message.set_destination_id(message.destination);
match message.payload {
CastMessagePayload::String(payload) => {
raw_message.set_payload_type(cast_channel::CastMessage_PayloadType::STRING);
raw_message.set_payload_utf8(payload);
}
CastMessagePayload::Binary(payload) => {
raw_message.set_payload_type(cast_channel::CastMessage_PayloadType::BINARY);
raw_message.set_payload_binary(payload);
}
};
let message_content_buffer = utils::to_vec(&raw_message)?;
let message_length_buffer =
utils::write_u32_to_buffer(message_content_buffer.len() as u32)?;
let writer = &mut *self.stream.borrow_mut();
writer.write_all(&message_length_buffer)?;
writer.write_all(&message_content_buffer)?;
debug!("Message sent: {:?}", raw_message);
Ok(())
}
pub fn receive(&self) -> Result<CastMessage, Error> {
let mut message_buffer = self.message_buffer.borrow_mut();
if message_buffer.is_empty() {
self.read()
} else {
Ok(message_buffer.remove(0))
}
}
pub fn receive_find_map<F, B>(&self, f: F) -> Result<B, Error>
where
F: Fn(&CastMessage) -> Result<Option<B>, Error>,
{
loop {
let message = self.read()?;
match f(&message)? {
Some(r) => return Ok(r),
None => self.message_buffer.borrow_mut().push(message),
}
}
}
pub fn generate_request_id(&self) -> i32 {
let request_id = *self.request_counter.borrow() + 1;
*self.request_counter.borrow_mut() = request_id;
request_id
}
fn read(&self) -> Result<CastMessage, Error> {
let mut buffer: [u8; 4] = [0; 4];
let reader = &mut *self.stream.borrow_mut();
reader.read_exact(&mut buffer)?;
let length = utils::read_u32_from_buffer(&buffer)?;
let mut buffer: Vec<u8> = Vec::with_capacity(length as usize);
let mut limited_reader = reader.take(u64::from(length));
limited_reader.read_to_end(&mut buffer)?;
let raw_message = utils::from_vec::<cast_channel::CastMessage>(buffer.to_vec())?;
debug!("Message received: {:?}", raw_message);
Ok(CastMessage {
namespace: raw_message.get_namespace().to_string(),
source: raw_message.get_source_id().to_string(),
destination: raw_message.get_destination_id().to_string(),
payload: match raw_message.get_payload_type() {
cast_channel::CastMessage_PayloadType::STRING => {
CastMessagePayload::String(raw_message.get_payload_utf8().to_string())
}
cast_channel::CastMessage_PayloadType::BINARY => {
CastMessagePayload::Binary(raw_message.get_payload_binary().to_owned())
}
},
})
}
}