use std::io;
use mio::{Evented, Poll, PollOpt, Ready, Token};
use futures::{
pin_mut,
stream::{FusedStream, Stream, StreamExt},
};
use rustdds::{
dds::{ReadError, ReadResult, WriteResult},
*,
};
use serde::{de::DeserializeOwned, Serialize};
use super::{gid::Gid, message_info::MessageInfo, node::Node};
pub struct Publisher<M: Serialize> {
datawriter: no_key::DataWriterCdr<M>,
}
impl<M: Serialize> Publisher<M> {
pub(crate) fn new(datawriter: no_key::DataWriterCdr<M>) -> Publisher<M> {
Publisher { datawriter }
}
pub fn publish(&self, message: M) -> WriteResult<(), M> {
self.datawriter.write(message, Some(Timestamp::now()))
}
pub fn assert_liveliness(&self) -> WriteResult<(), ()> {
self.datawriter.assert_liveliness()
}
pub fn guid(&self) -> rustdds::GUID {
self.datawriter.guid()
}
pub fn gid(&self) -> Gid {
self.guid().into()
}
pub fn get_subscription_count(&self, my_node: &Node) -> usize {
my_node.get_subscription_count(self.guid())
}
pub async fn wait_for_subscription(&self, my_node: &Node) {
my_node.wait_for_reader(self.guid()).await
}
pub async fn async_publish(&self, message: M) -> WriteResult<(), M> {
self
.datawriter
.async_write(message, Some(Timestamp::now()))
.await
}
#[allow(dead_code)] pub(crate) async fn async_publish_with_options(
&self,
message: M,
wo: WriteOptions,
) -> dds::WriteResult<rustdds::rpc::SampleIdentity, M> {
self.datawriter.async_write_with_options(message, wo).await
}
}
pub struct Subscription<M: DeserializeOwned> {
datareader: no_key::SimpleDataReaderCdr<M>,
}
impl<M: 'static + DeserializeOwned> Subscription<M> {
pub(crate) fn new(datareader: no_key::SimpleDataReaderCdr<M>) -> Subscription<M> {
Subscription { datareader }
}
pub fn take(&self) -> ReadResult<Option<(M, MessageInfo)>> {
self.datareader.drain_read_notifications();
let ds: Option<no_key::DeserializedCacheChange<M>> = self.datareader.try_take_one()?;
Ok(ds.map(dcc_to_value_and_messageinfo))
}
pub async fn async_take(&self) -> ReadResult<(M, MessageInfo)> {
let async_stream = self.datareader.as_async_stream();
pin_mut!(async_stream);
match async_stream.next().await {
Some(Err(e)) => Err(e),
Some(Ok(ds)) => Ok(dcc_to_value_and_messageinfo(ds)),
None => {
read_error_internal!("async_take(): SimpleDataReader value stream unexpectedly ended!")
}
}
}
pub fn async_stream(
&self,
) -> impl Stream<Item = ReadResult<(M, MessageInfo)>> + FusedStream + '_ {
self
.datareader
.as_async_stream()
.map(|result| result.map(dcc_to_value_and_messageinfo))
}
pub fn guid(&self) -> rustdds::GUID {
self.datareader.guid()
}
pub fn gid(&self) -> Gid {
self.guid().into()
}
pub fn get_publisher_count(&self, my_node: &Node) -> usize {
my_node.get_publisher_count(self.guid())
}
pub async fn wait_for_publisher(&self, my_node: &Node) {
my_node.wait_for_writer(self.guid()).await
}
}
#[inline]
fn dcc_to_value_and_messageinfo<M>(dcc: no_key::DeserializedCacheChange<M>) -> (M, MessageInfo)
where
M: DeserializeOwned,
{
let mi = MessageInfo::from(&dcc);
(dcc.into_value(), mi)
}
impl<D> Evented for Subscription<D>
where
D: DeserializeOwned,
{
fn register(&self, poll: &Poll, token: Token, interest: Ready, opts: PollOpt) -> io::Result<()> {
self.datareader.register(poll, token, interest, opts)
}
fn reregister(
&self,
poll: &Poll,
token: Token,
interest: Ready,
opts: PollOpt,
) -> io::Result<()> {
self.datareader.reregister(poll, token, interest, opts)
}
fn deregister(&self, poll: &Poll) -> io::Result<()> {
self.datareader.deregister(poll)
}
}