mod orb;
mod tango;
mod zmq;
use std::sync::{Arc, OnceLock};
use std::time::SystemTime;
use std::future::Future;
use std::borrow::Cow;
use std::pin::Pin;
use std::task;
use futures::future::{ready, BoxFuture};
use futures::stream::BoxStream;
use tracing::{debug, trace, warn};
use uuid::Uuid;
use crate::dev::{AttrConfig, AttrInfo, AttrQuality, AttrRead, AttrValueEvent, CommandInfo, DevInfo, DevSource, DevState, Named, NamedResult};
use crate::backend::{Backend, BackendDev, BackendServer, BackendServerConfig};
use crate::error::{Error, ErrorKind, Result};
use crate::value::{Scalar, Spectrum, Value};
use crate::future::resolve;
use orb::client::{Client, ClientObject, ClientRequest, ClientReply};
use orb::server::{Server, ServerObject, ServerRequest, ServerReply};
use orb::cdr::{CdrWrite, CdrRead, CdrReader, CdrWriter};
use orb::giop::ReplyStatus;
use zmq::{Consumer, Supplier};
use tango::{
tango_repo_id,
ClientIdent,
NamedError,
write_command_data, read_command_data,
read_attr, write_attr,
read_command_info, read_attr_info,
read_dev_info, write_dev_info,
write_command_info, write_attr_info,
};
const ALL_ATTRIBUTES: &str = "All attributes";
const ALL_ATTRIBUTES_3: &str = "All attributes_3";
static CLIENT_ID: ClientIdent = ClientIdent::Java {
main_class: Cow::Borrowed("rstango"),
uuid: Uuid::nil(),
};
#[derive(Debug)]
pub struct CorbaBackend {
client: OnceLock<Client>,
consumer: Arc<OnceLock<Consumer>>,
}
impl CorbaBackend {
pub fn new() -> Arc<Self> {
Arc::new(Self {
client: OnceLock::new(),
consumer: Arc::new(OnceLock::new()),
})
}
fn ensure_client(&self) -> Result<&Client> {
if let Some(client) = self.client.get() {
return Ok(client);
}
let client = Client::new()?;
Ok(self.client.get_or_init(|| client))
}
async fn check_device(&self, object: ClientObject) -> Result<Arc<dyn BackendDev>> {
let mut versions_it = DevVersion::ALL.into_iter().rev();
let version = loop {
let Some(version) = versions_it.next() else {
return Err(Error::new(ErrorKind::DevNotConnected)
.with_append_description(format_args!("Object {object:?}")));
};
let reply = object.request(b"_is_a", &version.repo_id()).await.map_err(corba_err)?;
let (mut reader, exception_id) = reply.read().map_err(corba_err)?;
if let Some(exception_id) = exception_id.as_deref() {
return Err(unexpected_user_exception_err(exception_id));
}
if reader.read_bool()? {
break version;
}
};
debug!(?version, "Connected to the device over CORBA");
Ok(Arc::new(CorbaDevBackend {
object,
version,
consumer: Arc::clone(&self.consumer),
}))
}
}
impl Backend for CorbaBackend {
fn connect(&self, itr: &str) -> BoxFuture<'_, Result<Arc<dyn BackendDev>>> {
if let Some(_corbaloc) = itr.strip_prefix("corbaloc:iiop:") {
Box::pin(ready(Err(Error::not_implemented()
.with_append_description("Scheme corbaloc:iiop:"))))
} else if let Some(ior) = itr.strip_prefix("IOR:") {
let ior = ior.to_string();
Box::pin(async move {
let client = self.ensure_client()?;
let object = client.connect_object_ior(&ior, &DevVersion::REPO_ID).await.map_err(corba_err)?;
self.check_device(object).await
})
} else {
Box::pin(ready(Err(Error::new(ErrorKind::NotHandled))))
}
}
fn connect_at(&self, host: &str, port: u16, name: &str) -> BoxFuture<'_, Result<Arc<dyn BackendDev>>> {
let host = host.to_string();
let object: Arc<[u8]> = Arc::from(name.to_ascii_lowercase().into_bytes());
Box::pin(async move {
let client = self.ensure_client()?;
let object = client.connect_object_iiop(&host, port, None, None, object).await.map_err(corba_err)?;
self.check_device(object).await
})
}
fn serve<'a>(&'a self, config: &'a BackendServerConfig) -> BoxFuture<'a, Result<Arc<dyn BackendServer>>> {
Box::pin(async move {
let host = config.host.as_deref().unwrap_or("0.0.0.0");
let port = config.port.unwrap_or(0);
let addrs = resolve(host, port).await?;
let mut last_err = None;
for addr in addrs {
match Server::bind(addr) {
Ok(server) => {
debug!(addr = %server.local_addr(), addrs = ?server.addrs(), "Serving devices over CORBA");
let supplier = config.admin_name.as_deref()
.map(|admin_name| Supplier::new(
admin_name,
config.tango_host.as_ref().map(|(host, port)| (&**host, *port)),
&server.host(),
server.local_addr().port(),
server.local_addr().ip(),
server.addrs(),
));
return Ok(Arc::new(CorbaServerBackend { server, supplier }) as Arc<dyn BackendServer>);
}
Err(err) => last_err = Some(err),
}
}
Err(last_err.unwrap_or_else(|| Error::new(ErrorKind::Corba)
.with_append_description(format_args!("No address to listen on for {host}:{port}"))))
})
}
}
#[derive(Debug)]
struct CorbaDevBackend {
object: ClientObject,
version: DevVersion,
consumer: Arc<OnceLock<Consumer>>,
}
impl CorbaDevBackend {
fn request(&self, operation: &[u8], input: &dyn CdrWrite) -> CorbaDevRequest {
CorbaDevRequest { inner: self.object.request(operation, input) }
}
fn consumer(&self) -> Result<&Consumer> {
if let Some(consumer) = self.consumer.get() {
return Ok(consumer);
}
let consumer = Consumer::start()?;
Ok(self.consumer.get_or_init(|| consumer))
}
}
impl BackendDev for CorbaDevBackend {
fn name(&self) -> BoxFuture<'_, Result<Arc<str>>> {
let request = self.request(b"_get_name", &());
Box::pin(async move {
let reply = request.await?;
Ok(Arc::from(reply.read::<String>()?))
})
}
fn admin_dev_name(&self) -> BoxFuture<'_, Result<Arc<str>>> {
let request = self.request(b"_get_adm_name", &());
Box::pin(async move {
let reply = request.await?;
Ok(Arc::from(reply.read::<String>()?))
})
}
fn description(&self) -> BoxFuture<'_, Result<Arc<str>>> {
let request = self.request(b"_get_description", &());
Box::pin(async move {
let reply = request.await?;
Ok(Arc::from(reply.read::<String>()?))
})
}
fn ping(&self) -> BoxFuture<'_, Result<()>> {
let request = self.request(b"ping", &());
Box::pin(async move {
let reply = request.await?;
reply.read::<()>()
})
}
fn info(&self) -> BoxFuture<'_, Result<DevInfo>> {
let (operation, info_version): (&[u8], DevVersion) = match self.version {
DevVersion::V1 |
DevVersion::V2 => (b"info", DevVersion::V1),
DevVersion::V3 |
DevVersion::V4 |
DevVersion::V5 => (b"info_3", DevVersion::V3),
DevVersion::V6 => (b"info_6", DevVersion::V6),
};
let request = self.request(operation, &());
Box::pin(async move {
let reply = request.await?;
let mut reader = reply.read_simple()?;
read_dev_info(&mut reader, info_version)
})
}
fn state(&self) -> BoxFuture<'_, Result<DevState>> {
let request = self.request(b"_get_state", &());
Box::pin(async move {
let reply = request.await?;
reply.read::<DevState>()
})
}
fn status(&self) -> BoxFuture<'_, Result<String>> {
let request = self.request(b"_get_status", &());
Box::pin(async move {
let reply = request.await?;
reply.read::<String>()
})
}
fn black_box(&self, max: Option<usize>) -> BoxFuture<'_, Result<Vec<String>>> {
let max = max.and_then(|n| i32::try_from(n).ok()).unwrap_or(i32::MAX);
let request = self.request(b"black_box", &max);
Box::pin(async move {
let reply = request.await?;
reply.read::<Vec<String>>()
})
}
fn command(&self, name: Scalar<str>, input: Value, source: DevSource) -> BoxFuture<'_, Result<Value>> {
struct Input<'a> {
version: DevVersion,
name: &'a str,
input: Value,
source: DevSource,
}
impl CdrWrite for Input<'_> {
fn write(&self, buf: &mut CdrWriter) -> Result<()> {
buf.write_string(self.name)?;
write_command_data(buf, &self.input)?;
if self.version >= DevVersion::V2 {
buf.write(&self.source)?;
}
if self.version >= DevVersion::V4 {
buf.write(&CLIENT_ID)?;
}
Ok(())
}
}
let operation: &[u8] = match self.version {
DevVersion::V1 => b"command_inout",
DevVersion::V2 |
DevVersion::V3 => b"command_inout_2",
DevVersion::V4 |
DevVersion::V5 |
DevVersion::V6 => b"command_inout_4",
};
let request = self.request(operation, &Input { version: self.version, name: name.value(), input, source });
Box::pin(async move {
let reply = request.await?;
read_command_data(&mut reply.read_simple()?)
})
}
fn read_attr(&self, names: Option<Spectrum<str>>, source: DevSource) -> BoxFuture<'_, Result<Vec<NamedResult<AttrRead<Value>>>>> {
struct Input<'a> {
version: DevVersion,
names: &'a dyn CdrWrite,
source: DevSource,
}
impl CdrWrite for Input<'_> {
fn write(&self, buf: &mut CdrWriter) -> Result<()> {
buf.write(&self.names)?;
if self.version >= DevVersion::V2 {
buf.write(&self.source)?;
}
if self.version >= DevVersion::V4 {
buf.write(&CLIENT_ID)?;
}
Ok(())
}
}
if let Some(names) = &names
&& let Err(err) = check_no_all_marker(names) {
return Box::pin(ready(Err(err)));
}
let operation: &[u8] = match self.version {
DevVersion::V1 => b"read_attributes",
DevVersion::V2 => b"read_attributes_2",
DevVersion::V3 => b"read_attributes_3",
DevVersion::V4 => b"read_attributes_4",
DevVersion::V5 |
DevVersion::V6 => b"read_attributes_5",
};
let version = self.version;
let names = match &names {
Some(names) => names as &dyn CdrWrite,
None => &[ALL_ATTRIBUTES],
};
let request = self.request(operation, &Input { version, names, source });
Box::pin(async move {
let reply = request.await?;
read_attr_list(&mut reply.read_simple()?, version)
})
}
fn write_attr(&self, values: Vec<Named<Value>>) -> BoxFuture<'_, Result<Vec<Result<()>>>> {
struct Input<'a> {
version: DevVersion,
values: &'a [Named<Value>],
}
impl CdrWrite for Input<'_> {
fn write(&self, buf: &mut CdrWriter) -> Result<()> {
buf.write_sequence_len(self.values.len())?;
for Named { name, value } in self.values {
write_attr(buf, self.version, name.value(), &Ok(AttrRead {
value: None,
w_value: Some(value.clone()),
quality: AttrQuality::Valid,
time: SystemTime::UNIX_EPOCH,
}))?;
}
if self.version >= DevVersion::V4 {
buf.write(&CLIENT_ID)?;
}
Ok(())
}
}
let (operation, input_version): (&[u8], DevVersion) = match self.version {
DevVersion::V1 |
DevVersion::V2 => (b"write_attributes", DevVersion::V1),
DevVersion::V3 => (b"write_attributes_3", DevVersion::V1),
DevVersion::V4 |
DevVersion::V5 |
DevVersion::V6 => (b"write_attributes_4", DevVersion::V4),
};
let request = self.request(operation, &Input { version: input_version, values: &values });
let count = values.len();
Box::pin(async move {
let reply = request.await?;
let (mut reader, exception_id) = reply.read_raw()?;
let Some(exception_id) = exception_id else {
return Ok((0..count).map(|_| Ok(())).collect());
};
if exception_id.as_ref() != tango_repo_id!("MultiDevFailed") {
return Err(unexpected_user_exception_err(&exception_id));
}
let mut results: Vec<Result<()>> = (0..count).map(|_| Ok(())).collect();
for failure in reader.read::<Vec<NamedError>>()? {
if let Some(result) = results.get_mut(failure.index) {
*result = Err(failure.err.unwrap_or_else(|| Error::new(ErrorKind::Internal)
.with_description("The device reported a failed attribute write holding no error")));
}
}
Ok(results)
})
}
fn write_read_attr(&self, values: Vec<Named<Value>>) -> BoxFuture<'_, Result<Vec<NamedResult<AttrRead<Value>>>>> {
struct Input<'a> {
version: DevVersion,
values: &'a [Named<Value>],
}
impl CdrWrite for Input<'_> {
fn write(&self, buf: &mut CdrWriter) -> Result<()> {
buf.write_sequence_len(self.values.len())?;
for Named { name, value } in self.values {
write_attr(buf, DevVersion::V4, name.value(), &Ok(AttrRead {
value: None,
w_value: Some(value.clone()),
quality: AttrQuality::Valid,
time: SystemTime::UNIX_EPOCH,
}))?;
}
if self.version >= DevVersion::V5 {
buf.write_sequence_len(self.values.len())?;
for write in self.values {
buf.write_string(write.name.value())?;
}
}
buf.write(&CLIENT_ID)?;
Ok(())
}
}
let (operation, reply_version): (&[u8], DevVersion) = match self.version {
DevVersion::V1 |
DevVersion::V2 |
DevVersion::V3 => {
return Box::pin(ready(Err(Error::not_implemented()
.with_append_description("Writing and reading attributes back in a single request needs a device of IDL v4 or later")
.with_append_description(format_args!("This one is {:?}", self.version)))));
}
DevVersion::V4 => (b"write_read_attributes_4", DevVersion::V4),
DevVersion::V5 |
DevVersion::V6 => (b"write_read_attributes_5", DevVersion::V5),
};
let request = self.request(operation, &Input { version: self.version, values: &values });
let names = values.into_iter().map(|write| write.name).collect::<Vec<_>>();
Box::pin(async move {
let reply = request.await?;
let (mut reader, exception_id) = reply.read_raw()?;
let Some(exception_id) = exception_id else {
return read_attr_list(&mut reader, reply_version);
};
if exception_id.as_ref() != tango_repo_id!("MultiDevFailed") {
return Err(unexpected_user_exception_err(&exception_id));
}
let mut failures = reader.read::<Vec<NamedError>>()?;
let entries = names.into_iter().enumerate().map(|(index, name)| {
let err = failures.iter_mut()
.find(|failure| failure.index == index)
.and_then(|failure| failure.err.take())
.unwrap_or_else(|| Error::new(ErrorKind::Internal)
.with_description("The device wrote this attribute but read none of them back, another one having failed"));
Named::new(name, Err(err))
}).collect();
Ok(entries)
})
}
fn command_info(&self, names: Option<Spectrum<str>>) -> BoxFuture<'_, Result<Vec<NamedResult<CommandInfo>>>> {
let (query_operation, list_operation, info_version): (&[u8], &[u8], DevVersion) = match self.version {
DevVersion::V1 => (b"command_query", b"command_list_query", DevVersion::V1),
_ => (b"command_query_2", b"command_list_query_2", DevVersion::V2),
};
if let Some(names) = names {
let requests = (0..names.len())
.map(|index| names.get_scalar([index]).expect("a name at every index"))
.map(|name| {
let request = self.request(query_operation, &name.value());
(name, request)
})
.collect::<Vec<_>>();
Box::pin(async move {
let mut entries = Vec::with_capacity(requests.len());
for (name, request) in requests {
let res: Result<CommandInfo> = async {
let reply = request.await?;
let mut reader = reply.read_simple()?;
read_command_info(&mut reader, info_version)?.1
}.await;
match res {
Err(err) if matches!(err.kind(), ErrorKind::TimedOut | ErrorKind::DevNotConnected) => {
return Err(err);
}
res => entries.push(Named::new(name, res)),
}
}
Ok(entries)
})
} else {
let request = self.request(list_operation, &());
Box::pin(async move {
let reply = request.await?;
let mut reader = reply.read_simple()?;
let len = reader.read_sequence_len()?;
let mut entries = Vec::with_capacity(len);
for _ in 0..len {
let (name, info) = read_command_info(&mut reader, info_version)?;
entries.push(Named::new(Scalar::<str>::from(&*name), info));
}
Ok(entries)
})
}
}
fn set_attr_config(&self, configs: Vec<Named<AttrConfig>>) -> BoxFuture<'_, Result<Vec<Result<()>>>> {
struct Input<'a> {
version: DevVersion,
infos: &'a [(Scalar<str>, AttrInfo)],
with_client: bool,
}
impl CdrWrite for Input<'_> {
fn write(&self, buf: &mut CdrWriter) -> Result<()> {
buf.write_sequence_len(self.infos.len())?;
for (name, info) in self.infos {
write_attr_info(buf, self.version, name.value(), info)?;
}
if self.with_client {
buf.write(&CLIENT_ID)?;
}
Ok(())
}
}
let (operation, config_version, with_client): (&'static [u8], DevVersion, bool) = match self.version {
DevVersion::V1 |
DevVersion::V2 => (b"set_attribute_config", DevVersion::V1, false),
DevVersion::V3 => (b"set_attribute_config_3", DevVersion::V3, false),
DevVersion::V4 => (b"set_attribute_config_4", DevVersion::V3, true),
DevVersion::V5 |
DevVersion::V6 => (b"set_attribute_config_5", DevVersion::V5, true),
};
let names = configs.iter().map(|config| config.name.value()).collect::<Vec<_>>();
let current = self.attr_info(Some(Spectrum::<str>::from(&names[..])));
let count = configs.len();
Box::pin(async move {
let mut infos = Vec::with_capacity(count);
for (Named { name, value: config }, current) in configs.into_iter().zip(current.await?) {
let mut info = current.value?;
info.config = config;
infos.push((name, info));
}
let reply = self.request(operation, &Input { version: config_version, infos: &infos, with_client }).await?;
let (mut reader, exception_id) = reply.inner.read().map_err(corba_err)?;
match exception_id.as_deref() {
None => Ok((0..count).map(|_| Ok(())).collect()),
Some(tango_repo_id!("DevFailed")) => {
let err = reader.read::<Error>()?;
match count {
1 => Ok(vec![Err(err)]),
_ => Err(err),
}
}
Some(exception_id) => Err(unexpected_user_exception_err(exception_id)),
}
})
}
fn attr_info(&self, names: Option<Spectrum<str>>) -> BoxFuture<'_, Result<Vec<NamedResult<AttrInfo>>>> {
let (operation, config_version): (&[u8], DevVersion) = match self.version {
DevVersion::V1 => (b"get_attribute_config", DevVersion::V1),
DevVersion::V2 => (b"get_attribute_config_2", DevVersion::V2),
DevVersion::V3 |
DevVersion::V4 => (b"get_attribute_config_3", DevVersion::V3),
DevVersion::V5 |
DevVersion::V6 => (b"get_attribute_config_5", DevVersion::V5),
};
if let Some(names) = &names
&& let Err(err) = check_no_all_marker(names) {
return Box::pin(ready(Err(err)));
}
let request = match &names {
Some(names) => self.request(operation, names),
None if config_version >= DevVersion::V3 => self.request(operation, &[ALL_ATTRIBUTES_3]),
None => self.request(operation, &[ALL_ATTRIBUTES]),
};
let expected = names;
Box::pin(async move {
let reply = request.await?;
let mut reader = reply.read_simple()?;
let len = reader.read_sequence_len()?;
let mut entries = Vec::with_capacity(len);
for index in 0..len {
let (name, info) = read_attr_info(&mut reader, config_version)?;
if let Some(expected) = expected.as_ref().and_then(|names| names.get([index]))
&& !name.eq_ignore_ascii_case(expected) {
return Err(Error::new(ErrorKind::InvalidData)
.with_description("The device answered with the configuration of another attribute")
.with_append_description(format_args!("Got {name:?} where {expected:?} was requested")));
}
entries.push(Named::new(Scalar::<str>::from(&*name), info));
}
Ok(entries)
})
}
fn subscribe_attr_value<'a>(&'a self, admin: Arc<dyn BackendDev>, name: &'a str, event: AttrValueEvent) -> BoxFuture<'a, Result<BoxStream<'static, Result<AttrRead<Value>>>>> {
match self.consumer() {
Ok(consumer) => zmq::subscribe_attr_value(consumer, self.name(), admin, name, event),
Err(err) => Box::pin(ready(Err(err))),
}
}
fn subscribe_attr_ready<'a>(&'a self, admin: Arc<dyn BackendDev>, name: &'a str) -> BoxFuture<'a, Result<BoxStream<'static, Result<()>>>> {
match self.consumer() {
Ok(consumer) => zmq::subscribe_attr_ready(consumer, self.name(), admin, name),
Err(err) => Box::pin(ready(Err(err))),
}
}
fn subscribe_attr_conf<'a>(&'a self, admin: Arc<dyn BackendDev>, name: &'a str) -> BoxFuture<'a, Result<BoxStream<'static, Result<AttrInfo>>>> {
match self.consumer() {
Ok(consumer) => zmq::subscribe_attr_conf(consumer, self.name(), admin, name),
Err(err) => Box::pin(ready(Err(err))),
}
}
fn command_history<'a>(&'a self, _name: &'a str, _max: usize) -> BoxFuture<'a, Result<Vec<(SystemTime, Result<Value>)>>> {
Box::pin(ready(Err(Error::not_implemented()
.with_append_description("The recorded history of a command"))))
}
fn attr_history<'a>(&'a self, _name: &'a str, _max: usize) -> BoxFuture<'a, Result<Vec<Result<AttrRead<Value>>>>> {
Box::pin(ready(Err(Error::not_implemented()
.with_append_description("The recorded history of an attribute"))))
}
}
struct CorbaDevRequest {
inner: ClientRequest,
}
impl Future for CorbaDevRequest {
type Output = Result<CorbaDevReply>;
fn poll(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> task::Poll<Self::Output> {
Pin::new(&mut self.get_mut().inner).poll(cx)
.map(|inner| inner
.map(|inner| CorbaDevReply { inner })
.map_err(corba_err))
}
}
fn corba_err(err: Error) -> Error {
if err.kind() != ErrorKind::Corba {
return err;
}
Error::new(ErrorKind::DevNotConnected).with_parent(err)
}
#[derive(Debug)]
struct CorbaDevReply {
inner: ClientReply,
}
impl CorbaDevReply {
fn read_raw(&self) -> Result<(CdrReader<'_>, Option<Cow<'_, str>>)> {
let (mut reader, exception_id) = self.inner.read().map_err(corba_err)?;
if let Some(tango_repo_id!("DevFailed")) = exception_id.as_deref() {
Err(match reader.read::<Error>() {
Ok(err) => err,
Err(err) => err,
})
} else {
Ok((reader, exception_id))
}
}
fn read_simple(&self) -> Result<CdrReader<'_>> {
let (reader, exception_id) = self.read_raw()?;
if let Some(exception_id) = exception_id.as_deref() {
Err(unexpected_user_exception_err(exception_id))
} else {
Ok(reader)
}
}
fn read<'a, O: CdrRead<'a>>(&'a self) -> Result<O> {
self.read_simple()?.read()
}
}
fn read_attr_list(reader: &mut CdrReader<'_>, version: DevVersion) -> Result<Vec<NamedResult<AttrRead<Value>>>> {
let len = reader.read_sequence_len()?;
let mut entries = Vec::with_capacity(len);
for _ in 0..len {
entries.push(read_attr(reader, version)?);
}
Ok(entries)
}
const SERVED_VERSION: DevVersion = DevVersion::V6;
const LIB_VERSION: i32 = 1000;
const SERVED_REPO_IDS: [&str; SERVED_VERSION.number() as usize] = {
let mut ids = [""; SERVED_VERSION.number() as usize];
let mut index = 0;
while index < ids.len() {
ids[index] = DevVersion::REPO_ID[SERVED_VERSION as usize - index];
index += 1;
}
ids
};
#[derive(Debug)]
struct CorbaServerBackend {
server: Server<CorbaServerObject>,
supplier: Option<Supplier>,
}
impl CorbaServerBackend {
fn supplier(&self) -> Result<&Supplier> {
self.supplier.as_ref().ok_or_else(|| Error::not_implemented()
.with_append_description("This server publishes no event over ZMQ"))
}
fn served(&self, name: &str) -> Option<Arc<dyn BackendDev>> {
self.server.get(name.as_bytes()).map(|object| Arc::clone(&object.dev))
}
fn is_served(&self, name: &str) -> bool {
self.server.is_registered(name.as_bytes())
}
}
impl BackendServer for CorbaServerBackend {
fn add(&self, name: &str, dev: Arc<dyn BackendDev>) -> Result<Arc<str>> {
let name = Arc::<str>::from(name.to_ascii_lowercase());
let itr = self.server.register(Arc::new(CorbaServerObject {
name: Arc::clone(&name),
dev,
})).map(Arc::<str>::from);
if itr.is_ok() {
debug!(dev = %name, "Serving the device over CORBA");
}
itr
}
fn remove(&self, name: &str) -> bool {
let id = name.to_ascii_lowercase().into_bytes();
self.server.deregister(&id)
}
fn host(&self) -> Arc<str> {
self.server.host()
}
fn port(&self) -> u16 {
self.server.local_addr().port()
}
fn version(&self) -> i32 {
SERVED_VERSION.number() as i32
}
fn stop(&self) {
if let Some(supplier) = &self.supplier {
supplier.stop();
}
self.server.stop();
}
fn wait(&self) -> BoxFuture<'_, ()> {
Box::pin(self.server.wait())
}
fn zmq_event_subscription_change(&self, args: Spectrum<str>) -> BoxFuture<'_, Result<(Spectrum<i32>, Spectrum<str>)>> {
Box::pin(async move {
let supplier = self.supplier()?;
supplier.subscription_change(&args, |name| self.served(name), SERVED_VERSION.number()).await
})
}
fn event_confirm_subscription(&self, args: Spectrum<str>) -> BoxFuture<'_, Result<()>> {
Box::pin(async move {
let supplier = self.supplier()?;
supplier.confirm_subscription(&args, |name| self.is_served(name)).await
})
}
}
#[derive(Debug)]
struct CorbaServerObject {
name: Arc<str>,
dev: Arc<dyn BackendDev>,
}
impl ServerObject for CorbaServerObject {
type RequestFuture = BoxFuture<'static, Result<ServerReply>>;
fn id(&self) -> &[u8] {
self.name.as_bytes()
}
fn repo_ids(&self) -> &[&str] {
&SERVED_REPO_IDS
}
fn request(self: Arc<Self>, request: ServerRequest) -> Self::RequestFuture {
Box::pin(async move { self.handle_request(&request).await })
}
}
fn answer(request: &ServerRequest, res: Result<ServerReply>) -> Result<ServerReply> {
let err = match res {
Ok(reply) => return Ok(reply),
Err(err) => err,
};
let mut reply = request.reply_user_exception(tango_repo_id!("DevFailed"))?;
reply.write(&err)?;
Ok(reply)
}
impl CorbaServerObject {
async fn handle_request(&self, request: &ServerRequest) -> Result<ServerReply> {
let operation = request.operation();
let res = self.dispatch(operation, request).await;
if let Err(err) = &res {
trace!(dev = %self.name, operation = %String::from_utf8_lossy(operation), %err, "Failed to run an operation");
}
answer(request, res)
}
async fn dispatch<'d>(&self, operation: &[u8], request: &'d ServerRequest) -> Result<ServerReply> {
let mut input = request.read();
let mut reply = request.reply(ReplyStatus::NoException)?;
let output = &mut reply;
match operation {
b"_get_name" => output.write_string(&self.dev.name().await?)?,
b"_get_adm_name" => output.write_string(&self.dev.admin_dev_name().await?)?,
b"_get_description" => output.write_string(&self.dev.description().await?)?,
b"_get_state" => output.write(&self.dev.state().await?)?,
b"_get_status" => output.write_string(&self.dev.status().await?)?,
b"ping" => self.dev.ping().await?,
b"info" | b"info_3" | b"info_6" => {
let version = match operation {
b"info" => DevVersion::V1,
b"info_3" => DevVersion::V3,
_ => DevVersion::V6,
};
write_dev_info(output, version, &self.dev.info().await?)?;
}
b"black_box" => {
let max = input.read_long()?;
let entries = self.dev.black_box(usize::try_from(max).ok()).await?;
output.write(&entries)?;
}
b"command_inout" |
b"command_inout_2" |
b"command_inout_4" => {
let name = input.read::<Scalar<str>>()?;
let arg = read_command_data(&mut input)?;
let source = if operation == b"command_inout" {
DevSource::Dev
} else {
input.read()?
};
let answer = self.dev.command(name, arg, source).await?;
write_command_data(output, &answer)?;
}
b"read_attributes" |
b"read_attributes_2" |
b"read_attributes_3" |
b"read_attributes_4" |
b"read_attributes_5" => {
let version = match operation {
b"read_attributes" => DevVersion::V1,
b"read_attributes_2" => DevVersion::V2,
b"read_attributes_3" => DevVersion::V3,
b"read_attributes_4" => DevVersion::V4,
_ => DevVersion::V5,
};
let names = input.read::<Spectrum<str>>()?;
let source = if version >= DevVersion::V2 {
input.read()?
} else {
DevSource::Dev
};
let readings = self.dev.read_attr(all_attributes(names), source).await?;
output.write_sequence_len(readings.len())?;
for Named { name, value: read } in &readings {
write_attr(output, version, name.value(), read)?;
}
}
b"write_attributes" |
b"write_attributes_3" |
b"write_attributes_4" => {
let version = if operation == b"write_attributes_4" {
DevVersion::V4
} else {
DevVersion::V1
};
let values = read_attr_values(&mut input, version)?;
let names = values.iter().map(|write| write.name.clone()).collect::<Vec<_>>();
let results = self.dev.write_attr(values).await?;
let mut failures = results.into_iter()
.enumerate()
.filter_map(|(index, res)| res.err().map(|err| (index, err)))
.collect::<Vec<_>>();
if !failures.is_empty() {
if version < DevVersion::V4 {
return Err(failures.swap_remove(0).1);
}
*output = request.reply_user_exception(tango_repo_id!("MultiDevFailed"))?;
output.write_sequence_len(failures.len())?;
for (index, err) in failures {
output.write(&NamedError {
name: Cow::Borrowed(names.get(index).map(|name| name.value()).unwrap_or("")),
index,
err: Some(err),
})?;
}
}
}
b"write_read_attributes_4" |
b"write_read_attributes_5" => {
let reply_version = if operation == b"write_read_attributes_5" {
DevVersion::V5
} else {
DevVersion::V4
};
let values = read_attr_values(&mut input, DevVersion::V4)?;
let readings = self.dev.write_read_attr(values).await?;
output.write_sequence_len(readings.len())?;
for Named { name, value: read } in &readings {
write_attr(output, reply_version, name.value(), read)?;
}
}
b"command_query" |
b"command_query_2" => {
let version = if operation == b"command_query_2" {
DevVersion::V2
} else {
DevVersion::V1
};
let name = input.read::<Scalar<str>>()?;
let names = name.clone().reshape([1]).expect("one name");
let infos = self.dev.command_info(Some(names)).await?;
let Some(Named { name, value: info }) = infos.into_iter().next() else {
return Err(Error::new(ErrorKind::CommandNotFound)
.with_append_description(format_args!("Command {name:?}")));
};
write_command_info(output, version, name.value(), &info?)?;
}
b"command_list_query" |
b"command_list_query_2" => {
let version = if operation == b"command_list_query_2" {
DevVersion::V2
} else {
DevVersion::V1
};
let infos = self.dev.command_info(None).await?;
let out = &mut **output;
let len = SequenceLen::start(out)?;
let mut count = 0;
for Named { name, value: info } in infos {
let info = match info {
Ok(info) => info,
Err(err) => {
warn!(command = %name.value(), %err, "Failed to describe the command, leaving it out of the listing");
continue;
}
};
write_command_info(out, version, name.value(), &info)?;
count += 1;
}
len.finish(out, count)?;
}
b"set_attribute_config" |
b"set_attribute_config_3" |
b"set_attribute_config_4" |
b"set_attribute_config_5" => {
let version = match operation {
b"set_attribute_config" => DevVersion::V1,
b"set_attribute_config_5" => DevVersion::V5,
_ => DevVersion::V3,
};
let len = input.read_sequence_len()?;
let mut configs = Vec::with_capacity(len);
for _ in 0..len {
let (name, info) = read_attr_info(&mut input, version)?;
configs.push(Named::new(Scalar::from(&*name), info?.config));
}
let results = self.dev.set_attr_config(configs).await?;
if let Some(err) = results.into_iter().find_map(Result::err) {
return Err(err);
}
}
b"get_attribute_config" |
b"get_attribute_config_2" |
b"get_attribute_config_3" |
b"get_attribute_config_5" => {
let version = match operation {
b"get_attribute_config" => DevVersion::V1,
b"get_attribute_config_2" => DevVersion::V2,
b"get_attribute_config_3" => DevVersion::V3,
_ => DevVersion::V5,
};
let names = input.read::<Spectrum<str>>()?;
let requested = all_attributes(names);
let named = requested.is_some();
let infos = self.dev.attr_info(requested).await?;
let out = &mut **output;
let len = SequenceLen::start(out)?;
let mut count = 0;
for Named { name, value: info } in infos {
let info = match info {
Ok(info) => info,
Err(err) if named => return Err(err),
Err(err) => {
warn!(attr = %name.value(), %err, "Failed to describe the attribute, leaving it out of the listing");
continue;
}
};
write_attr_info(out, version, name.value(), &info)?;
count += 1;
}
len.finish(out, count)?;
}
b"command_inout_history_2" |
b"command_inout_history_4" |
b"read_attribute_history_2" |
b"read_attribute_history_3" |
b"read_attribute_history_4" |
b"read_attribute_history_5" |
b"get_pipe_config_5" |
b"set_pipe_config_5" |
b"read_pipe_5" |
b"write_pipe_5" |
b"write_read_pipe_5" => {
return Err(Error::not_implemented()
.with_append_description(format_args!("Operation {:?}, in this server", String::from_utf8_lossy(operation))));
}
_ => {
return Err(Error::new(ErrorKind::NotHandled));
}
}
Ok(reply)
}
}
fn all_attributes(names: Spectrum<str>) -> Option<Spectrum<str>> {
if has_all_marker(&names) {
None
} else {
Some(names)
}
}
fn has_all_marker(names: &Spectrum<str>) -> bool {
names.iter().any(|name| name == ALL_ATTRIBUTES || name == ALL_ATTRIBUTES_3)
}
fn check_no_all_marker(names: &Spectrum<str>) -> Result<()> {
if has_all_marker(names) {
return Err(Error::new(ErrorKind::AttrNameInvalid)
.with_append_description("Given attribute names contains the 'all attributes' marker")
.with_append_description(format_args!("{names:?}")));
}
Ok(())
}
fn read_attr_values(input: &mut CdrReader<'_>, version: DevVersion) -> Result<Vec<Named<Value>>> {
let len = input.read_sequence_len()?;
let mut values = Vec::with_capacity(len);
for _ in 0..len {
let Named { name, value: read } = read_attr(input, version)?;
let value = read?.w_value.ok_or_else(|| Error::new(ErrorKind::InvalidData)
.with_description(format!("The value to write of the attribute {:?} is missing", name.value())))?;
values.push(Named::new(name, value));
}
Ok(values)
}
struct SequenceLen(usize);
impl SequenceLen {
fn start(buf: &mut CdrWriter) -> Result<Self> {
let pos = buf.position();
buf.write_sequence_len(0)?;
Ok(Self(pos))
}
fn finish(self, buf: &mut CdrWriter, len: usize) -> Result<()> {
let end = buf.position();
buf.set_position(self.0);
buf.write_sequence_len(len)?;
buf.set_position(end);
Ok(())
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
pub(crate) enum DevVersion {
V1,
V2,
V3,
V4,
V5,
V6,
}
impl DevVersion {
pub const ALL: [Self; 6] = [
Self::V1,
Self::V2,
Self::V3,
Self::V4,
Self::V5,
Self::V6,
];
const REPO_ID: [&str; 6] = [
tango_repo_id!("Device"),
tango_repo_id!("Device_2"),
tango_repo_id!("Device_3"),
tango_repo_id!("Device_4"),
tango_repo_id!("Device_5"),
tango_repo_id!("Device_6"),
];
#[inline]
pub const fn repo_id(&self) -> &'static str {
Self::REPO_ID[*self as usize]
}
#[inline]
pub const fn number(&self) -> u32 {
*self as u32 + 1
}
#[inline]
pub const fn from_number(number: u32) -> Option<Self> {
match number {
1..=6 => Some(Self::ALL[number as usize - 1]),
_ => None,
}
}
}
fn unexpected_user_exception_err<'a>(exception_id: &'a str) -> Error {
Error::new(ErrorKind::Internal)
.with_description(format!("Unexpected user exception: {exception_id}"))
}
#[cfg(test)]
mod tests {
use std::net::TcpListener;
use std::time::Duration;
use crate::future::block_on_timeout;
use super::*;
#[test]
fn connect_refused() {
let backend = CorbaBackend::new();
let port = TcpListener::bind("127.0.0.1:0").unwrap().local_addr().unwrap().port();
let err = block_on_timeout(backend.connect_at("127.0.0.1", port, "test/dev/1"), Duration::from_secs(5))
.unwrap()
.unwrap_err();
assert_eq!(err.kind(), ErrorKind::DevNotConnected);
assert_eq!(err.parent().unwrap().kind(), ErrorKind::Corba);
}
}