use std::path::Path;
use futures_util::stream::{Stream, StreamExt};
use crate::internal;
use crate::message::Message;
use crate::error::Error;
pub struct BliveClient {
room_id: u64,
cookie_source: internal::cookie::CookieSource,
task_handle: Option<tokio::task::JoinHandle<()>>,
}
impl Drop for BliveClient {
fn drop(&mut self) {
if let Some(handle) = self.task_handle.take() {
handle.abort();
}
}
}
impl BliveClient {
pub fn new(room_id: u64, cookie_path: impl AsRef<Path>) -> Result<Self, Error> {
Ok(Self {
room_id,
cookie_source: internal::cookie::CookieSource::from(cookie_path.as_ref()),
task_handle: None,
})
}
pub fn from_cookie_string(room_id: u64, cookie: String) -> Result<Self, Error> {
Ok(Self {
room_id,
cookie_source: internal::cookie::CookieSource::Inline(cookie),
task_handle: None,
})
}
pub async fn stream(&mut self) -> Result<impl Stream<Item = Message>, Error> {
let (rx, handle) = internal::ws::start_connection(self.room_id, &self.cookie_source).await?;
self.task_handle = Some(handle);
Ok(tokio_stream::wrappers::UnboundedReceiverStream::new(rx))
}
pub async fn on_message<F>(&mut self, mut callback: F) -> Result<(), Error>
where
F: FnMut(Message) + Send + 'static,
{
let mut stream = self.stream().await?;
while let Some(message) = stream.next().await {
callback(message);
}
Ok(())
}
pub async fn on_danmu<F>(&mut self, mut callback: F) -> Result<(), Error>
where
F: FnMut(crate::message::Danmu) + Send + 'static,
{
self.on_message(move |msg| {
if let Message::Danmu(danmu) = msg {
callback(danmu);
}
}).await
}
pub async fn on_gift<F>(&mut self, mut callback: F) -> Result<(), Error>
where
F: FnMut(crate::message::Gift) + Send + 'static,
{
self.on_message(move |msg| {
if let Message::Gift(gift) = msg {
callback(gift);
}
}).await
}
pub async fn on_super_chat<F>(&mut self, mut callback: F) -> Result<(), Error>
where
F: FnMut(crate::message::SuperChat) + Send + 'static,
{
self.on_message(move |msg| {
if let Message::SuperChat(sc) = msg {
callback(sc);
}
}).await
}
}