use super::{ConnectionProcessState, ConnectionRuntime, Frame, FrameAction, OpenLink};
use crate::link::calls::{CallError, CallExecution, CallPacket, CallReply, CallRequest};
use std::collections::BTreeMap;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex, mpsc};
#[path = "executor.rs"]
mod executor;
#[cfg(test)]
#[path = "calls_tests.rs"]
mod tests;
struct Pending {
subject: String,
correlation: (u32, u64),
execution: Box<dyn CallExecution>,
}
struct Completion {
id: u64,
subject: String,
answer: Result<CallReply, CallError>,
}
pub(super) struct IncomingCalls {
last_id: u64,
pending: BTreeMap<u64, Pending>,
sender: mpsc::SyncSender<Completion>,
receiver: Mutex<mpsc::Receiver<Completion>>,
bound: usize,
failed: Arc<AtomicBool>,
}
impl std::fmt::Debug for IncomingCalls {
fn fmt(&self, out: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
out.debug_struct("IncomingCalls")
.field("pending", &self.pending.len())
.finish_non_exhaustive()
}
}
impl IncomingCalls {
pub(super) fn new(bound: usize) -> Self {
let (sender, receiver) = mpsc::sync_channel(bound);
Self {
last_id: 0,
pending: BTreeMap::new(),
sender,
receiver: Mutex::new(receiver),
bound,
failed: Arc::new(AtomicBool::new(false)),
}
}
pub(super) fn start(
&mut self,
pid: u64,
runtime: &ConnectionRuntime,
request: CallRequest,
) -> Result<(), CallError> {
if self.bound == 0 || self.pending.len() >= self.bound {
return Err(CallError::Capacity);
}
let id = request.id();
if id <= self.last_id {
return Err(CallError::Correlation);
}
let host = runtime.link_host().ok_or(CallError::Unavailable)?;
let peers = host.peers().ok_or(CallError::Unavailable)?;
let subject = request.subject().to_owned();
let correlation = request.correlation()?;
if !peers.service_prefixes.iter().any(|prefix| {
subject == *prefix
|| subject
.strip_prefix(prefix)
.is_some_and(|suffix| suffix.starts_with('.'))
}) {
return Err(CallError::NoServer);
}
let executor = host.call_executor().ok_or(CallError::Unavailable)?;
self.last_id = id;
let sender = self.sender.clone();
let ready = runtime.ready_waker(pid).ok_or(CallError::Unavailable)?;
let failed = Arc::clone(&self.failed);
let completed_subject = subject.clone();
let execution = executor.start(
request,
Box::new(move |answer| {
match sender.try_send(Completion {
id,
subject: completed_subject,
answer,
}) {
Ok(()) => {}
Err(mpsc::TrySendError::Disconnected(_)) => {
tracing::debug!("linked call owner departed before its answer");
}
Err(mpsc::TrySendError::Full(_)) => {
failed.store(true, Ordering::Release);
tracing::error!("link_call_completion_queue_refused");
}
}
if !ready.fire() {
tracing::debug!("linked call owner departed before its ready signal");
}
}),
)?;
self.pending.insert(
id,
Pending {
subject,
correlation,
execution,
},
);
Ok(())
}
pub(super) fn cancel(&mut self, id: u64) -> Result<(), CallError> {
if let Some(mut pending) = self.pending.remove(&id) {
pending.execution.finish()?;
}
Ok(())
}
pub(super) fn finish_all(&mut self) -> Result<(), CallError> {
let mut failure = None;
for mut pending in std::mem::take(&mut self.pending).into_values() {
if let Err(error) = pending.execution.finish() {
failure = Some(error);
}
}
failure.map_or(Ok(()), Err)
}
fn drain(&self) -> Result<Vec<Completion>, CallError> {
if self.failed.load(Ordering::Acquire) {
return Err(CallError::Unavailable);
}
let receiver = self.receiver.lock().map_err(|_| CallError::Unavailable)?;
let mut completed = Vec::new();
loop {
match receiver.try_recv() {
Ok(completion) => completed.push(completion),
Err(mpsc::TryRecvError::Empty) => break,
Err(mpsc::TryRecvError::Disconnected) => return Err(CallError::Unavailable),
}
}
Ok(completed)
}
}
fn answered(
pending: &mut Pending,
completion: Completion,
release: impl Fn(&str) -> Result<(), CallError>,
) -> Result<CallReply, CallError> {
if pending.subject != completion.subject {
return Err(CallError::Correlation);
}
if let Err(error) = pending.execution.finish() {
tracing::warn!(
call = completion.id,
subject = %pending.subject,
%error,
"a far call's execution could not be retired; its answer is released"
);
}
match release(&pending.subject).and(completion.answer) {
Ok(reply) => Ok(reply),
Err(error) => CallReply::new(
completion.id,
Frame::ConversationError {
flags: 0,
stream_id: pending.correlation.0,
conversation_id: pending.correlation.1,
reason_code: 0xFFFF,
message: Some(error.to_string()),
},
),
}
}
pub(super) fn dispatch(
pid: u64,
runtime: &ConnectionRuntime,
state: &ConnectionProcessState,
stream: u32,
subject: &str,
packet: CallPacket,
) -> FrameAction {
let result = (|| {
let mut contents = state.links.lock().map_err(|_| CallError::Unavailable)?;
let open = contents.links.get_mut(&stream).ok_or(CallError::PeerGone)?;
let result = match packet {
CallPacket::Request(request) if request.subject() == subject => {
open.calls.start(pid, runtime, request)
}
CallPacket::Cancel(id) => open.calls.cancel(id),
CallPacket::Topology(_) | CallPacket::Reply(_) | CallPacket::Request(_) => {
Err(CallError::Invalid)
}
};
drop(contents);
result
})();
match result {
Ok(()) => FrameAction::NoResponse,
Err(error) => FrameAction::Respond(Frame::PublishError {
flags: 0,
stream_id: stream,
reason_code: super::LINK_CROSSING_REFUSED_CODE,
message: Some(error.to_string()),
}),
}
}
impl Drop for IncomingCalls {
fn drop(&mut self) {
if let Err(error) = self.finish_all() {
tracing::error!(%error, "far call cancellation failed");
}
}
}
pub(super) fn settled(
open: &mut OpenLink,
stream_id: u32,
bound: usize,
release: impl Fn(&OpenLink, &str) -> Result<(), CallError>,
) -> Result<Vec<Frame>, CallError> {
let mut frames = Vec::new();
for completion in open.calls.drain()? {
let Some(mut pending) = open.calls.pending.remove(&completion.id) else {
continue;
};
let subject = completion.subject.clone();
let reply = answered(&mut pending, completion, |subject| release(open, subject))?;
let sealed = CallPacket::Reply(reply).seal(&mut open.transport, bound)?;
frames.push(Frame::LinkMessage {
flags: 0,
stream_id,
subject,
sealed,
});
}
Ok(frames)
}