use std::sync::{Arc, OnceLock, RwLock};
use std::time::{Duration, SystemTime};
use std::fmt::{self, Debug, Display};
use std::task::{Context, Poll};
use std::hash::{Hash, Hasher};
use std::future::Future;
use std::cmp::Ordering;
use std::str::FromStr;
use std::borrow::Cow;
use std::{env, mem};
use std::pin::Pin;
use futures::stream::{BoxStream, Stream, StreamExt};
use futures::ready;
use tracing::{debug, instrument, warn};
use futures::future::{poll_fn, BoxFuture};
use crate::value::{AttrDataFormat, AttrDataKind, CommandDataKind, FromValue, IntoValue, Scalar, Spectrum, Value};
use crate::database::{prop_params, set_prop_params, DatabaseProxy, GetPropSpec, PropCommands, SetPropSpec};
use crate::backend::{default_backend, Backend, BackendDev};
use crate::error::{Error, ErrorKind, Result, reason};
use crate::name::{ArcTrl, IntoTrl, TrlObject};
use crate::future::block_on_timeout;
use crate::local;
const DEFAULT_TIMEOUT: Duration = Duration::from_secs(3);
pub(crate) const DATABASE_DEV_NAME: &str = "database";
#[derive(Debug)]
pub struct DevProxy {
inner: Box<ProxyInner>,
state: RwLock<ProxyState>,
}
#[derive(Debug, Clone)]
struct ProxyInner {
backend: Arc<dyn Backend>,
source: DevSource,
timeout: Duration,
target: ProxyTarget,
}
#[derive(Debug, Clone)]
struct ProxyState {
backend: Option<Arc<dyn BackendDev>>,
database_host_and_port: Option<(Arc<str>, u16)>,
}
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub(crate) enum ProxyTarget {
DatabaseFromEnv,
DatabaseAt {
host: Arc<str>,
port: u16,
},
RegularFromDatabase {
database: DatabaseProxy,
name: Arc<str>,
},
RegularAt {
host: Arc<str>,
port: u16,
name: Arc<str>,
},
}
impl DevProxy {
pub fn new<'a>(trl: impl IntoTrl<'a>) -> Self {
Self::new_with_backend(default_backend(), trl)
}
pub fn try_new<'a>(trl: impl IntoTrl<'a>) -> Option<Self> {
Self::try_new_with_backend(default_backend(), trl)
}
pub fn new_from_database(database: &DatabaseProxy, name: &str) -> Self {
Self::new_from_database_with_backend(default_backend(), database, name)
}
pub fn new_with_backend<'a>(backend: Arc<dyn Backend>, trl: impl IntoTrl<'a>) -> Self {
Self::try_new_with_backend(backend, trl).expect("The given TRL does not target a device")
}
pub fn try_new_with_backend<'a>(backend: Arc<dyn Backend>, trl: impl IntoTrl<'a>) -> Option<Self> {
let trl = trl.into_trl().ok()?;
if trl.property().is_some() {
return None;
}
let name = match trl.object() {
TrlObject::Alias(name) => name,
TrlObject::Dev(name) => name,
TrlObject::Attr(_, _) => return None,
};
let target = match (trl.db(), trl.host_and_port()) {
(true, Some((host, port))) => ProxyTarget::RegularFromDatabase {
database: DatabaseProxy::new_with_backend(Arc::clone(&backend), host, port),
name: Arc::from(name),
},
(true, None) => ProxyTarget::RegularFromDatabase {
database: DatabaseProxy::new_from_env_with_backend(Arc::clone(&backend)),
name: Arc::from(name),
},
(false, Some((host, port))) => ProxyTarget::RegularAt {
host: Arc::from(host),
port,
name: Arc::from(name),
},
(false, None) => return None,
};
Some(Self::new_from_target_with_backend(backend, target))
}
pub fn new_from_database_with_backend(backend: Arc<dyn Backend>, database: &DatabaseProxy, name: &str) -> Self {
Self::new_from_target_with_backend(backend, ProxyTarget::RegularFromDatabase {
database: database.clone(),
name: Arc::from(name),
})
}
pub(crate) fn new_from_target_with_backend(backend: Arc<dyn Backend>, target: ProxyTarget) -> Self {
Self {
inner: Box::new(ProxyInner {
backend,
target,
source: DevSource::CacheDev,
timeout: DEFAULT_TIMEOUT,
}),
state: RwLock::new(ProxyState {
backend: None,
database_host_and_port: None,
}),
}
}
pub fn backend(&self) -> &Arc<dyn Backend> {
&self.inner.backend
}
pub fn timeout(&self) -> Duration {
self.inner.timeout
}
pub fn set_timeout(&mut self, timeout: Duration) -> Duration {
mem::replace(&mut self.inner.timeout, timeout)
}
pub fn source(&self) -> DevSource {
self.inner.source
}
pub fn set_source(&mut self, source: DevSource) -> DevSource {
mem::replace(&mut self.inner.source, source)
}
fn connect_backend(&self) -> BoxFuture<'_, Result<(Arc<dyn BackendDev>, Option<(Arc<str>, u16)>)>> {
Box::pin(async move {
match self.inner.target {
ProxyTarget::DatabaseFromEnv => {
debug!("Connecting to the database of TANGO_HOST");
let databases = env_databases();
if databases.is_empty() {
return Err(Error::new(ErrorKind::TangoHostNotSet)
.with_append_description(match env::var("TANGO_HOST") {
Ok(var) => format!("TANGO_HOST is {var:?}"),
Err(_) => "TANGO_HOST is not defined in the environment".to_string(),
}));
}
let mut last_err = None;
for (host, port) in databases {
if let Some(dev_backend) = local::find_at(host, *port, DATABASE_DEV_NAME) {
debug!(%host, port, "Connected to the database, served by this process");
return Ok((dev_backend, Some((host.clone(), *port))));
}
match self.inner.backend.connect_at(host, *port, DATABASE_DEV_NAME).await {
Ok(dev_backend) => {
debug!(%host, port, "Connected to the database");
return Ok((dev_backend, Some((host.clone(), *port))));
}
Err(err) => {
warn!(%host, port, %err, "Failed to connect to the database, trying the next one");
last_err = Some(err);
}
}
}
let mut err = Error::new(ErrorKind::DatabaseNotConnected)
.with_append_description(format_args!("Tried, from TANGO_HOST: {}", DisplayHostAndPorts(databases)));
if let Some(last_err) = last_err {
err.set_parent(last_err);
}
Err(err)
}
ProxyTarget::DatabaseAt { ref host, port } => {
debug!(%host, port, "Connecting to the database");
if let Some(dev_backend) = local::find_at(host, port, DATABASE_DEV_NAME) {
debug!(%host, port, "Connected to the database, served by this process");
return Ok((dev_backend, Some((host.clone(), port))));
}
match self.inner.backend.connect_at(&host, port, DATABASE_DEV_NAME).await {
Ok(dev_backend) => {
debug!(%host, port, "Connected to the database");
Ok((dev_backend, Some((host.clone(), port))))
}
Err(err) => Err(Error::new(ErrorKind::DatabaseNotConnected)
.with_append_description(format_args!("Tried: {host}:{port}"))
.with_parent(err)),
}
}
ProxyTarget::RegularFromDatabase { ref database, ref name } => {
debug!(dev = %name, "Connecting to the device, through the database");
let info = match database.import_dev_async(name).await {
Ok(info) if !info.exported =>
return Err(Error::new(ErrorKind::DevNotFound)
.with_description("The device is defined in the database but no running server exports it")
.with_append_description(format_args!("Device {name:?} of the server {:?}", info.server))),
Ok(info) => info,
Err(err) if let ErrorKind::DevNotConnected = err.kind() =>
return Err(Error::new(ErrorKind::DatabaseNotConnected)
.with_parent(err)),
Err(err) => return Err(err),
};
if let Some(dev_backend) = local::find(info.itr.value()) {
debug!(dev = %name, "Connected to the device, served by this process");
return Ok((dev_backend, None));
}
match self.inner.backend.connect(info.itr.value()).await {
Ok(dev_backend) => {
debug!(dev = %name, "Connected to the device");
Ok((dev_backend, None))
}
Err(err) => Err(err.with_append_description(format_args!(
"While connecting to the device {name:?} of the server {:?}, exported by the database on host {:?}",
info.server, info.host))),
}
}
ProxyTarget::RegularAt { ref host, port, ref name } => {
debug!(%host, port, dev = %name, "Connecting to the device");
if let Some(dev_backend) = local::find_at(host, port, name) {
debug!(%host, port, dev = %name, "Connected to the device, served by this process");
return Ok((dev_backend, None));
}
match self.inner.backend.connect_at(&host, port, &name).await {
Ok(dev_backend) => {
debug!(%host, port, dev = %name, "Connected to the device");
Ok((dev_backend, None))
}
Err(err) => Err(err.with_append_description(format_args!(
"While connecting to the device {name:?} at {host}:{port}"))),
}
}
}
})
}
async fn ensure_backend(&self) -> Result<Arc<dyn BackendDev>> {
if let Some(backend) = self.state.read().unwrap().backend.as_ref() {
return Ok(Arc::clone(backend));
}
let (backend, database_host_and_port) = self.connect_backend().await?;
let mut state = self.state.write().unwrap();
match state.backend {
Some(ref backend) => Ok(Arc::clone(backend)),
None => {
state.backend = Some(Arc::clone(&backend));
state.database_host_and_port = database_host_and_port;
Ok(backend)
}
}
}
fn reset_backend(&self) {
let mut state = self.state.write().unwrap();
state.backend = None;
state.database_host_and_port = None;
}
fn check_backend<T>(&self, res: Result<T>) -> Result<T> {
if let Err(err) = &res && err.kind() == ErrorKind::DevNotConnected {
self.reset_backend();
}
res
}
pub(crate) fn block_on<F: Future>(&self, fut: F) -> Result<F::Output> {
block_on_timeout(fut, self.inner.timeout)
}
fn database_host_and_port(&self) -> Result<Option<(Arc<str>, u16)>> {
self.block_on(self.database_host_and_port_async())?
}
pub(crate) async fn database_host_and_port_async(&self) -> Result<Option<(Arc<str>, u16)>> {
self.ensure_backend().await?;
Ok(self.state.read().unwrap().database_host_and_port.clone())
}
pub fn database(&self) -> Result<&DatabaseProxy> {
match self.inner.target {
ProxyTarget::DatabaseFromEnv |
ProxyTarget::DatabaseAt { .. } =>
Ok(DatabaseProxy::from_dev(self)),
ProxyTarget::RegularFromDatabase { ref database, .. } =>
Ok(database),
ProxyTarget::RegularAt { .. } =>
Err(Error::new(ErrorKind::DatabaseNeeded)),
}
}
#[instrument(skip(self), fields(target = %self.inner.target))]
pub fn trl(&self) -> Result<ArcTrl> {
match self.inner.target {
ProxyTarget::DatabaseFromEnv => {
let name = self.name()?;
let (host, port) = self.database_host_and_port()?
.expect("This should be defined if the target is a database");
Ok(format!("tango://{host}:{port}/{name}").parse::<ArcTrl>().unwrap())
}
ProxyTarget::DatabaseAt { ref host, port } => {
let name = self.name()?;
Ok(format!("tango://{host}:{port}/{name}").parse::<ArcTrl>().unwrap())
}
ProxyTarget::RegularFromDatabase { ref database, ref name } => {
let (host, port) = database.database_host_and_port()?
.expect("This should be defined if the target is a database");
Ok(format!("tango://{host}:{port}/{name}").parse::<ArcTrl>().unwrap())
}
ProxyTarget::RegularAt { ref host, port, ref name } => {
Ok(format!("tango://{host}:{port}/{name}#dbase=no").parse::<ArcTrl>().unwrap())
}
}
}
#[instrument(skip(self), fields(target = %self.inner.target))]
pub fn connect(&self) -> Result<()> {
self.block_on(self.connect_async())?
}
pub async fn connect_async(&self) -> Result<()> {
self.ensure_backend().await.map(drop)
}
#[instrument(skip(self), fields(target = %self.inner.target))]
pub fn name(&self) -> Result<Arc<str>> {
self.block_on(self.name_async())?
}
pub async fn name_async(&self) -> Result<Arc<str>> {
let backend = self.ensure_backend().await?;
self.check_backend(backend.name().await)
}
#[instrument(skip(self), fields(target = %self.inner.target))]
pub fn admin_dev_name(&self) -> Result<Arc<str>> {
self.block_on(self.admin_dev_name_async())?
}
pub async fn admin_dev_name_async(&self) -> Result<Arc<str>> {
let backend = self.ensure_backend().await?;
self.check_backend(backend.admin_dev_name().await)
}
#[instrument(skip(self), fields(target = %self.inner.target))]
pub fn admin_dev(&self) -> Result<DevProxy> {
Ok(self.admin_dev_of(self.admin_dev_name()?))
}
pub async fn admin_dev_async(&self) -> Result<DevProxy> {
Ok(self.admin_dev_of(self.admin_dev_name_async().await?))
}
fn admin_dev_of(&self, name: Arc<str>) -> DevProxy {
let backend = Arc::clone(&self.inner.backend);
DevProxy::new_from_target_with_backend(backend, match &self.inner.target {
ProxyTarget::DatabaseFromEnv => todo!(),
ProxyTarget::DatabaseAt { .. } => {
ProxyTarget::RegularFromDatabase {
database: DatabaseProxy::with_dev(self.clone()),
name,
}
}
ProxyTarget::RegularFromDatabase { database, .. } => {
ProxyTarget::RegularFromDatabase {
database: database.clone(),
name,
}
}
ProxyTarget::RegularAt { host, port, .. } => {
ProxyTarget::RegularAt {
host: Arc::clone(host),
port: *port,
name,
}
}
})
}
#[instrument(skip(self), fields(target = %self.inner.target))]
pub fn info(&self) -> Result<DevInfo> {
self.block_on(self.info_async())?
}
pub async fn info_async(&self) -> Result<DevInfo> {
let backend = self.ensure_backend().await?;
self.check_backend(backend.info().await)
}
#[instrument(skip(self), fields(target = %self.inner.target))]
pub fn ping(&self) -> Result<()> {
self.block_on(self.ping_async())?
}
pub async fn ping_async(&self) -> Result<()> {
let backend = self.ensure_backend().await?;
self.check_backend(backend.ping().await)
}
#[instrument(skip(self), fields(target = %self.inner.target))]
pub fn state(&self) -> Result<DevState> {
self.block_on(self.state_async())?
}
pub async fn state_async(&self) -> Result<DevState> {
let backend = self.ensure_backend().await?;
self.check_backend(backend.state().await)
}
#[instrument(skip(self), fields(target = %self.inner.target))]
pub fn status(&self) -> Result<String> {
self.block_on(self.status_async())?
}
pub async fn status_async(&self) -> Result<String> {
let backend = self.ensure_backend().await?;
self.check_backend(backend.status().await)
}
pub fn init(&self) -> Result<()> {
self.command("Init", ())
}
pub fn init_async(&self) -> impl Future<Output = Result<()>> + Send {
self.command_async("Init", ())
}
#[instrument(skip(self), fields(target = %self.inner.target))]
pub fn black_box(&self, max: Option<usize>) -> Result<Vec<String>> {
self.block_on(self.black_box_async(max))?
}
pub async fn black_box_async(&self, max: Option<usize>) -> Result<Vec<String>> {
let backend = self.ensure_backend().await?;
self.check_backend(backend.black_box(max).await)
}
#[instrument(skip(self, input), fields(target = %self.inner.target))]
pub fn command_async<I, O>(&self, name: &str, input: I) -> impl Future<Output = Result<O>> + Send
where
I: IntoValue,
O: FromValue,
{
let command = Scalar::from(name);
let input = input.into_value();
async move {
let backend = self.ensure_backend().await?;
let res = backend.command(command, input, self.inner.source).await;
O::from_value(self.check_backend(res).map_err(command_err)?)
}
}
pub fn command<I, O>(&self, name: &str, input: I) -> Result<O>
where
I: IntoValue,
O: FromValue,
{
self.block_on(self.command_async(name, input))?
}
#[instrument(skip(self, names), fields(target = %self.inner.target))]
pub fn read_attr_async<Spec, Names: ToNames>(&self, names: Names) -> impl Future<Output = Result<Spec::Output>> + Send
where
Spec: ReadAttrSpec<Names>,
{
let names = names.to_names();
async move {
let backend = self.ensure_backend().await?;
let res = backend.read_attr(names, self.inner.source).await;
Spec::collect(self.check_backend(res)?)
}
}
pub fn read_attr<Spec, Names: ToNames>(&self, names: Names) -> Result<Spec::Output>
where
Spec: ReadAttrSpec<Names>,
{
self.block_on(self.read_attr_async::<Spec, Names>(names))?
}
#[instrument(skip(self, spec), fields(target = %self.inner.target))]
pub fn write_attr_async<'a, Spec>(&self, spec: Spec) -> impl Future<Output = Result<Spec::Output>> + Send
where
Spec: WriteAttrSpec<'a>,
{
let values = spec.into_values();
async move {
let backend = self.ensure_backend().await?;
let res = backend.write_attr(values).await;
let outcomes = self.check_backend(res)?.into_iter()
.map(|res| res.map_err(write_attr_err))
.collect();
Spec::collect(outcomes)
}
}
pub fn write_attr<'a, Spec>(&self, spec: Spec) -> Result<Spec::Output>
where
Spec: WriteAttrSpec<'a>,
{
self.block_on(self.write_attr_async(spec))?
}
#[instrument(skip(self, spec), fields(target = %self.inner.target))]
pub fn write_read_attr_async<'a, Spec>(&self, spec: Spec) -> impl Future<Output = Result<<Spec::ReadSpec as ReadAttrSpec<Spec::ReadNames>>::Output>> + Send
where
Spec: WriteAttrSpec<'a>,
Spec::ReadSpec: ReadAttrSpec<Spec::ReadNames>,
{
let values = spec.into_values();
async move {
let backend = self.ensure_backend().await?;
let res = backend.write_read_attr(values).await;
Spec::ReadSpec::collect(self.check_backend(res)?)
}
}
pub fn write_read_attr<'a, Spec>(&self, spec: Spec) -> Result<<Spec::ReadSpec as ReadAttrSpec<Spec::ReadNames>>::Output>
where
Spec: WriteAttrSpec<'a>,
Spec::ReadSpec: ReadAttrSpec<Spec::ReadNames>,
{
self.block_on(self.write_read_attr_async(spec))?
}
#[instrument(skip(self, spec), fields(target = %self.inner.target))]
pub fn command_info_async<Spec: ToNamesSpec<CommandInfo>>(&self, spec: Spec) -> impl Future<Output = Result<Spec::Output>> + Send {
let names = spec.to_names();
async move {
let backend = self.ensure_backend().await?;
let res = backend.command_info(names).await;
Spec::collect(self.check_backend(res)?)
}
}
pub fn command_info<Spec: ToNamesSpec<CommandInfo>>(&self, spec: Spec) -> Result<Spec::Output> {
self.block_on(self.command_info_async(spec))?
}
#[instrument(skip(self, spec), fields(target = %self.inner.target))]
pub fn attr_info_async<Spec: ToNamesSpec<AttrInfo>>(&self, spec: Spec) -> impl Future<Output = Result<Spec::Output>> + Send {
let names = spec.to_names();
async move {
let backend = self.ensure_backend().await?;
let res = backend.attr_info(names).await;
Spec::collect(self.check_backend(res)?)
}
}
pub fn attr_info<Spec: ToNamesSpec<AttrInfo>>(&self, spec: Spec) -> Result<Spec::Output> {
self.block_on(self.attr_info_async(spec))?
}
#[instrument(skip(self, values), fields(target = %self.inner.target))]
pub async fn set_attr_config_async(&self, values: &[(&str, AttrConfig)]) -> Result<Vec<Result<()>>> {
let configs = values.iter()
.map(|(name, config)| Named::new(Scalar::from(*name), config.clone()))
.collect();
let backend = self.ensure_backend().await?;
let res = backend.set_attr_config(configs).await;
self.check_backend(res).map_err(write_attr_err)
}
pub fn set_attr_config(&self, values: &[(&str, AttrConfig)]) -> Result<Vec<Result<()>>> {
self.block_on(self.set_attr_config_async(values))?
}
#[instrument(skip(self), fields(target = %self.inner.target))]
pub async fn command_history_async<T: FromValue>(&self, name: &str, max: usize) -> Result<Vec<(SystemTime, Result<T>)>> {
let backend = self.ensure_backend().await?;
let res = backend.command_history(name, max).await;
Ok(self.check_backend(res).map_err(command_err)?.into_iter()
.map(|(time, value)| (time, value.and_then(T::from_value)))
.collect())
}
pub fn command_history<T: FromValue>(&self, name: &str, max: usize) -> Result<Vec<(SystemTime, Result<T>)>> {
self.block_on(self.command_history_async(name, max))?
}
#[instrument(skip(self), fields(target = %self.inner.target))]
pub async fn attr_history_async<T: FromValue>(&self, name: &str, max: usize) -> Result<Vec<Result<AttrRead<T>>>> {
let backend = self.ensure_backend().await?;
let res = backend.attr_history(name, max).await;
Ok(self.check_backend(res)?.into_iter()
.map(|read| read.and_then(AttrRead::parse))
.collect())
}
pub fn attr_history<T: FromValue>(&self, name: &str, max: usize) -> Result<Vec<Result<AttrRead<T>>>> {
self.block_on(self.attr_history_async(name, max))?
}
pub async fn list_prop_async(&self, filter: &str) -> Result<Spectrum<str>> {
let dev_name = self.name_async().await?;
self.database()?.list_dev_prop_async(&dev_name, filter).await
}
pub fn list_prop(&self, filter: &str) -> Result<Spectrum<str>> {
self.block_on(self.list_prop_async(filter))?
}
pub fn get_prop_async<Spec, Names: ToNames>(&self, names: Names) -> impl Future<Output = Result<Spec::Output>> + Send
where
Spec: GetPropSpec<Names>,
{
let names = names.to_names();
async move {
let dev_name = self.name_async().await?;
let params = names.map(|names| prop_params(&dev_name, names.iter()));
let mut reply = self.database()?.get_raw_prop_async(PropCommands::DEV, &dev_name, params).await?;
Spec::collect(&mut reply)
}
}
pub fn get_prop<Spec, Names: ToNames>(&self, names: Names) -> Result<Spec::Output>
where
Spec: GetPropSpec<Names>,
{
self.block_on(self.get_prop_async::<Spec, Names>(names))?
}
pub fn set_prop_async<'a, Spec>(&self, spec: Spec) -> impl Future<Output = Result<()>> + Send
where
Spec: SetPropSpec<'a>,
{
let params = set_prop_params(spec);
async move {
let dev_name = self.name_async().await?;
self.database()?.prop_command_async("DbPutDeviceProperty", prop_params(&dev_name, params.iter().map(|param| &**param))).await
}
}
pub fn set_prop<'a, Spec>(&self, spec: Spec) -> Result<()>
where
Spec: SetPropSpec<'a>,
{
self.block_on(self.set_prop_async(spec))?
}
pub fn del_prop_async<Names: ToNamesNotAll>(&self, names: Names) -> impl Future<Output = Result<()>> + Send {
let names = names.to_names_not_all();
async move {
let dev_name = self.name_async().await?;
self.database()?.prop_command_async("DbDeleteDeviceProperty", prop_params(&dev_name, names.iter())).await
}
}
pub fn del_prop<Names: ToNamesNotAll>(&self, names: Names) -> Result<()> {
self.block_on(self.del_prop_async(names))?
}
pub fn subscribe_attr_value<T: FromValue>(&self, name: &str, event: AttrValueEvent) -> DevChannel<AttrRead<T>> {
DevChannel::new(self.clone(), name, ChannelKind::Value(event), ChannelEvents::poll_value)
}
pub fn subscribe_attr_ready(&self, name: &str) -> DevChannel<()> {
DevChannel::new(self.clone(), name, ChannelKind::Ready, ChannelEvents::poll_ready)
}
pub fn subscribe_attr_conf(&self, name: &str) -> DevChannel<AttrInfo> {
DevChannel::new(self.clone(), name, ChannelKind::Conf, ChannelEvents::poll_conf)
}
}
impl Clone for DevProxy {
fn clone(&self) -> Self {
Self {
inner: Box::new(ProxyInner {
backend: self.inner.backend.clone(),
target: self.inner.target.clone(),
source: self.inner.source,
timeout: self.inner.timeout,
}),
state: self.state.try_read().ok()
.map(|guard| RwLock::new((*guard).clone()))
.unwrap_or_else(|| RwLock::new(ProxyState {
backend: None,
database_host_and_port: None,
})),
}
}
}
impl PartialEq for DevProxy {
#[inline]
fn eq(&self, other: &Self) -> bool {
Arc::ptr_eq(&self.inner.backend, &other.inner.backend) && self.inner.target == other.inner.target
}
}
impl Eq for DevProxy { }
impl Hash for DevProxy {
#[inline]
fn hash<H: Hasher>(&self, state: &mut H) {
Arc::as_ptr(&self.inner.backend).hash(state);
self.inner.target.hash(state);
}
}
impl fmt::Display for ProxyTarget {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
ProxyTarget::DatabaseFromEnv =>
write!(f, "db"),
ProxyTarget::DatabaseAt { host, port } =>
write!(f, "db@{host}:{port}"),
ProxyTarget::RegularFromDatabase { name, .. } =>
write!(f, "{name}"),
ProxyTarget::RegularAt { host, port, name } =>
write!(f, "{name}@{host}:{port}"),
}
}
}
pub trait ToNames: Sized {
fn iter_names(&self) -> Option<impl Iterator<Item = &str>>;
#[inline]
fn to_names(self) -> Option<Spectrum<str>> {
self.iter_names().map(names_buf)
}
}
fn names_buf<'a>(names: impl Iterator<Item = &'a str>) -> Spectrum<str> {
Spectrum::from_strs(names).expect("names of more than 4 GiB")
}
pub trait ToNamesNotAll: ToNames {
#[inline]
fn iter_names_not_all(&self) -> impl Iterator<Item = &str> {
self.iter_names().expect("should not return the 'all' marker")
}
#[inline]
fn to_names_not_all(self) -> Spectrum<str> {
self.to_names().expect("should not return the 'all' marker")
}
}
#[derive(Debug, Clone, PartialEq)]
pub struct Named<T> {
pub name: Scalar<str>,
pub value: T,
}
impl<T> Named<T> {
#[inline]
pub fn new(name: Scalar<str>, value: T) -> Self {
Self { name, value }
}
#[inline]
pub fn map<U>(self, f: impl FnOnce(T) -> U) -> Named<U> {
Named { name: self.name, value: f(self.value) }
}
}
pub type NamedResult<T> = Named<Result<T>>;
pub struct All;
impl ToNames for All {
#[inline]
fn iter_names(&self) -> Option<impl Iterator<Item = &str>> {
None::<std::iter::Empty<&str>>
}
}
impl ToNames for &str {
#[inline]
fn iter_names(&self) -> Option<impl Iterator<Item = &str>> {
Some(std::iter::once(*self))
}
}
impl ToNamesNotAll for &str { }
impl ToNames for &[&str] {
#[inline]
fn iter_names(&self) -> Option<impl Iterator<Item = &str>> {
Some(self.iter().copied())
}
}
impl ToNamesNotAll for &[&str] { }
impl<const N: usize> ToNames for [&str; N] {
#[inline]
fn iter_names(&self) -> Option<impl Iterator<Item = &str>> {
Some(self.iter().copied())
}
}
impl<const N: usize> ToNamesNotAll for [&str; N] { }
impl ToNames for Spectrum<str> {
#[inline]
fn iter_names(&self) -> Option<impl Iterator<Item = &str>> {
Some(self.iter())
}
#[inline]
fn to_names(self) -> Option<Spectrum<str>> {
Some(self)
}
}
impl ToNamesNotAll for Spectrum<str> { }
impl ToNames for &Spectrum<str> {
#[inline]
fn iter_names(&self) -> Option<impl Iterator<Item = &str>> {
Some(self.iter())
}
#[inline]
fn to_names(self) -> Option<Spectrum<str>> {
Some(self.clone())
}
}
impl ToNamesNotAll for &Spectrum<str> { }
macro_rules! impl_to_names_tuple {
( @name: $comp:ident ) => { &str };
( @one: $comp:ident ) => { 1 };
( $( $comp:ident )* ) => {
#[allow(unused_variables)]
impl ToNames for ( $(impl_to_names_tuple!(@name: $comp),)* ) {
#[inline]
fn iter_names(&self) -> Option<impl Iterator<Item = &str>> {
let ( $($comp,)* ) = *self;
Some(<[&str; 0 $( + impl_to_names_tuple!(@one: $comp) )*]>::into_iter([ $($comp,)* ]))
}
}
impl ToNamesNotAll for ( $(impl_to_names_tuple!(@name: $comp),)* ) { }
};
}
impl_to_names_tuple!();
impl_to_names_tuple!(a);
impl_to_names_tuple!(a b);
impl_to_names_tuple!(a b c);
impl_to_names_tuple!(a b c d);
impl_to_names_tuple!(a b c d e);
impl_to_names_tuple!(a b c d e f);
pub trait ToNamesSpec<T>: ToNames {
type Output;
fn collect(entries: Vec<NamedResult<T>>) -> Result<Self::Output>;
}
impl<T> ToNamesSpec<T> for &str {
type Output = T;
fn collect(entries: Vec<NamedResult<T>>) -> Result<Self::Output> {
match entries.into_iter().next() {
Some(entry) => entry.value,
None => Err(unexpected_len_err(1)),
}
}
}
impl<T> ToNamesSpec<T> for &[&str] {
type Output = Vec<Result<T>>;
fn collect(entries: Vec<NamedResult<T>>) -> Result<Self::Output> {
Ok(entries.into_iter().map(|entry| entry.value).collect())
}
}
impl<T> ToNamesSpec<T> for Spectrum<str> {
type Output = Vec<Result<T>>;
fn collect(entries: Vec<NamedResult<T>>) -> Result<Self::Output> {
<&[&str] as ToNamesSpec<T>>::collect(entries)
}
}
impl<T> ToNamesSpec<T> for &Spectrum<str> {
type Output = Vec<Result<T>>;
fn collect(entries: Vec<NamedResult<T>>) -> Result<Self::Output> {
<&[&str] as ToNamesSpec<T>>::collect(entries)
}
}
impl<T, const N: usize> ToNamesSpec<T> for [&str; N] {
type Output = [Result<T>; N];
fn collect(entries: Vec<NamedResult<T>>) -> Result<Self::Output> {
take_array(entries.into_iter().map(|entry| entry.value))
}
}
impl<T> ToNamesSpec<T> for All {
type Output = Vec<NamedResult<T>>;
fn collect(entries: Vec<NamedResult<T>>) -> Result<Self::Output> {
Ok(entries)
}
}
pub(crate) fn take_array<T, const N: usize>(mut entries: impl Iterator<Item = T>) -> Result<[T; N]> {
let mut slots = [const { None }; N];
for slot in &mut slots {
*slot = Some(entries.next().ok_or_else(|| unexpected_len_err(N))?);
}
if entries.next().is_some() {
return Err(unexpected_len_err(N));
}
Ok(slots.map(|slot| slot.expect("every slot is filled")))
}
macro_rules! impl_to_names_spec_tuple {
( @one: $index:tt ) => { 1 };
( @name: $index:tt ) => { &str };
( @t: $index:tt ) => { T };
( @next: $index:tt, $entries:ident ) => { $entries.next().ok_or_else(|| unexpected_len_err(LEN))?.value };
( $( $index:tt )* ) => {
impl<T> ToNamesSpec<T> for ( $( impl_to_names_spec_tuple!(@name: $index), )* ) {
type Output = ( $( Result< impl_to_names_spec_tuple!(@t: $index) >,)* );
fn collect(entries: Vec<NamedResult<T>>) -> Result<Self::Output> {
const LEN: usize = 0 $( + impl_to_names_spec_tuple!(@one: $index) )*;
let mut entries = entries.into_iter();
let output = ( $( impl_to_names_spec_tuple!(@next: $index, entries), )* );
if entries.next().is_some() {
return Err(unexpected_len_err(LEN));
}
Ok(output)
}
}
};
}
impl_to_names_spec_tuple!(0);
impl_to_names_spec_tuple!(0 1);
impl_to_names_spec_tuple!(0 1 2);
impl_to_names_spec_tuple!(0 1 2 3);
impl_to_names_spec_tuple!(0 1 2 3 4);
impl_to_names_spec_tuple!(0 1 2 3 4 5);
pub trait ReadAttrSpec<Names> {
type Output;
fn collect(reads: Vec<NamedResult<AttrRead<Value>>>) -> Result<Self::Output>;
}
impl<'name, T: FromValue> ReadAttrSpec<&'name str> for T {
type Output = AttrRead<T>;
fn collect(reads: Vec<NamedResult<AttrRead<Value>>>) -> Result<Self::Output> {
match reads.into_iter().next() {
Some(entry) => entry.value?.parse(),
None => Err(unexpected_len_err(1)),
}
}
}
impl<'name, T: FromValue> ReadAttrSpec<&'_ [&'name str]> for T {
type Output = Vec<Result<AttrRead<T>>>;
fn collect(reads: Vec<NamedResult<AttrRead<Value>>>) -> Result<Self::Output> {
Ok(reads.into_iter().map(|entry| entry.value.and_then(AttrRead::parse)).collect())
}
}
impl<T: FromValue> ReadAttrSpec<Spectrum<str>> for T {
type Output = Vec<Result<AttrRead<T>>>;
fn collect(reads: Vec<NamedResult<AttrRead<Value>>>) -> Result<Self::Output> {
<T as ReadAttrSpec<&[&str]>>::collect(reads)
}
}
impl<T: FromValue> ReadAttrSpec<&Spectrum<str>> for T {
type Output = Vec<Result<AttrRead<T>>>;
fn collect(reads: Vec<NamedResult<AttrRead<Value>>>) -> Result<Self::Output> {
<T as ReadAttrSpec<&[&str]>>::collect(reads)
}
}
impl<'name, T: FromValue, const N: usize> ReadAttrSpec<[&'name str; N]> for T {
type Output = [Result<AttrRead<T>>; N];
fn collect(reads: Vec<NamedResult<AttrRead<Value>>>) -> Result<Self::Output> {
take_array(reads.into_iter().map(|entry| entry.value.and_then(AttrRead::parse)))
}
}
impl<T: FromValue> ReadAttrSpec<All> for T {
type Output = Vec<NamedResult<AttrRead<T>>>;
fn collect(reads: Vec<NamedResult<AttrRead<Value>>>) -> Result<Self::Output> {
Ok(reads.into_iter().map(|entry| entry.map(|read| read.and_then(AttrRead::parse))).collect())
}
}
macro_rules! impl_read_attr_spec_tuple {
( @one: $comp:ident ) => { 1 };
( @name: $comp:ident ) => { &'name str };
( $( $comp:ident )* ) => {
impl<'name, $($comp,)* > ReadAttrSpec<( $( impl_read_attr_spec_tuple!(@name: $comp), )* )> for ( $($comp,)* )
where
$( $comp: FromValue, )*
{
type Output = ( $( Result<AttrRead<$comp>>,)* );
fn collect(reads: Vec<NamedResult<AttrRead<Value>>>) -> Result<Self::Output> {
const LEN: usize = 0 $( + impl_read_attr_spec_tuple!(@one: $comp) )*;
let mut reads = reads.into_iter();
let output = ( $( reads.next().ok_or_else(|| unexpected_len_err(LEN))?.value.and_then(AttrRead::parse::<$comp>), )* );
if reads.next().is_some() {
return Err(unexpected_len_err(LEN));
}
Ok(output)
}
}
};
}
impl_read_attr_spec_tuple!(A);
impl_read_attr_spec_tuple!(A B);
impl_read_attr_spec_tuple!(A B C);
impl_read_attr_spec_tuple!(A B C D);
impl_read_attr_spec_tuple!(A B C D E);
impl_read_attr_spec_tuple!(A B C D E F);
pub trait WriteAttrSpec<'a>: Sized {
fn into_values(self) -> Vec<Named<Value>>;
type Output;
fn collect(writes: Vec<Result<()>>) -> Result<Self::Output>;
type ReadNames;
type ReadSpec;
}
impl<'a, T: IntoValue> WriteAttrSpec<'a> for (&'a str, T) {
#[inline]
fn into_values(self) -> Vec<Named<Value>> {
vec![Named::new(Scalar::from(self.0), self.1.into_value())]
}
type Output = ();
fn collect(writes: Vec<Result<()>>) -> Result<Self::Output> {
match writes.into_iter().next() {
Some(res) => res,
None => Err(unexpected_len_err(1)),
}
}
type ReadNames = &'a str;
type ReadSpec = T;
}
impl<'a, T: IntoValue, const N: usize> WriteAttrSpec<'a> for [(&'a str, T); N] {
#[inline]
fn into_values(self) -> Vec<Named<Value>> {
named_values(self.into_iter().map(|(name, value)| (name, value.into_value())).collect())
}
type Output = [Result<()>; N];
fn collect(writes: Vec<Result<()>>) -> Result<Self::Output> {
take_array(writes.into_iter())
}
type ReadNames = [&'a str; N];
type ReadSpec = T;
}
impl<'a, T: IntoValue> WriteAttrSpec<'a> for Vec<(&'a str, T)> {
#[inline]
fn into_values(self) -> Vec<Named<Value>> {
named_values(self.into_iter().map(|(name, value)| (name, value.into_value())).collect())
}
type Output = Vec<Result<()>>;
fn collect(writes: Vec<Result<()>>) -> Result<Self::Output> {
Ok(writes)
}
type ReadNames = &'a [&'a str];
type ReadSpec = T;
}
fn named_values(values: Vec<(&str, Value)>) -> Vec<Named<Value>> {
let names = names_buf(values.iter().map(|&(name, _)| name));
values.into_iter()
.enumerate()
.map(|(index, (_, value))| Named::new(names.get_scalar([index]).expect("as many names as values"), value))
.collect()
}
macro_rules! impl_write_attr_spec_tuple {
( @name: $comp:ident ) => { &'a str };
( @one: $comp:ident ) => { 1 };
( @unit: $comp:ident ) => { Result<()> };
( @next: $comp:ident, $writes:ident ) => { $writes.next().ok_or_else(|| unexpected_len_err(LEN))? };
( $( $comp:ident )* ) => {
#[allow(non_snake_case)]
impl<'a, $($comp: IntoValue, )*> WriteAttrSpec<'a> for ( $( (&'a str, $comp), )* ) {
#[inline]
fn into_values(self) -> Vec<Named<Value>> {
let ( $($comp,)* ) = self;
named_values(vec![$(($comp.0, $comp.1.into_value()), )* ])
}
type Output = ( $( impl_write_attr_spec_tuple!(@unit: $comp), )* );
fn collect(writes: Vec<Result<()>>) -> Result<Self::Output> {
const LEN: usize = 0 $( + impl_write_attr_spec_tuple!(@one: $comp) )*;
let mut writes = writes.into_iter();
let output = ( $( impl_write_attr_spec_tuple!(@next: $comp, writes), )* );
if writes.next().is_some() {
return Err(unexpected_len_err(LEN));
}
Ok(output)
}
type ReadNames = ( $(impl_write_attr_spec_tuple!(@name: $comp), )* );
type ReadSpec = ( $($comp, )* );
}
};
}
impl_write_attr_spec_tuple!(A);
impl_write_attr_spec_tuple!(A B);
impl_write_attr_spec_tuple!(A B C);
impl_write_attr_spec_tuple!(A B C D);
impl_write_attr_spec_tuple!(A B C D E);
impl_write_attr_spec_tuple!(A B C D E F);
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CommandInfo {
pub display_level: DisplayLevel,
pub in_kind: CommandDataKind,
pub out_kind: CommandDataKind,
pub in_doc: Option<Cow<'static, str>>,
pub out_doc: Option<Cow<'static, str>>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct AttrRead<T> {
pub value: Option<T>,
pub w_value: Option<T>,
pub quality: AttrQuality,
pub time: SystemTime,
}
impl<T> AttrRead<T> {
pub fn new(value: T) -> Self {
Self {
value: Some(value),
w_value: None,
quality: AttrQuality::Valid,
time: SystemTime::now(),
}
}
pub fn with_quality(mut self, quality: AttrQuality) -> Self {
self.quality = quality;
self
}
pub fn with_time(mut self, time: SystemTime) -> Self {
self.time = time;
self
}
}
impl AttrRead<Value> {
fn parse<T: FromValue>(self) -> Result<AttrRead<T>> {
Ok(AttrRead {
value: match self.value {
Some(value) => Some(T::from_value(value)
.map_err(|err| err.with_append_description("Failed to parse read value"))?),
None => None,
},
w_value: match self.w_value {
Some(value) => Some(T::from_value(value)
.map_err(|err| err.with_append_description("Failed to parse write value"))?),
None => None,
},
quality: self.quality,
time: self.time,
})
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum AttrQuality {
Valid,
Invalid,
Alarm,
Changing,
Warning,
}
#[derive(Debug, Clone, PartialEq)]
pub struct AttrInfo {
pub config: AttrConfig,
pub mode: AttrMode,
pub data_kind: AttrDataKind,
pub data_format: AttrDataFormat,
pub display_level: DisplayLevel,
pub memorized: Option<AttrMemorized>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum AttrMode {
Read,
ReadWithWrite {
write_attr_name: Cow<'static, str>,
},
Write,
ReadWrite,
}
#[derive(Debug, Clone, PartialEq, Default)]
pub struct AttrConfig {
pub doc: Option<Cow<'static, str>>,
pub label: Option<Cow<'static, str>>,
pub unit: Option<Cow<'static, str>>,
pub standard_unit: Option<f64>,
pub display_unit: Option<Cow<'static, str>>,
pub display_format: Option<Cow<'static, str>>,
pub range: AttrRangeConfig,
pub alarm: AttrRangeConfig,
pub warning: AttrRangeConfig,
pub delta: Option<AttrDeltaConfig>,
pub change_event: AttrChangeEventConfig,
pub periodic_event: AttrPeriodicEventConfig,
pub archive_event: AttrArchiveEventConfig,
pub enum_labels: Vec<Cow<'static, str>>,
pub extensions: Vec<Cow<'static, str>>,
}
#[derive(Debug, Clone, Copy)]
pub enum AttrNumber {
Double(f64),
Long(i64),
ULong(u64),
}
impl AttrNumber {
pub fn as_f64(self) -> f64 {
match self {
Self::Double(value) => value,
Self::Long(value) => value as f64,
Self::ULong(value) => value as f64,
}
}
}
impl PartialEq for AttrNumber {
#[inline]
fn eq(&self, other: &Self) -> bool {
self.partial_cmp(other) == Some(Ordering::Equal)
}
}
impl PartialOrd for AttrNumber {
fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
match (*self, *other) {
(Self::Long(a), Self::Long(b)) => Some(a.cmp(&b)),
(Self::ULong(a), Self::ULong(b)) => Some(a.cmp(&b)),
(Self::Long(a), Self::ULong(b)) => Some(i128::from(a).cmp(&i128::from(b))),
(Self::ULong(a), Self::Long(b)) => Some(i128::from(a).cmp(&i128::from(b))),
(a, b) => a.as_f64().partial_cmp(&b.as_f64()),
}
}
}
impl fmt::Display for AttrNumber {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::Double(value) => Display::fmt(value, f),
Self::Long(value) => Display::fmt(value, f),
Self::ULong(value) => Display::fmt(value, f),
}
}
}
macro_rules! impl_attr_number_from {
($variant:ident: $($ty:ty),* $(,)?) => { $(
impl From<$ty> for AttrNumber {
#[inline]
fn from(value: $ty) -> Self {
Self::$variant(value.into())
}
}
)* };
}
impl_attr_number_from!(Double: f32, f64);
impl_attr_number_from!(Long: i8, i16, i32, i64);
impl_attr_number_from!(ULong: u8, u16, u32, u64);
#[derive(Debug, Clone, Copy, PartialEq, Default)]
pub struct AttrRangeConfig {
pub min: Option<AttrNumber>,
pub max: Option<AttrNumber>,
}
impl AttrRangeConfig {
#[inline]
pub fn is_set(&self) -> bool {
self.min.is_some() || self.max.is_some()
}
}
#[derive(Debug, Clone, Copy, PartialEq)]
pub struct AttrDeltaConfig {
pub value: AttrNumber,
pub time: Option<Duration>,
}
#[derive(Debug, Clone, Copy, PartialEq, Default)]
pub struct AttrChangeEventConfig {
pub rel_change: Option<f64>,
pub abs_change: Option<f64>,
}
impl AttrChangeEventConfig {
#[inline]
pub fn is_set(&self) -> bool {
self.rel_change.is_some() || self.abs_change.is_some()
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub struct AttrPeriodicEventConfig {
pub period: Option<Duration>,
}
impl AttrPeriodicEventConfig {
#[inline]
pub fn is_set(&self) -> bool {
self.period.is_some()
}
}
#[derive(Debug, Clone, Copy, PartialEq, Default)]
pub struct AttrArchiveEventConfig {
pub rel_change: Option<f64>,
pub abs_change: Option<f64>,
pub period: Option<Duration>,
}
impl AttrArchiveEventConfig {
#[inline]
pub fn is_set(&self) -> bool {
self.rel_change.is_some() || self.abs_change.is_some() || self.period.is_some()
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum AttrMemorized {
RestoreSetPoint,
RestoreAndWrite,
}
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct DevInfo {
pub class: String,
pub server_id: String,
pub server_host: String,
pub server_version: i32,
pub doc_url: Option<Cow<'static, str>>,
pub dev_type: Option<Cow<'static, str>>,
pub version_info: Vec<(Cow<'static, str>, Cow<'static, str>)>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
#[repr(u8)]
pub enum DevState {
On,
Off,
Close,
Open,
Insert,
Extract,
Moving,
Standby,
Fault,
Init,
Running,
Alarm,
Disable,
Unknown,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum DevSource {
Dev,
Cache,
CacheDev,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Default)]
pub enum DisplayLevel {
#[default]
Operator,
Expert,
}
enum ChannelEvents {
Value(BoxStream<'static, Result<AttrRead<Value>>>),
Ready(BoxStream<'static, Result<()>>),
Conf(BoxStream<'static, Result<AttrInfo>>),
}
impl ChannelEvents {
fn poll_value<T: FromValue>(&mut self, cx: &mut Context<'_>) -> Poll<Option<Result<AttrRead<T>>>> {
let Self::Value(stream) = self else { unreachable!("a value channel subscribes to values") };
stream.poll_next_unpin(cx).map(|item| item.map(|read| read.and_then(AttrRead::parse)))
}
fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Option<Result<()>>> {
let Self::Ready(stream) = self else { unreachable!("a ready channel subscribes to readiness") };
stream.poll_next_unpin(cx)
}
fn poll_conf(&mut self, cx: &mut Context<'_>) -> Poll<Option<Result<AttrInfo>>> {
let Self::Conf(stream) = self else { unreachable!("a configuration channel subscribes to configurations") };
stream.poll_next_unpin(cx)
}
}
#[derive(Debug, Clone, Copy)]
enum ChannelKind {
Value(AttrValueEvent),
Ready,
Conf,
}
pub struct DevChannel<T> {
dev: Option<DevProxy>,
name: Arc<str>,
kind: ChannelKind,
poll_events: fn(&mut ChannelEvents, &mut Context<'_>) -> Poll<Option<Result<T>>>,
timeout: Duration,
state: ChannelState,
}
enum ChannelState {
Subscribed(ChannelEvents),
Subscribing(BoxFuture<'static, (DevProxy, Result<ChannelEvents>)>),
}
impl ChannelState {
fn subscribing(dev: DevProxy, name: Arc<str>, kind: ChannelKind) -> Self {
Self::Subscribing(Box::pin(async move {
let res = async {
let backend = dev.ensure_backend().await?;
let admin = dev.admin_dev_async().await?;
let admin = admin.ensure_backend().await?;
let res = match kind {
ChannelKind::Value(event) => backend.subscribe_attr_value(admin, &name, event).await.map(ChannelEvents::Value),
ChannelKind::Ready => backend.subscribe_attr_ready(admin, &name).await.map(ChannelEvents::Ready),
ChannelKind::Conf => backend.subscribe_attr_conf(admin, &name).await.map(ChannelEvents::Conf),
};
dev.check_backend(res)
}.await;
(dev, res)
}))
}
}
impl<T> DevChannel<T> {
fn new(
dev: DevProxy,
name: &str,
kind: ChannelKind,
poll_events: fn(&mut ChannelEvents, &mut Context<'_>) -> Poll<Option<Result<T>>>,
) -> Self {
let timeout = dev.timeout();
let name = Arc::<str>::from(name);
let state = ChannelState::subscribing(dev, Arc::clone(&name), kind);
Self { dev: None, name, kind, poll_events, timeout, state }
}
fn resubscribe(&mut self) {
let dev = self.dev.take().expect("the device is moved into one attempt at a time");
self.state = ChannelState::subscribing(dev, Arc::clone(&self.name), self.kind);
}
fn poll_subscribed(&mut self, cx: &mut Context<'_>) -> Poll<Result<()>> {
let ChannelState::Subscribing(attempt) = &mut self.state else {
return Poll::Ready(Ok(()));
};
let (dev, res) = ready!(attempt.as_mut().poll(cx));
let dev = &*self.dev.insert(dev);
Poll::Ready(match res {
Ok(events) => {
debug!(dev = %dev.inner.target, attr = %self.name, kind = ?self.kind, "Subscribed to the events");
self.state = ChannelState::Subscribed(events);
Ok(())
}
Err(err) => {
debug!(dev = %dev.inner.target, attr = %self.name, kind = ?self.kind, %err, "Failed to subscribe to the events, retrying on the next poll");
self.resubscribe();
Err(err)
}
})
}
pub async fn ensure_subscribed_async(&mut self) -> Result<()> {
poll_fn(|cx| self.poll_subscribed(cx)).await
}
pub fn ensure_subscribed(&mut self) -> Result<()> {
let timeout = self.timeout;
block_on_timeout(self.ensure_subscribed_async(), timeout)?
}
pub fn wait(&mut self, timeout: Duration) -> Result<T> {
match block_on_timeout(self.next(), timeout)? {
Some(item) => item,
None => unreachable!("a device channel never ends"),
}
}
}
impl<T> Stream for DevChannel<T> {
type Item = Result<T>;
fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
let this = self.get_mut();
loop {
if let Err(err) = ready!(this.poll_subscribed(cx)) {
return Poll::Ready(Some(Err(err)));
}
let ChannelState::Subscribed(events) = &mut this.state else {
unreachable!("subscribed once the attempt succeeded");
};
match ready!((this.poll_events)(events, cx)) {
Some(Err(err)) if err.kind() == ErrorKind::DevNotConnected => {
debug!(attr = %this.name, kind = ?this.kind, %err, "Lost the subscription to the events, subscribing again on the next poll");
if let Some(dev) = &this.dev {
dev.reset_backend();
}
this.resubscribe();
return Poll::Ready(Some(Err(err)));
}
Some(item) => return Poll::Ready(Some(item)),
None => {
debug!(attr = %this.name, kind = ?this.kind, "The subscription to the events ended, subscribing again");
this.resubscribe();
}
}
}
}
}
impl<T> Debug for DevChannel<T> {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("DevChannel")
.field("dev", &self.dev)
.field("name", &self.name)
.finish_non_exhaustive()
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum AttrValueEvent {
Change,
Periodic,
Archive,
User,
Alarm,
}
#[cold]
pub(crate) fn unexpected_len_err(expected: usize) -> Error {
Error::new(ErrorKind::Internal)
.with_description("The device did not answer with as many entries as were requested")
.with_append_description(format_args!("{expected} were expected"))
}
fn command_err(err: Error) -> Error {
match err.kind() {
ErrorKind::Reason(reason::API_CommandNotFound) => {
err.with_kind(ErrorKind::CommandNotFound)
}
ErrorKind::Reason(reason::API_IncompatibleCmdArgumentType) => {
let expected_kind = expected_cpp_tango_type(err.description())
.and_then(command_data_kind_from_cpp_tango);
let err = err.with_kind(ErrorKind::CommandArgumentKindInvalid);
match expected_kind {
Some(kind) => err.with_append_description(format_args!("Expected kind: {kind:?}")),
None => err,
}
}
_ => err,
}
}
fn write_attr_err(err: Error) -> Error {
match err.kind() {
ErrorKind::Reason(reason::API_AttrNotFound) => {
err.with_kind(ErrorKind::AttrNotFound)
}
ErrorKind::Reason(reason::API_AttrNotWritable) => {
err.with_kind(ErrorKind::AttrNotWritable)
}
ErrorKind::Reason(reason::API_IncompatibleAttrDataType) => {
let expected_kind = expected_cpp_tango_type(err.description())
.and_then(attr_data_kind_from_cpp_tango);
let err = err.with_kind(ErrorKind::AttrKindInvalid);
match expected_kind {
Some(kind) => err.with_append_description(format_args!("Expected kind: {kind:?}")),
None => err,
}
}
_ => err,
}
}
fn expected_cpp_tango_type(description: &str) -> Option<&str> {
description.split_ascii_whitespace()
.find_map(|word| word.strip_prefix("Tango::"))
.map(|ty| ty.trim_matches(|c: char| !c.is_ascii_alphanumeric()))
}
fn command_data_kind_from_cpp_tango(cpp_type: &str) -> Option<CommandDataKind> {
Some(match cpp_type {
"DevVoid" => CommandDataKind::Void,
"DevBoolean" => CommandDataKind::Bool,
"DevUChar" => CommandDataKind::Octet,
"DevShort" => CommandDataKind::Short,
"DevUShort" => CommandDataKind::UShort,
"DevLong" => CommandDataKind::Long,
"DevULong" => CommandDataKind::ULong,
"DevLong64" => CommandDataKind::Long64,
"DevULong64" => CommandDataKind::ULong64,
"DevFloat" => CommandDataKind::Float,
"DevDouble" => CommandDataKind::Double,
"DevState" => CommandDataKind::State,
"DevString" |
"ConstDevString" => CommandDataKind::String,
"DevVarBooleanArray" => CommandDataKind::BoolArray,
"DevVarCharArray" => CommandDataKind::OctetArray,
"DevVarShortArray" => CommandDataKind::ShortArray,
"DevVarUShortArray" => CommandDataKind::UShortArray,
"DevVarLongArray" => CommandDataKind::LongArray,
"DevVarULongArray" => CommandDataKind::ULongArray,
"DevVarLong64Array" => CommandDataKind::Long64Array,
"DevVarULong64Array" => CommandDataKind::ULong64Array,
"DevVarFloatArray" => CommandDataKind::FloatArray,
"DevVarDoubleArray" => CommandDataKind::DoubleArray,
"DevVarStateArray" => CommandDataKind::StateArray,
"DevVarStringArray" => CommandDataKind::StringArray,
"DevVarLongStringArray" => CommandDataKind::LongStringArray,
"DevVarDoubleStringArray" => CommandDataKind::DoubleStringArray,
"DevEncoded" => CommandDataKind::Encoded,
_ => return None,
})
}
fn attr_data_kind_from_cpp_tango(cpp_type: &str) -> Option<AttrDataKind> {
Some(match cpp_type {
"DevBoolean" | "DevVarBooleanArray" => AttrDataKind::Bool,
"DevUChar" | "DevVarCharArray" => AttrDataKind::Octet,
"DevShort" | "DevVarShortArray" => AttrDataKind::Short,
"DevUShort" | "DevVarUShortArray" => AttrDataKind::UShort,
"DevLong" | "DevVarLongArray" => AttrDataKind::Long,
"DevULong" | "DevVarULongArray" => AttrDataKind::ULong,
"DevLong64" | "DevVarLong64Array" => AttrDataKind::Long64,
"DevULong64" | "DevVarULong64Array" => AttrDataKind::ULong64,
"DevFloat" | "DevVarFloatArray" => AttrDataKind::Float,
"DevDouble" | "DevVarDoubleArray" => AttrDataKind::Double,
"DevState" | "DevVarStateArray" => AttrDataKind::State,
"DevString" | "ConstDevString" | "DevVarStringArray" => AttrDataKind::String,
"DevEncoded" | "DevVarEncodedArray" => AttrDataKind::Encoded,
_ => return None,
})
}
struct DisplayHostAndPorts<'a>(&'a [(Arc<str>, u16)]);
impl Display for DisplayHostAndPorts<'_> {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
for (i, (host, port)) in self.0.iter().enumerate() {
if i != 0 {
f.write_str(", ")?;
}
write!(f, "{host}:{port}")?;
}
Ok(())
}
}
fn env_databases() -> &'static [(Arc<str>, u16)] {
static CACHE: OnceLock<Box<[(Arc<str>, u16)]>> = OnceLock::new();
&**CACHE.get_or_init(|| {
let mut ret = Vec::new();
if let Ok(var) = env::var("TANGO_HOST") {
for part in var.split(',') {
if let Some((host, port_raw)) = part.rsplit_once(':')
&& let Ok(port) = u16::from_str(port_raw) {
ret.push((host.into(), port));
}
}
}
ret.into_boxed_slice()
})
}
#[cfg(test)]
mod tests {
use crate::value::Scalar;
use super::*;
#[allow(dead_code)]
fn props_in_send_future(proxy: &DevProxy) -> BoxFuture<'_, Result<()>> {
Box::pin(async move {
let _speed = proxy.get_prop_async::<f64, _>("speed").await?;
let _all = proxy.get_prop_async::<Spectrum<str>, _>(All).await?;
proxy.set_prop_async(("speed", 2.5)).await?;
proxy.del_prop_async(["speed", "accel"]).await
})
}
#[allow(dead_code)]
fn requests_in_send_future(proxy: &DevProxy) -> BoxFuture<'_, Result<()>> {
Box::pin(async move {
let _: f64 = proxy.command_async("Add", 1.0).await?;
let _: String = proxy.command_async("Echo", "text").await?;
let _ = proxy.read_attr_async::<f64, _>("speed").await?;
let _ = proxy.read_attr_async::<f64, _>(["speed", "accel"]).await?;
let _ = proxy.read_attr_async::<Value, _>(All).await?;
proxy.write_attr_async(("speed", 2.5)).await?;
let _ = proxy.write_attr_async([("speed", 2.5), ("accel", 1.0)]).await?;
let _ = proxy.write_read_attr_async(("speed", 2.5)).await?;
let _ = proxy.command_info_async("Add").await?;
let _ = proxy.attr_info_async(["speed"]).await?;
let names = proxy.list_prop_async("*").await?;
let _ = proxy.read_attr_async::<f64, _>(&names).await?;
let _ = proxy.attr_info_async(names).await?;
Ok(())
})
}
#[allow(dead_code)]
fn props_of_any_names<'a, Names>(proxy: &'a DevProxy, names: Names) -> BoxFuture<'a, Result<<f64 as GetPropSpec<Names>>::Output>>
where
Names: ToNames + 'a,
f64: GetPropSpec<Names>,
{
Box::pin(proxy.get_prop_async::<f64, Names>(names))
}
#[test]
fn expected_kind() {
let cmd = "Incompatible command argument type, expected type is : Tango::DevVarDoubleStringArray\0";
assert_eq!(expected_cpp_tango_type(cmd), Some("DevVarDoubleStringArray"));
assert_eq!(
expected_cpp_tango_type(cmd).and_then(command_data_kind_from_cpp_tango),
Some(CommandDataKind::DoubleStringArray),
);
let attr = "Incompatible attribute type: expected Tango::DevVarDoubleArray (even for single value), found Tango::DevVarStringArray\0";
assert_eq!(expected_cpp_tango_type(attr), Some("DevVarDoubleArray"));
assert_eq!(
expected_cpp_tango_type(attr).and_then(attr_data_kind_from_cpp_tango),
Some(AttrDataKind::Double),
);
assert_eq!(attr_data_kind_from_cpp_tango("DevVarEncodedArray"), Some(AttrDataKind::Encoded));
assert_eq!(command_data_kind_from_cpp_tango("DevEncoded"), Some(CommandDataKind::Encoded));
assert_eq!(expected_cpp_tango_type("Attribute my_attr is not writable"), None);
assert_eq!(command_data_kind_from_cpp_tango("DevVarEncodedArray"), None);
}
fn reads<const N: usize>(entries: [(&str, Value); N]) -> Vec<NamedResult<AttrRead<Value>>> {
entries.into_iter()
.map(|(name, value)| Named::new(Scalar::from(name), Ok(AttrRead::new(value))))
.collect()
}
#[test]
fn single_reading() {
let reads = reads([("double_scalar", Value::from(2.5))]);
let read = <f64 as ReadAttrSpec<&str>>::collect(reads).unwrap();
assert_eq!(read.value, Some(2.5));
}
#[test]
fn readings_in_order() {
let reads = reads([("first", Value::from(1.0)), ("second", Value::from(2.0))]);
let [first, second] = <f64 as ReadAttrSpec<[&str; 2]>>::collect(reads).unwrap();
assert_eq!(first.unwrap().value, Some(1.0));
assert_eq!(second.unwrap().value, Some(2.0));
}
#[test]
fn readings_of_their_own_types() {
let reads = reads([("double_scalar", Value::from(2.5)), ("string_scalar", Value::from("hello"))]);
let (double, string) = <(f64, String) as ReadAttrSpec<(&str, &str)>>::collect(reads).unwrap();
assert_eq!(double.unwrap().value, Some(2.5));
assert_eq!(string.unwrap().value.as_deref(), Some("hello"));
}
#[test]
fn missing_reading() {
let reads = reads([("first", Value::from(1.0))]);
let err = <f64 as ReadAttrSpec<[&str; 2]>>::collect(reads).unwrap_err();
assert_eq!(err.kind(), ErrorKind::Internal);
}
#[test]
fn every_reading_named() {
let reads = reads([("first", Value::from(1.0)), ("second", Value::from(2.0))]);
let all = <Value as ReadAttrSpec<All>>::collect(reads).unwrap();
let names = all.iter().map(|read| read.name.value()).collect::<Vec<_>>();
assert_eq!(names, ["first", "second"]);
}
#[test]
fn write_outcomes() {
let writes = vec![Ok(()), Err(Error::new(ErrorKind::AttrNotWritable))];
let (first, second) = <((&str, f64), (&str, f64)) as WriteAttrSpec>::collect(writes).unwrap();
assert!(first.is_ok());
assert_eq!(second.unwrap_err().kind(), ErrorKind::AttrNotWritable);
}
#[test]
fn unexpected_outcome() {
let writes = vec![Ok(())];
let err = <((&str, f64), (&str, f64)) as WriteAttrSpec>::collect(writes).unwrap_err();
assert_eq!(err.kind(), ErrorKind::Internal);
}
}