use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use std::time::Duration;
use zerodds_ast::{Ast, ExceptionNode, TypeNode};
use zerodds_ast_core::Id;
use zerodds_dcps::participant::DomainParticipant;
use zerodds_lang_rules::adapter::interface::{
self, InterfaceError, Operation, Raise, Service, all_operations_flattened,
exception_member_name, in_struct_name, operation_hash, out_struct_name, services,
};
use zerodds_rpc::codegen::{
DUMMY_MEMBER, RESULT_MEMBER, RETCODE_OK, RETURN_MEMBER, UNKNOWN_OP_MEMBER,
};
use zerodds_rpc::endpoint_options::RpcEndpointOptions;
use zerodds_rpc::qos_profile::RpcQos;
use zerodds_rpc::requester::Requester;
use zerodds_rpc::topic_naming::{AnnotatedTopics, BindingStyle};
use zerodds_types::builder::{Extensibility, TypeObjectBuilder};
use zerodds_types::dynamic::DynamicData;
use zerodds_types::{CompleteTypeObject, PrimitiveKind, TypeIdentifier};
use crate::dynamic_rpc::{self, DynamicRpcValue};
use crate::dynxml;
use crate::types_registry::{ResolvedType, TypeRegistry};
#[derive(Debug)]
pub enum RpcDispatchError {
Model(String),
UnknownOperation(String),
UnsupportedParameterType(String),
BadRequest(String),
Rpc(String),
RemoteException(u32),
OperationException {
operation: String,
exception: String,
fields_xml: String,
},
Timeout,
}
impl std::fmt::Display for RpcDispatchError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::Model(m) => write!(f, "interface model: {m}"),
Self::UnknownOperation(op) => {
write!(f, "no configured interface declares operation `{op}`")
}
Self::UnsupportedParameterType(m) => write!(f, "unsupported parameter type: {m}"),
Self::BadRequest(m) => write!(f, "bad request: {m}"),
Self::Rpc(m) => write!(f, "rpc: {m}"),
Self::RemoteException(code) => write!(f, "remote exception, code {code}"),
Self::OperationException {
operation,
exception,
fields_xml,
} => write!(
f,
"operation `{operation}` raised `{exception}`: {fields_xml}"
),
Self::Timeout => f.write_str("timed out waiting for a reply"),
}
}
}
impl std::error::Error for RpcDispatchError {}
const RPC_TIMEOUT: Duration = Duration::from_secs(10);
struct OpModel {
in_resolved: ResolvedType,
#[cfg_attr(not(test), allow(dead_code))]
result_resolved: ResolvedType,
out_resolved: ResolvedType,
exceptions: Vec<(i32, String, String, ResolvedType)>,
}
struct InterfaceModel {
call_resolved: ResolvedType,
return_resolved: ResolvedType,
operations: HashMap<i32, OpModel>,
}
pub struct RpcDispatcher {
ast: Arc<Ast>,
services: Vec<Service>,
participant: DomainParticipant,
types: Arc<TypeRegistry>,
models: Mutex<HashMap<String, Arc<InterfaceModel>>>,
requesters: Mutex<HashMap<String, Arc<Requester<DynamicRpcValue, DynamicRpcValue>>>>,
}
impl RpcDispatcher {
pub fn new(ast: Arc<Ast>, participant: DomainParticipant) -> Result<Self, RpcDispatchError> {
let services = services(&ast).map_err(interface_error)?;
Ok(Self {
ast,
services,
participant,
types: Arc::new(TypeRegistry::new()),
models: Mutex::new(HashMap::new()),
requesters: Mutex::new(HashMap::new()),
})
}
fn find_operation(&self, op_name: &str) -> Result<(&Service, Operation), RpcDispatchError> {
for service in &self.services {
let flattened =
all_operations_flattened(&self.ast, service.id).map_err(interface_error)?;
if let Some(op) = flattened.into_iter().find(|op| op.name == op_name) {
return Ok((service, op));
}
}
Err(RpcDispatchError::UnknownOperation(op_name.to_string()))
}
#[must_use]
pub fn knows_operation(&self, op_name: &str) -> bool {
self.find_operation(op_name).is_ok()
}
fn model_for(&self, service: &Service) -> Result<Arc<InterfaceModel>, RpcDispatchError> {
{
let models = self
.models
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
if let Some(m) = models.get(&service.qualified_name) {
return Ok(Arc::clone(m));
}
}
let model = Arc::new(build_interface_model(&self.ast, service, &self.types)?);
let mut models = self
.models
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
models.insert(service.qualified_name.clone(), Arc::clone(&model));
Ok(model)
}
pub fn dispatch(
&self,
op_name: &str,
request_body_xml: &str,
) -> Result<String, RpcDispatchError> {
let (service, operation) = self.find_operation(op_name)?;
let model = self.model_for(service)?;
let disc = operation_hash(&operation.name);
let op_model = model.operations.get(&disc).ok_or_else(|| {
RpcDispatchError::Model(format!(
"operation `{op_name}`: not found in its own interface's model"
))
})?;
let in_data =
dynxml::from_xml_document(request_body_xml, &op_model.in_resolved.dynamic_type)
.map_err(|e| RpcDispatchError::BadRequest(e.to_string()))?;
let call_ast = Arc::clone(model.call_resolved.dynamic_type.ast());
let call_type_name = model.call_resolved.dynamic_type.name().to_string();
let call_data = DynamicData::Aggregate(vec![
(
"_d".to_string(),
DynamicData::integer(zerodds_ast::PrimitiveKind::Int32, i128::from(disc)),
),
(operation.name.clone(), in_data),
]);
let dialect = self
.participant
.runtime()
.map(|rt| rt.codec_dialect())
.unwrap_or_default();
let call_value =
DynamicRpcValue::new(call_ast, call_type_name, call_data).in_dialect(dialect);
let requester = self.requester_for(service)?;
let return_ast = Arc::clone(model.return_resolved.dynamic_type.ast());
let return_type_name = model.return_resolved.dynamic_type.name().to_string();
let reply =
dynamic_rpc::with_decode_context(&return_ast, &return_type_name, dialect, || {
requester.send_request_blocking(&call_value, Some(RPC_TIMEOUT))
})
.map_err(|e| match e {
zerodds_rpc::error::RpcError::Timeout => RpcDispatchError::Timeout,
zerodds_rpc::error::RpcError::RemoteException(code) => {
RpcDispatchError::RemoteException(code)
}
other => RpcDispatchError::Rpc(other.to_string()),
})?;
let result_data = reply.data.member(&operation.name).ok_or_else(|| {
RpcDispatchError::Rpc(format!(
"reply named no `{}` case (Return union discriminator mismatch)",
operation.name
))
})?;
let result_disc = match result_data.member("_d") {
Some(DynamicData::Integer { value, .. }) => *value as i32,
_ => {
return Err(RpcDispatchError::Rpc(
"Result union reply carried no discriminator".to_string(),
));
}
};
if result_disc == RETCODE_OK {
let out_resolved = &op_model.out_resolved;
let out_data = result_data.member(RESULT_MEMBER).ok_or_else(|| {
RpcDispatchError::Rpc(format!(
"Result union's `{RESULT_MEMBER}` case named no `{RESULT_MEMBER}` member"
))
})?;
return dynxml::to_xml_document("data", out_data, &out_resolved.dynamic_type)
.map_err(|e| RpcDispatchError::Rpc(e.to_string()));
}
let (_, member_name, exception_name, exc_resolved) = op_model
.exceptions
.iter()
.find(|(disc, _, _, _)| *disc == result_disc)
.ok_or_else(|| {
RpcDispatchError::Rpc(format!(
"Result union reply named an unrecognised discriminator {result_disc}"
))
})?;
let exc_data = result_data.member(member_name).ok_or_else(|| {
RpcDispatchError::Rpc(format!("Result union named no `{member_name}` member"))
})?;
let fields_xml = dynxml::to_xml_document("data", exc_data, &exc_resolved.dynamic_type)
.map_err(|e| RpcDispatchError::Rpc(e.to_string()))?;
Err(RpcDispatchError::OperationException {
operation: operation.name.clone(),
exception: exception_name.clone(),
fields_xml,
})
}
fn requester_for(
&self,
service: &Service,
) -> Result<Arc<Requester<DynamicRpcValue, DynamicRpcValue>>, RpcDispatchError> {
let mut requesters = self
.requesters
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
if let Some(r) = requesters.get(&service.qualified_name) {
return Ok(Arc::clone(r));
}
let scope = interface::module_chain(&self.ast, service.module)
.map_err(interface_error)?
.into_iter()
.map(|(_, name)| name)
.collect::<Vec<_>>()
.join("::");
let options = RpcEndpointOptions::for_interface(&scope, &service.name);
let requester = Requester::with_options(
&self.participant,
&service.qualified_name,
"",
&RpcQos::default(),
BindingStyle::FunctionCall,
&AnnotatedTopics::none(),
&options,
)
.map_err(|e| RpcDispatchError::Rpc(e.to_string()))?;
let requester = Arc::new(requester);
requesters.insert(service.qualified_name.clone(), Arc::clone(&requester));
Ok(requester)
}
}
fn interface_error(e: InterfaceError) -> RpcDispatchError {
RpcDispatchError::Model(format!("{e:?}"))
}
fn build_interface_model(
ast: &Ast,
service: &Service,
types: &TypeRegistry,
) -> Result<InterfaceModel, RpcDispatchError> {
let qualified = &service.qualified_name;
let flattened = all_operations_flattened(ast, service.id).map_err(interface_error)?;
let mut call_cases: Vec<(i32, String, ResolvedType)> = Vec::new();
let mut return_cases: Vec<(i32, String, ResolvedType)> = Vec::new();
let mut operations: HashMap<i32, OpModel> = HashMap::new();
for op in &flattened {
let disc = operation_hash(&op.name);
let in_type_name = in_struct_name(qualified, &op.name);
let in_members: Vec<(String, Id<TypeNode>)> =
op.inputs().map(|p| (p.name.clone(), p.type_)).collect();
let in_resolved = resolve_synthetic_struct(types, ast, &in_type_name, &in_members)?;
call_cases.push((disc, op.name.clone(), in_resolved.clone()));
let out_type_name = out_struct_name(qualified, &op.name);
let mut out_members: Vec<(String, Id<TypeNode>)> =
op.outputs().map(|p| (p.name.clone(), p.type_)).collect();
if let Some(ret) = op.returns {
out_members.push((RETURN_MEMBER.to_string(), ret));
}
let out_resolved = resolve_synthetic_struct(types, ast, &out_type_name, &out_members)?;
let mut result_cases: Vec<(i32, String, ResolvedType)> =
vec![(RETCODE_OK, RESULT_MEMBER.to_string(), out_resolved.clone())];
let mut exceptions = Vec::new();
for raise in &op.raises {
let ex_disc = interface::exception_hash(&raise.scoped_name);
let ex_member = exception_member_name(&raise.name);
let ex_type_name = format!("{qualified}_{}_{}_Ex", op.name, raise.name);
let ex_members = exception_members(ast, raise)
.map_err(RpcDispatchError::UnsupportedParameterType)?;
let ex_resolved = resolve_synthetic_struct(types, ast, &ex_type_name, &ex_members)?;
result_cases.push((ex_disc, ex_member.clone(), ex_resolved.clone()));
exceptions.push((ex_disc, ex_member, raise.name.clone(), ex_resolved));
}
let result_type_name = format!("{qualified}_{}_Result", op.name);
let result_resolved =
resolve_synthetic_union(types, &result_type_name, &result_cases, None)?;
return_cases.push((disc, op.name.clone(), result_resolved.clone()));
operations.insert(
disc,
OpModel {
in_resolved,
result_resolved,
out_resolved,
exceptions,
},
);
}
let call_type_name = format!("{qualified}_Call");
let call_resolved =
resolve_synthetic_union(types, &call_type_name, &call_cases, Some(UNKNOWN_OP_MEMBER))?;
let return_type_name = format!("{qualified}_Return");
let return_resolved = resolve_synthetic_union(
types,
&return_type_name,
&return_cases,
Some(UNKNOWN_OP_MEMBER),
)?;
Ok(InterfaceModel {
call_resolved,
return_resolved,
operations,
})
}
fn exception_members(ast: &Ast, raise: &Raise) -> Result<Vec<(String, Id<TypeNode>)>, String> {
let node: &ExceptionNode = ast
.exception(raise.id)
.map_err(|e| format!("exception `{}`: {e:?}", raise.name))?;
node.members
.iter()
.map(|member_id| {
let member = ast
.member(*member_id)
.map_err(|e| format!("exception `{}`: {e:?}", raise.name))?;
let type_id = member.type_.resolved().ok_or_else(|| {
format!(
"exception `{}`, member `{}`: type not resolved",
raise.name, member.name
)
})?;
Ok((member.name.clone(), type_id))
})
.collect()
}
fn resolve_synthetic_struct(
types: &TypeRegistry,
ast: &Ast,
type_name: &str,
members: &[(String, Id<TypeNode>)],
) -> Result<ResolvedType, RpcDispatchError> {
if let Some(r) = types.get(type_name) {
return Ok(r);
}
let (root, dependencies) = build_synthetic_struct(ast, type_name, members)
.map_err(RpcDispatchError::UnsupportedParameterType)?;
types
.resolve_synthetic(type_name, root, dependencies)
.map_err(|e| RpcDispatchError::Model(e.to_string()))
}
fn build_synthetic_struct(
ast: &Ast,
type_name: &str,
members: &[(String, Id<TypeNode>)],
) -> Result<(CompleteTypeObject, Vec<CompleteTypeObject>), String> {
use zerodds_types::from_ast::{
Form, FromAstError, complete_type_object, dependency_type_objects, type_identifier,
};
let mut builder = TypeObjectBuilder::struct_type(type_name).extensibility(Extensibility::Final);
let mut dependencies = Vec::new();
if members.is_empty() {
builder = builder.member(
DUMMY_MEMBER,
TypeIdentifier::Primitive(PrimitiveKind::Byte),
|mb| mb,
);
} else {
for (name, type_id) in members {
let ti = type_identifier(ast, *type_id, Form::Complete)
.map_err(|e| format!("member `{name}`: {e:?}"))?;
builder = builder.member(name.clone(), ti, |mb| mb);
match complete_type_object(ast, *type_id) {
Ok(object) => dependencies.push(object),
Err(FromAstError::Unrepresentable { .. }) => {
}
Err(e) => return Err(format!("member `{name}`: {e:?}")),
}
for dep in dependency_type_objects(ast, *type_id)
.map_err(|e| format!("member `{name}`: {e:?}"))?
{
dependencies.push(dep);
}
}
}
Ok((
CompleteTypeObject::Struct(builder.build_complete()),
dependencies,
))
}
fn resolve_synthetic_union(
types: &TypeRegistry,
type_name: &str,
cases: &[(i32, String, ResolvedType)],
default_member: Option<&str>,
) -> Result<ResolvedType, RpcDispatchError> {
use zerodds_types::from_ast::{
Form, complete_type_object, dependency_type_objects, type_identifier,
};
if let Some(r) = types.get(type_name) {
return Ok(r);
}
let mut builder =
TypeObjectBuilder::union_type(type_name, TypeIdentifier::Primitive(PrimitiveKind::Int32))
.extensibility(Extensibility::Final);
let mut dependencies = Vec::new();
for (label, member_name, case_resolved) in cases {
let case_ast = case_resolved.dynamic_type.ast();
let case_id = case_resolved.dynamic_type.id();
let case_type_id = type_identifier(case_ast, case_id, Form::Complete)
.map_err(|e| RpcDispatchError::Model(format!("{member_name}: {e:?}")))?;
builder = builder.case(member_name.clone(), case_type_id, vec![*label]);
dependencies.push(
complete_type_object(case_ast, case_id)
.map_err(|e| RpcDispatchError::Model(format!("{member_name}: {e:?}")))?,
);
for dep in dependency_type_objects(case_ast, case_id)
.map_err(|e| RpcDispatchError::Model(format!("{member_name}: {e:?}")))?
{
dependencies.push(dep);
}
}
if let Some(default_member) = default_member {
builder = builder.default_case(
default_member,
TypeIdentifier::Primitive(PrimitiveKind::Byte),
);
}
let root = CompleteTypeObject::Union(builder.build_complete());
types
.resolve_synthetic(type_name, root, dependencies)
.map_err(|e| RpcDispatchError::Model(e.to_string()))
}
#[cfg(test)]
#[allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
mod tests {
use super::*;
fn idl_ast(idl: &str) -> Arc<Ast> {
Arc::new(zerodds_wire_probe::encode::model_from_idl(idl, "t.idl").unwrap())
}
fn leased_participant() -> (zerodds_test_domains::DomainLease, DomainParticipant) {
let lease = zerodds_test_domains::lease_domain();
let participant = zerodds_dcps::DomainParticipantFactory::instance()
.create_participant(lease.id(), zerodds_dcps::DomainParticipantQos::default())
.expect("participant");
(lease, participant)
}
#[test]
fn finds_a_declared_operation_across_flattened_services() {
let ast = idl_ast("interface Calculator { long add(in long a, in long b); };");
let (_lease, participant) = leased_participant();
let dispatcher = RpcDispatcher::new(ast, participant).expect("new");
assert!(dispatcher.knows_operation("add"));
assert!(!dispatcher.knows_operation("subtract"));
}
#[test]
fn an_operation_with_no_params_gets_a_dummy_member() {
let ast = idl_ast("interface Fyi { void ping(); };");
let (root, _deps) = build_synthetic_struct(&ast, "Fyi_ping::In", &[]).expect("build");
let CompleteTypeObject::Struct(s) = root else {
panic!("expected struct")
};
assert_eq!(s.member_seq.len(), 1);
assert_eq!(s.member_seq[0].detail.name, DUMMY_MEMBER);
}
#[test]
fn every_synthetic_type_is_final_not_appendable() {
let ast = idl_ast("interface Calculator { long add(in long a, in long b); };");
let members = vec![("a".to_string(), {
let svc = services(&ast).unwrap().into_iter().next().unwrap();
let op = all_operations_flattened(&ast, svc.id)
.unwrap()
.into_iter()
.next()
.unwrap();
op.params[0].type_
})];
let (root, _deps) = build_synthetic_struct(&ast, "Calculator_add::In", &members).unwrap();
let CompleteTypeObject::Struct(s) = root else {
panic!("expected struct")
};
use zerodds_types::type_object::flags::StructTypeFlag;
assert!(!s.struct_flags.has(StructTypeFlag::IS_APPENDABLE));
assert!(!s.struct_flags.has(StructTypeFlag::IS_MUTABLE));
}
#[test]
fn builds_one_interface_wide_call_and_return_union() {
let ast = idl_ast(
"interface Calculator { long add(in long a, in long b); void log(in string msg); };",
);
let (_lease, participant) = leased_participant();
let dispatcher = RpcDispatcher::new(Arc::clone(&ast), participant).expect("new");
let service = services(&ast).unwrap().into_iter().next().unwrap();
let model = dispatcher.model_for(&service).expect("model");
assert_eq!(model.call_resolved.dynamic_type.name(), "Calculator_Call");
assert_eq!(model.call_resolved.dynamic_type.name(), "Calculator_Call");
assert_eq!(model.operations.len(), 2);
let add_disc = operation_hash("add");
let log_disc = operation_hash("log");
assert!(model.operations.contains_key(&add_disc));
assert!(model.operations.contains_key(&log_disc));
assert_eq!(
model.operations[&log_disc]
.result_resolved
.dynamic_type
.name(),
"Calculator_log_Result"
);
assert_eq!(
model.operations[&add_disc]
.result_resolved
.dynamic_type
.name(),
"Calculator_add_Result"
);
}
}