use std::collections::VecDeque;
use std::marker::PhantomData;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use std::time::Duration;
use arc_swap::ArcSwapOption;
use tokio::sync::Notify;
use tokio::task::JoinHandle;
use zenoh::bytes::Encoding;
use zenoh::key_expr::OwnedKeyExpr;
use zenoh::sample::Sample;
use crate::LogicalTime;
use crate::abi::{CodecId, encoding_string, parse_encoding_string};
use crate::codec::{Codec, MessagePack};
use crate::contract::ContractBody;
use crate::error::{BusError, Result};
use crate::metadata::{BusMetadata, Source};
use crate::query::{QueryError, QueryFailure};
use crate::runtime_metrics::RuntimeMetricHandle;
use crate::session::Bus;
use crate::session::OUTBOUND_CAPACITY;
use crate::topic::{AskQuery, Publish, Subscribe, Topic};
pub const DEFAULT_QUERY_TIMEOUT: Duration = Duration::from_secs(5);
pub struct Publisher<B> {
bus: Bus,
key: String,
metric: RuntimeMetricHandle,
_body: PhantomData<fn() -> B>,
}
impl<B> Clone for Publisher<B> {
fn clone(&self) -> Self {
Publisher {
bus: self.bus.clone(),
key: self.key.clone(),
metric: self.metric.clone(),
_body: PhantomData,
}
}
}
impl<B: ContractBody> Publisher<B> {
#[doc(hidden)]
pub fn new(bus: Bus, topic: &Topic<Publish<B>>) -> Result<Self> {
let topic_key = topic.publish_key()?;
let metric = bus
.runtime_metrics()
.register_outbound(topic_key, OUTBOUND_CAPACITY);
let key = bus.full_key(topic_key);
Ok(Publisher {
bus,
key,
metric,
_body: PhantomData,
})
}
#[allow(clippy::unused_async)]
pub async fn publish_at(&self, at: LogicalTime, body: B) -> Result<()> {
self.try_publish(at, body)
}
pub fn try_publish(&self, at: LogicalTime, body: B) -> Result<()> {
let payload = MessagePack::encode(&body)?;
let metadata = BusMetadata {
codec: MessagePack::ID.as_u8(),
produced_at_ns: at.time_ns(),
epoch: at.epoch(),
source: Source {
participant: self.bus.participant().to_string(),
incarnation: self.bus.incarnation(),
sequence: self.bus.next_sequence(),
},
};
let encoding = encoding_string(MessagePack::ID);
self.bus.enqueue(
self.key.clone(),
encoding,
metadata.encode(),
payload,
self.metric.clone(),
)
}
}
pub struct Querier<Req, Resp> {
bus: Bus,
key: String,
timeout: Duration,
_p: PhantomData<fn() -> (Req, Resp)>,
}
impl<Req, Resp> Clone for Querier<Req, Resp> {
fn clone(&self) -> Self {
Querier {
bus: self.bus.clone(),
key: self.key.clone(),
timeout: self.timeout,
_p: PhantomData,
}
}
}
impl<Req, Resp> Querier<Req, Resp>
where
Req: ContractBody,
Resp: ContractBody,
{
#[doc(hidden)]
pub fn new(bus: Bus, topic: &Topic<AskQuery<Req, Resp>>, timeout: Duration) -> Result<Self> {
let key = bus.full_key(topic.publish_key()?);
Ok(Querier {
bus,
key,
timeout,
_p: PhantomData,
})
}
pub async fn query(&self, request: Req) -> std::result::Result<Resp, QueryError> {
let payload =
MessagePack::encode(&request).map_err(|e| QueryError::Protocol(e.to_string()))?;
let metadata = BusMetadata {
codec: MessagePack::ID.as_u8(),
produced_at_ns: 0,
epoch: 0,
source: Source {
participant: self.bus.participant().to_string(),
incarnation: self.bus.incarnation(),
sequence: self.bus.next_sequence(),
},
};
let key = OwnedKeyExpr::new(self.key.clone())
.map_err(|e| QueryError::Protocol(format!("invalid query key '{}': {e}", self.key)))?;
let replies = self
.bus
.session()
.get(key)
.payload(payload)
.encoding(Encoding::from(encoding_string(MessagePack::ID)))
.attachment(metadata.encode())
.target(zenoh::query::QueryTarget::All)
.consolidation(zenoh::query::ConsolidationMode::None)
.await
.map_err(|e| QueryError::Protocol(e.to_string()))?;
let deadline = tokio::time::Instant::now() + self.timeout;
let mut outcome: Option<std::result::Result<Resp, QueryError>> = None;
loop {
match tokio::time::timeout_at(deadline, replies.recv_async()).await {
Ok(Ok(reply)) => {
if outcome.is_some() {
return Err(QueryError::TooManyResponders);
}
outcome = Some(decode_reply_result::<Resp>(reply.into_result()));
}
Ok(Err(_)) => break, Err(_elapsed) => {
return outcome.unwrap_or_else(|| {
Err(QueryError::Timeout(QueryFailure::deadline_exceeded(
"query deadline exceeded",
)))
});
}
}
}
outcome.unwrap_or(Err(QueryError::Unavailable))
}
}
fn decode_reply_result<Resp: ContractBody>(
result: std::result::Result<Sample, zenoh::query::ReplyError>,
) -> std::result::Result<Resp, QueryError> {
match result {
Ok(sample) => decode_reply::<Resp>(&sample),
Err(reply_error) => {
let bytes = reply_error.payload().to_bytes();
match crate::query::QueryFailure::decode(bytes.as_ref()) {
Ok(failure) => Err(QueryError::Server(failure)),
Err(e) => Err(QueryError::Protocol(format!("malformed error reply: {e}"))),
}
}
}
}
fn decode_reply<Resp: ContractBody>(sample: &Sample) -> std::result::Result<Resp, QueryError> {
match decode_sample::<Resp>(sample, Resp::TOPIC) {
Ok((body, _)) => Ok(body),
Err(e) => Err(QueryError::Decode(e.to_string())),
}
}
#[derive(Clone, Debug)]
pub struct Received<B> {
pub body: B,
pub metadata: BusMetadata,
}
pub struct Latest<B> {
slot: Arc<ArcSwapOption<B>>,
_guard: Arc<SubscriptionGuard>,
}
impl<B> Clone for Latest<B> {
fn clone(&self) -> Self {
Latest {
slot: Arc::clone(&self.slot),
_guard: Arc::clone(&self._guard),
}
}
}
impl<B: ContractBody> Latest<B> {
#[doc(hidden)]
pub async fn new(bus: &Bus, topic: &Topic<Subscribe<B>>) -> Result<Self> {
let slot: Arc<ArcSwapOption<B>> = Arc::new(ArcSwapOption::from(None));
let store = Arc::clone(&slot);
let metric = bus.runtime_metrics().register_latest(topic.key());
let observe = metric.clone();
let guard = spawn_subscription::<B, _>(
bus,
topic.key(),
move |body, _meta| {
let overwrote = store.swap(Some(Arc::new(body))).is_some();
observe.record_latest(overwrote);
},
metric,
)
.await?;
Ok(Latest {
slot,
_guard: Arc::new(guard),
})
}
pub fn latest(&self) -> Option<B> {
self.slot.load_full().map(|arc| (*arc).clone())
}
}
pub struct Subscriber<B> {
ring: Arc<Ring<B>>,
_guard: Arc<SubscriptionGuard>,
}
impl<B> Clone for Subscriber<B> {
fn clone(&self) -> Self {
Subscriber {
ring: Arc::clone(&self.ring),
_guard: Arc::clone(&self._guard),
}
}
}
impl<B: ContractBody> Subscriber<B> {
#[doc(hidden)]
pub async fn new(bus: &Bus, topic: &Topic<Subscribe<B>>, depth: usize) -> Result<Self> {
let depth = depth.max(1);
let metric = bus
.runtime_metrics()
.register_subscriber(topic.key(), depth);
let ring = Arc::new(Ring::new(depth, metric.clone()));
let push = Arc::clone(&ring);
let drops = bus.clone();
let guard = spawn_subscription::<B, _>(
bus,
topic.key(),
move |body, metadata| {
let (evicted, _current_depth) = push.push(Received { body, metadata });
if evicted {
drops.health().inbound_drops.fetch_add(1, Ordering::Relaxed);
}
},
metric.clone(),
)
.await?;
Ok(Subscriber {
ring,
_guard: Arc::new(guard),
})
}
pub async fn recv(&self) -> Result<Received<B>> {
let (received, _current_depth) = self.ring.recv().await;
Ok(received)
}
pub fn try_recv(&self) -> Option<Received<B>> {
self.ring
.try_pop()
.map(|(received, _current_depth)| received)
}
pub fn dropped(&self) -> u64 {
self.ring.dropped.load(Ordering::Relaxed)
}
}
struct Ring<B> {
buf: Mutex<VecDeque<Received<B>>>,
notify: Notify,
cap: usize,
dropped: AtomicU64,
metric: RuntimeMetricHandle,
}
impl<B> Ring<B> {
fn new(cap: usize, metric: RuntimeMetricHandle) -> Self {
Ring {
buf: Mutex::new(VecDeque::with_capacity(cap)),
notify: Notify::new(),
cap,
dropped: AtomicU64::new(0),
metric,
}
}
fn push(&self, item: Received<B>) -> (bool, usize) {
let mut dropped = false;
let mut buf = self.buf.lock().expect("ring mutex poisoned");
if buf.len() == self.cap {
buf.pop_front();
dropped = true;
self.dropped.fetch_add(1, Ordering::Relaxed);
}
buf.push_back(item);
let depth = buf.len();
self.metric.record_subscriber(dropped, depth);
drop(buf);
self.notify.notify_one();
(dropped, depth)
}
fn try_pop(&self) -> Option<(Received<B>, usize)> {
let mut buf = self.buf.lock().expect("ring mutex poisoned");
let item = buf.pop_front()?;
let depth = buf.len();
self.metric.record_subscriber_pop(depth);
Some((item, depth))
}
async fn recv(&self) -> (Received<B>, usize) {
loop {
let notified = self.notify.notified();
if let Some(item) = self.try_pop() {
return item;
}
notified.await;
}
}
}
struct SubscriptionGuard {
task: JoinHandle<()>,
}
impl Drop for SubscriptionGuard {
fn drop(&mut self) {
self.task.abort();
}
}
async fn spawn_subscription<B, F>(
bus: &Bus,
topic_key: &str,
mut on_sample: F,
metric: RuntimeMetricHandle,
) -> Result<SubscriptionGuard>
where
B: ContractBody,
F: FnMut(B, BusMetadata) + Send + 'static,
{
let full_key = bus.full_key(topic_key);
let key_expr = OwnedKeyExpr::new(full_key.clone())
.map_err(|e| BusError::Namespace(format!("invalid subscribe key '{full_key}': {e}")))?;
let subscriber = bus
.session()
.declare_subscriber(key_expr)
.await
.map_err(|e| BusError::Transport(e.to_string()))?;
let topic_owned = topic_key.to_string();
let health_bus = bus.clone();
let task = tokio::spawn(async move {
while let Ok(sample) = subscriber.recv_async().await {
match decode_sample::<B>(&sample, &topic_owned) {
Ok((body, metadata)) => on_sample(body, metadata),
Err(err) => {
metric.record_decode_error();
health_bus
.health()
.decode_errors
.fetch_add(1, Ordering::Relaxed);
tracing::warn!(target: "phoxal.bus", topic = %topic_owned, error = %err, "dropped inbound sample");
}
}
}
});
Ok(SubscriptionGuard { task })
}
pub(crate) fn decode_sample<B: ContractBody>(
sample: &Sample,
topic: &str,
) -> Result<(B, BusMetadata)> {
let encoding =
parse_encoding_string(&sample.encoding().to_string()).map_err(|e| BusError::Metadata {
topic: topic.to_string(),
detail: format!("malformed encoding string: {e}"),
})?;
match encoding.codec_id() {
Some(CodecId::MessagePack) => {}
None => {
return Err(BusError::UnsupportedCodec(
encoding.codec,
topic.to_string(),
));
}
}
let attachment = sample.attachment().ok_or_else(|| BusError::Metadata {
topic: topic.to_string(),
detail: "missing BusMetadata attachment".to_string(),
})?;
let metadata =
BusMetadata::decode(attachment.to_bytes().as_ref()).map_err(|e| BusError::Metadata {
topic: topic.to_string(),
detail: format!("malformed BusMetadata: {e}"),
})?;
if metadata.codec != encoding.codec {
return Err(BusError::Metadata {
topic: topic.to_string(),
detail: format!(
"encoding/BusMetadata codec mismatch: encoding codec={}, metadata codec={}",
encoding.codec, metadata.codec
),
});
}
match metadata.codec_id() {
Some(CodecId::MessagePack) => {}
None => {
return Err(BusError::UnsupportedCodec(
metadata.codec,
topic.to_string(),
));
}
}
let body = MessagePack::decode::<B>(sample.payload().to_bytes().as_ref())?;
Ok((body, metadata))
}
#[cfg(test)]
mod subscriber_ring_tests {
use super::*;
fn received(body: u8) -> Received<u8> {
Received {
body,
metadata: BusMetadata {
codec: CodecId::MessagePack.as_u8(),
produced_at_ns: 0,
epoch: 0,
source: Source {
participant: "test".to_string(),
incarnation: 0,
sequence: u64::from(body),
},
},
}
}
#[test]
fn ring_counts_each_drop_oldest_eviction_cumulatively() {
let metrics = crate::runtime_metrics::RuntimeMetrics::default();
let metric = metrics.register_subscriber("v0.1/test/state", 1);
let ring = Ring::new(1, metric);
assert_eq!(ring.push(received(1)), (false, 1));
assert_eq!(ring.push(received(2)), (true, 1));
assert_eq!(ring.push(received(3)), (true, 1));
assert_eq!(ring.dropped.load(Ordering::Relaxed), 2);
let (received, depth) = ring.try_pop().unwrap();
assert_eq!(received.body, 3);
assert_eq!(depth, 0);
let row = metrics.take().pop().unwrap();
assert_eq!(row.count, 3);
assert_eq!(row.drops, 2);
assert_eq!(row.bounded_evictions, 2);
assert_eq!(row.current_depth, 0);
assert_eq!(row.high_water_depth, 1);
}
}