use super::subscribe_loop::{ChannelRx, ControlCommand, ControlTx, ListenerType, SubscriptionID};
use crate::data::message::Message;
use crate::runtime::Runtime;
use futures_channel::mpsc;
use futures_util::sink::SinkExt;
use futures_util::stream::Stream;
use futures_util::task::{Context, Poll};
use log::debug;
use std::pin::Pin;
#[derive(Debug)]
pub struct Subscription<TRuntime: Runtime> {
pub(crate) runtime: TRuntime, pub(crate) name: ListenerType, pub(crate) id: SubscriptionID, pub(crate) control_tx: ControlTx, pub(crate) channel_rx: ChannelRx, }
impl<TRuntime: Runtime> Stream for Subscription<TRuntime> {
type Item = Message;
fn poll_next(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
Stream::poll_next(Pin::new(&mut self.get_mut().channel_rx), cx)
}
fn size_hint(&self) -> (usize, Option<usize>) {
Stream::size_hint(&self.channel_rx)
}
}
impl<TRuntime: Runtime> Subscription<TRuntime> {
fn drop_command(&self) -> ControlCommand {
ControlCommand::Drop(self.id, self.name.clone())
}
}
impl<TRuntime: Runtime> Drop for Subscription<TRuntime> {
fn drop(&mut self) {
debug!("Dropping Subscription: {:?}", self.name);
let command = self.drop_command();
let mut control_tx = self.control_tx.clone();
self.runtime.spawn(async move {
let drop_send_result = control_tx.send(command).await;
if is_drop_send_result_error(drop_send_result) {
panic!("Unable to unsubscribe");
}
});
}
}
fn is_drop_send_result_error(result: Result<(), mpsc::SendError>) -> bool {
match result {
Ok(_) => false, Err(err @ mpsc::SendError { .. }) if err.is_disconnected() => {
false
}
_ => true,
}
}