1use std::sync::{Arc, OnceLock, RwLock};
4use std::time::{Duration, SystemTime};
5use std::fmt::{self, Debug, Display};
6use std::task::{Context, Poll};
7use std::hash::{Hash, Hasher};
8use std::future::Future;
9use std::cmp::Ordering;
10use std::str::FromStr;
11use std::borrow::Cow;
12use std::{env, mem};
13use std::pin::Pin;
14
15use futures::stream::{BoxStream, Stream, StreamExt};
16use futures::ready;
17use tracing::{debug, instrument, warn};
18use futures::future::{poll_fn, BoxFuture};
19
20use crate::value::{AttrDataFormat, AttrDataKind, CommandDataKind, FromValue, IntoValue, Scalar, Spectrum, Value};
21use crate::database::{prop_params, set_prop_params, DatabaseProxy, GetPropSpec, PropCommands, SetPropSpec};
22use crate::backend::{default_backend, Backend, BackendDev};
23use crate::error::{Error, ErrorKind, Result, reason};
24use crate::name::{ArcTrl, IntoTrl, TrlObject};
25use crate::future::block_on_timeout;
26use crate::local;
27
28
29const DEFAULT_TIMEOUT: Duration = Duration::from_secs(3);
31
32pub(crate) const DATABASE_DEV_NAME: &str = "database";
34
35#[derive(Debug)]
41pub struct DevProxy {
42 inner: Box<ProxyInner>,
44 state: RwLock<ProxyState>,
46}
47
48#[derive(Debug, Clone)]
49struct ProxyInner {
50 backend: Arc<dyn Backend>,
52 source: DevSource,
54 timeout: Duration,
56 target: ProxyTarget,
58}
59
60#[derive(Debug, Clone)]
61struct ProxyState {
62 backend: Option<Arc<dyn BackendDev>>,
64 database_host_and_port: Option<(Arc<str>, u16)>,
67}
68
69#[derive(Debug, Clone, PartialEq, Eq, Hash)]
73pub(crate) enum ProxyTarget {
74 DatabaseFromEnv,
76 DatabaseAt {
78 host: Arc<str>,
79 port: u16,
80 },
81 RegularFromDatabase {
83 database: DatabaseProxy,
84 name: Arc<str>,
85 },
86 RegularAt {
89 host: Arc<str>,
90 port: u16,
91 name: Arc<str>,
92 },
93}
94
95impl DevProxy {
96
97 pub fn new<'a>(trl: impl IntoTrl<'a>) -> Self {
99 Self::new_with_backend(default_backend(), trl)
100 }
101
102 pub fn try_new<'a>(trl: impl IntoTrl<'a>) -> Option<Self> {
106 Self::try_new_with_backend(default_backend(), trl)
107 }
108
109 pub fn new_from_database(database: &DatabaseProxy, name: &str) -> Self {
111 Self::new_from_database_with_backend(default_backend(), database, name)
112 }
113
114 pub fn new_with_backend<'a>(backend: Arc<dyn Backend>, trl: impl IntoTrl<'a>) -> Self {
117 Self::try_new_with_backend(backend, trl).expect("The given TRL does not target a device")
118 }
119
120 pub fn try_new_with_backend<'a>(backend: Arc<dyn Backend>, trl: impl IntoTrl<'a>) -> Option<Self> {
124
125 let trl = trl.into_trl().ok()?;
126
127 if trl.property().is_some() {
129 return None;
130 }
131
132 let name = match trl.object() {
133 TrlObject::Alias(name) => name,
134 TrlObject::Dev(name) => name,
135 TrlObject::Attr(_, _) => return None,
136 };
137
138 let target = match (trl.db(), trl.host_and_port()) {
139 (true, Some((host, port))) => ProxyTarget::RegularFromDatabase {
140 database: DatabaseProxy::new_with_backend(Arc::clone(&backend), host, port),
141 name: Arc::from(name),
142 },
143 (true, None) => ProxyTarget::RegularFromDatabase {
144 database: DatabaseProxy::new_from_env_with_backend(Arc::clone(&backend)),
145 name: Arc::from(name),
146 },
147 (false, Some((host, port))) => ProxyTarget::RegularAt {
148 host: Arc::from(host),
149 port,
150 name: Arc::from(name),
151 },
152 (false, None) => return None,
153 };
154
155 Some(Self::new_from_target_with_backend(backend, target))
156
157 }
158
159 pub fn new_from_database_with_backend(backend: Arc<dyn Backend>, database: &DatabaseProxy, name: &str) -> Self {
161 Self::new_from_target_with_backend(backend, ProxyTarget::RegularFromDatabase {
162 database: database.clone(),
163 name: Arc::from(name),
164 })
165 }
166
167 pub(crate) fn new_from_target_with_backend(backend: Arc<dyn Backend>, target: ProxyTarget) -> Self {
169 Self {
170 inner: Box::new(ProxyInner {
171 backend,
172 target,
173 source: DevSource::CacheDev,
174 timeout: DEFAULT_TIMEOUT,
175 }),
176 state: RwLock::new(ProxyState {
177 backend: None,
178 database_host_and_port: None,
179 }),
180 }
181 }
182
183 pub fn backend(&self) -> &Arc<dyn Backend> {
184 &self.inner.backend
185 }
186
187 pub fn timeout(&self) -> Duration {
189 self.inner.timeout
190 }
191
192 pub fn set_timeout(&mut self, timeout: Duration) -> Duration {
194 mem::replace(&mut self.inner.timeout, timeout)
195 }
196
197 pub fn source(&self) -> DevSource {
199 self.inner.source
200 }
201
202 pub fn set_source(&mut self, source: DevSource) -> DevSource {
204 mem::replace(&mut self.inner.source, source)
205 }
206
207 fn connect_backend(&self) -> BoxFuture<'_, Result<(Arc<dyn BackendDev>, Option<(Arc<str>, u16)>)>> {
212 Box::pin(async move {
213
214 match self.inner.target {
215 ProxyTarget::DatabaseFromEnv => {
216
217 debug!("Connecting to the database of TANGO_HOST");
218
219 let databases = env_databases();
220 if databases.is_empty() {
221 return Err(match env::var("TANGO_HOST") {
222 Ok(var) => Error::new(ErrorKind::TangoHostNotSet)
223 .with_description("TANGO_HOST is not a comma separated list of host:port")
224 .with_append_description(format_args!("Got: {var:?}")),
225 Err(_) => Error::new(ErrorKind::TangoHostNotSet)
226 .with_description("TANGO_HOST is not defined in the environment"),
227 });
228 }
229
230 let mut last_err = None;
231 for (host, port) in databases {
232
233 if let Some(dev_backend) = local::find_at(host, *port, DATABASE_DEV_NAME) {
234 debug!(%host, port, "Connected to the database, served by this process");
235 return Ok((dev_backend, Some((host.clone(), *port))));
236 }
237
238 match self.inner.backend.connect_at(host, *port, DATABASE_DEV_NAME).await {
239 Ok(dev_backend) => {
240 debug!(%host, port, "Connected to the database");
241 return Ok((dev_backend, Some((host.clone(), *port))));
242 }
243 Err(err) => {
244 warn!(%host, port, %err, "Failed to connect to the database, trying the next one");
245 last_err = Some(err);
246 }
247 }
248 }
249
250 let mut err = Error::new(ErrorKind::DatabaseNotConnected)
254 .with_description("Failed to connect to any database")
255 .with_append_description(format_args!("Tried, from TANGO_HOST: {}", DisplayHostAndPorts(databases)));
256 if let Some(last_err) = last_err {
257 err.set_parent(last_err);
258 }
259 Err(err)
260
261 }
262 ProxyTarget::DatabaseAt { ref host, port } => {
263
264 debug!(%host, port, "Connecting to the database");
265
266 if let Some(dev_backend) = local::find_at(host, port, DATABASE_DEV_NAME) {
267 debug!(%host, port, "Connected to the database, served by this process");
268 return Ok((dev_backend, Some((host.clone(), port))));
269 }
270
271 match self.inner.backend.connect_at(host, port, DATABASE_DEV_NAME).await {
272 Ok(dev_backend) => {
273 debug!(%host, port, "Connected to the database");
274 Ok((dev_backend, Some((host.clone(), port))))
275 }
276 Err(err) => Err(Error::new(ErrorKind::DatabaseNotConnected)
277 .with_description("Failed to connect to the database")
278 .with_append_description(format_args!("Tried: {host}:{port}"))
279 .with_parent(err)),
280 }
281
282 }
283 ProxyTarget::RegularFromDatabase { ref database, ref name } => {
284
285 debug!(dev = %name, "Connecting to the device, through the database");
286
287 let info = match database.import_dev_async(name).await {
288 Ok(info) if !info.exported =>
289 return Err(Error::new(ErrorKind::DevNotExported)
290 .with_description(format!("Device {name:?} is defined in the database but its server {:?} does not export it",
291 info.server))),
292 Ok(info) => info,
293 Err(err) if let ErrorKind::DevNotConnected = err.kind() =>
294 return Err(Error::new(ErrorKind::DatabaseNotConnected)
295 .with_description(format!("Failed to import device {name:?} from the database"))
296 .with_parent(err)),
297 Err(err) => return Err(err),
298 };
299
300 if let Some(dev_backend) = local::find(info.itr.value()) {
303 debug!(dev = %name, "Connected to the device, served by this process");
304 return Ok((dev_backend, None));
305 }
306
307 match self.inner.backend.connect(info.itr.value()).await {
310 Ok(dev_backend) => {
311 debug!(dev = %name, "Connected to the device");
312 Ok((dev_backend, None))
313 }
314 Err(err) => Err(context_err(err, format!(
315 "Failed to connect to device {name:?} of server {:?}, exported on host {:?}",
316 info.server, info.host))),
317 }
318
319 }
320 ProxyTarget::RegularAt { ref host, port, ref name } => {
321
322 debug!(%host, port, dev = %name, "Connecting to the device");
323
324 if let Some(dev_backend) = local::find_at(host, port, name) {
325 debug!(%host, port, dev = %name, "Connected to the device, served by this process");
326 return Ok((dev_backend, None));
327 }
328
329 match self.inner.backend.connect_at(host, port, name).await {
330 Ok(dev_backend) => {
331 debug!(%host, port, dev = %name, "Connected to the device");
332 Ok((dev_backend, None))
333 }
334 Err(err) => Err(context_err(err, format!(
335 "Failed to connect to device {name:?} at {host}:{port}"))),
336 }
337
338 }
339 }
340
341 })
342 }
343
344 async fn ensure_backend(&self) -> Result<Arc<dyn BackendDev>> {
353
354 if let Some(backend) = self.state.read().unwrap().backend.as_ref() {
355 return Ok(Arc::clone(backend));
356 }
357
358 let (backend, database_host_and_port) = self.connect_backend().await?;
359
360 let mut state = self.state.write().unwrap();
361 match state.backend {
362 Some(ref backend) => Ok(Arc::clone(backend)),
363 None => {
364 state.backend = Some(Arc::clone(&backend));
365 state.database_host_and_port = database_host_and_port;
366 Ok(backend)
367 }
368 }
369
370 }
371
372 fn reset_backend(&self) {
373 let mut state = self.state.write().unwrap();
374 state.backend = None;
375 state.database_host_and_port = None;
376 }
377
378 fn check_backend<T>(&self, res: Result<T>) -> Result<T> {
379 if let Err(err) = &res && err.kind() == ErrorKind::DevNotConnected {
380 self.reset_backend();
381 }
382 res
383 }
384
385 pub(crate) fn block_on<F: Future>(&self, fut: F) -> Result<F::Output> {
386 block_on_timeout(fut, self.inner.timeout)
387 }
388
389 fn database_host_and_port(&self) -> Result<Option<(Arc<str>, u16)>> {
390 self.block_on(self.database_host_and_port_async())?
391 }
392
393 pub(crate) async fn database_host_and_port_async(&self) -> Result<Option<(Arc<str>, u16)>> {
396 self.ensure_backend().await?;
397 Ok(self.state.read().unwrap().database_host_and_port.clone())
398 }
399
400 pub fn database(&self) -> Result<&DatabaseProxy> {
404 match self.inner.target {
405 ProxyTarget::DatabaseFromEnv |
406 ProxyTarget::DatabaseAt { .. } =>
407 Ok(DatabaseProxy::from_dev(self)),
408 ProxyTarget::RegularFromDatabase { ref database, .. } =>
409 Ok(database),
410 ProxyTarget::RegularAt { ref host, port, ref name } =>
411 Err(Error::new(ErrorKind::DatabaseNeeded)
412 .with_description(format!("Device {name:?} is reached at {host}:{port} rather than through a database"))),
413 }
414 }
415
416 #[instrument(skip(self), fields(target = %self.inner.target))]
418 pub fn trl(&self) -> Result<ArcTrl> {
419 match self.inner.target {
420 ProxyTarget::DatabaseFromEnv => {
421 let name = self.name()?;
422 let (host, port) = self.database_host_and_port()?
423 .expect("This should be defined if the target is a database");
424 Ok(format!("tango://{host}:{port}/{name}").parse::<ArcTrl>().unwrap())
425 }
426 ProxyTarget::DatabaseAt { ref host, port } => {
427 let name = self.name()?;
428 Ok(format!("tango://{host}:{port}/{name}").parse::<ArcTrl>().unwrap())
429 }
430 ProxyTarget::RegularFromDatabase { ref database, ref name } => {
431 let (host, port) = database.database_host_and_port()?
432 .expect("This should be defined if the target is a database");
433 Ok(format!("tango://{host}:{port}/{name}").parse::<ArcTrl>().unwrap())
434 }
435 ProxyTarget::RegularAt { ref host, port, ref name } => {
436 Ok(format!("tango://{host}:{port}/{name}#dbase=no").parse::<ArcTrl>().unwrap())
437 }
438 }
439 }
440
441 #[instrument(skip(self), fields(target = %self.inner.target))]
444 pub fn connect(&self) -> Result<()> {
445 self.block_on(self.connect_async())?
446 }
447
448 pub async fn connect_async(&self) -> Result<()> {
450 self.ensure_backend().await.map(drop)
451 }
452
453 #[instrument(skip(self), fields(target = %self.inner.target))]
455 pub fn name(&self) -> Result<Arc<str>> {
456 self.block_on(self.name_async())?
457 }
458
459 pub async fn name_async(&self) -> Result<Arc<str>> {
461 let backend = self.ensure_backend().await?;
462 self.check_backend(backend.name().await)
463 }
464
465 #[instrument(skip(self), fields(target = %self.inner.target))]
467 pub fn admin_dev_name(&self) -> Result<Arc<str>> {
468 self.block_on(self.admin_dev_name_async())?
469 }
470
471 pub async fn admin_dev_name_async(&self) -> Result<Arc<str>> {
473 let backend = self.ensure_backend().await?;
474 self.check_backend(backend.admin_dev_name().await)
475 }
476
477 #[instrument(skip(self), fields(target = %self.inner.target))]
479 pub fn admin_dev(&self) -> Result<DevProxy> {
480 Ok(self.admin_dev_of(self.admin_dev_name()?))
481 }
482
483 pub async fn admin_dev_async(&self) -> Result<DevProxy> {
485 Ok(self.admin_dev_of(self.admin_dev_name_async().await?))
486 }
487
488 fn admin_dev_of(&self, name: Arc<str>) -> DevProxy {
492
493 let backend = Arc::clone(&self.inner.backend);
494
495 DevProxy::new_from_target_with_backend(backend, match &self.inner.target {
496 ProxyTarget::DatabaseFromEnv |
497 ProxyTarget::DatabaseAt { .. } => {
498 ProxyTarget::RegularFromDatabase {
500 database: DatabaseProxy::with_dev(self.clone()),
501 name,
502 }
503 }
504 ProxyTarget::RegularFromDatabase { database, .. } => {
505 ProxyTarget::RegularFromDatabase {
506 database: database.clone(),
507 name,
508 }
509 }
510 ProxyTarget::RegularAt { host, port, .. } => {
511 ProxyTarget::RegularAt {
512 host: Arc::clone(host),
513 port: *port,
514 name,
515 }
516 }
517 })
518
519 }
520
521 #[instrument(skip(self), fields(target = %self.inner.target))]
527 pub fn info(&self) -> Result<DevInfo> {
528 self.block_on(self.info_async())?
529 }
530
531 pub async fn info_async(&self) -> Result<DevInfo> {
533 let backend = self.ensure_backend().await?;
534 self.check_backend(backend.info().await)
535 }
536
537 #[instrument(skip(self), fields(target = %self.inner.target))]
539 pub fn ping(&self) -> Result<()> {
540 self.block_on(self.ping_async())?
541 }
542
543 pub async fn ping_async(&self) -> Result<()> {
545 let backend = self.ensure_backend().await?;
546 self.check_backend(backend.ping().await)
547 }
548
549 #[instrument(skip(self), fields(target = %self.inner.target))]
551 pub fn state(&self) -> Result<DevState> {
552 self.block_on(self.state_async())?
553 }
554
555 pub async fn state_async(&self) -> Result<DevState> {
557 let backend = self.ensure_backend().await?;
558 self.check_backend(backend.state().await)
559 }
560
561 #[instrument(skip(self), fields(target = %self.inner.target))]
563 pub fn status(&self) -> Result<Scalar<str>> {
564 self.block_on(self.status_async())?
565 }
566
567 pub async fn status_async(&self) -> Result<Scalar<str>> {
569 let backend = self.ensure_backend().await?;
570 self.check_backend(backend.status().await)
571 }
572
573 pub fn init(&self) -> Result<()> {
575 self.command("Init", ())
576 }
577
578 pub fn init_async(&self) -> impl Future<Output = Result<()>> + Send {
580 self.command_async("Init", ())
581 }
582
583 #[instrument(skip(self), fields(target = %self.inner.target))]
585 pub fn black_box(&self, max: Option<usize>) -> Result<Spectrum<str>> {
586 self.block_on(self.black_box_async(max))?
587 }
588
589 pub async fn black_box_async(&self, max: Option<usize>) -> Result<Spectrum<str>> {
591 let backend = self.ensure_backend().await?;
592 self.check_backend(backend.black_box(max).await)
593 }
594
595 #[instrument(skip(self, input), fields(target = %self.inner.target))]
602 pub fn command_async<I, O>(&self, name: &str, input: I) -> impl Future<Output = Result<O>> + Send
603 where
604 I: IntoValue,
605 O: FromValue,
606 {
607
608 let command = Scalar::from(name);
615 let input = input.into_value();
616 async move {
617 let backend = self.ensure_backend().await?;
618 let res = backend.command(command, input, self.inner.source).await;
619 O::from_value(self.check_backend(res).map_err(command_err)?)
620 }
621
622 }
623
624 pub fn command<I, O>(&self, name: &str, input: I) -> Result<O>
627 where
628 I: IntoValue,
629 O: FromValue,
630 {
631 self.block_on(self.command_async(name, input))?
632 }
633
634 #[instrument(skip(self, names), fields(target = %self.inner.target))]
661 pub fn read_attr_async<Spec, Names: ToNames>(&self, names: Names) -> impl Future<Output = Result<Spec::Output>> + Send
662 where
663 Spec: ReadAttrSpec<Names>,
664 {
665 let names = names.to_names();
666 async move {
667 let backend = self.ensure_backend().await?;
668 let res = backend.read_attr(names, self.inner.source).await;
669 Spec::collect(self.check_backend(res)?)
670 }
671 }
672
673 pub fn read_attr<Spec, Names: ToNames>(&self, names: Names) -> Result<Spec::Output>
675 where
676 Spec: ReadAttrSpec<Names>,
677 {
678 self.block_on(self.read_attr_async::<Spec, Names>(names))?
679 }
680
681 #[instrument(skip(self, spec), fields(target = %self.inner.target))]
698 pub fn write_attr_async<'a, Spec>(&self, spec: Spec) -> impl Future<Output = Result<Spec::Output>> + Send
699 where
700 Spec: WriteAttrSpec<'a>,
701 {
702
703 let values = spec.into_values();
704
705 async move {
706
707 let backend = self.ensure_backend().await?;
708 let res = backend.write_attr(values).await;
709
710 let outcomes = self.check_backend(res)?.into_iter()
713 .map(|res| res.map_err(write_attr_err))
714 .collect();
715
716 Spec::collect(outcomes)
717
718 }
719
720 }
721
722 pub fn write_attr<'a, Spec>(&self, spec: Spec) -> Result<Spec::Output>
724 where
725 Spec: WriteAttrSpec<'a>,
726 {
727 self.block_on(self.write_attr_async(spec))?
728 }
729
730 #[instrument(skip(self, spec), fields(target = %self.inner.target))]
746 pub fn write_read_attr_async<'a, Spec>(&self, spec: Spec) -> impl Future<Output = Result<<Spec::ReadSpec as ReadAttrSpec<Spec::ReadNames>>::Output>> + Send
747 where
748 Spec: WriteAttrSpec<'a>,
749 Spec::ReadSpec: ReadAttrSpec<Spec::ReadNames>,
750 {
751 let values = spec.into_values();
752 async move {
753 let backend = self.ensure_backend().await?;
754 let res = backend.write_read_attr(values).await;
755 Spec::ReadSpec::collect(self.check_backend(res)?)
756 }
757 }
758
759 pub fn write_read_attr<'a, Spec>(&self, spec: Spec) -> Result<<Spec::ReadSpec as ReadAttrSpec<Spec::ReadNames>>::Output>
761 where
762 Spec: WriteAttrSpec<'a>,
763 Spec::ReadSpec: ReadAttrSpec<Spec::ReadNames>,
764 {
765 self.block_on(self.write_read_attr_async(spec))?
766 }
767
768 #[instrument(skip(self, spec), fields(target = %self.inner.target))]
783 pub fn command_info_async<Spec: ToNamesSpec<CommandInfo>>(&self, spec: Spec) -> impl Future<Output = Result<Spec::Output>> + Send {
784 let names = spec.to_names();
785 async move {
786 let backend = self.ensure_backend().await?;
787 let res = backend.command_info(names).await;
788 Spec::collect(self.check_backend(res)?)
789 }
790 }
791
792 pub fn command_info<Spec: ToNamesSpec<CommandInfo>>(&self, spec: Spec) -> Result<Spec::Output> {
794 self.block_on(self.command_info_async(spec))?
795 }
796
797 #[instrument(skip(self, spec), fields(target = %self.inner.target))]
810 pub fn attr_info_async<Spec: ToNamesSpec<AttrInfo>>(&self, spec: Spec) -> impl Future<Output = Result<Spec::Output>> + Send {
811 let names = spec.to_names();
812 async move {
813 let backend = self.ensure_backend().await?;
814 let res = backend.attr_info(names).await;
815 Spec::collect(self.check_backend(res)?)
816 }
817 }
818
819 pub fn attr_info<Spec: ToNamesSpec<AttrInfo>>(&self, spec: Spec) -> Result<Spec::Output> {
821 self.block_on(self.attr_info_async(spec))?
822 }
823
824 #[instrument(skip(self, values), fields(target = %self.inner.target))]
839 pub async fn set_attr_config_async(&self, values: &[(&str, AttrConfig)]) -> Result<Vec<Result<()>>> {
840 let configs = values.iter()
841 .map(|(name, config)| Named::new(Scalar::from(*name), config.clone()))
842 .collect();
843 let backend = self.ensure_backend().await?;
844 let res = backend.set_attr_config(configs).await;
845 self.check_backend(res).map_err(write_attr_err)
846 }
847
848 pub fn set_attr_config(&self, values: &[(&str, AttrConfig)]) -> Result<Vec<Result<()>>> {
851 self.block_on(self.set_attr_config_async(values))?
852 }
853
854 #[instrument(skip(self), fields(target = %self.inner.target))]
864 pub async fn command_history_async<T: FromValue>(&self, name: &str, max: usize) -> Result<Vec<RecordedResult<T>>> {
865 let backend = self.ensure_backend().await?;
866 let res = backend.command_history(Scalar::from(name), max).await;
867 Ok(self.check_backend(res).map_err(command_err)?.into_iter()
868 .map(|entry| entry.map(|value| value.and_then(T::from_value)))
869 .collect())
870 }
871
872 pub fn command_history<T: FromValue>(&self, name: &str, max: usize) -> Result<Vec<RecordedResult<T>>> {
874 self.block_on(self.command_history_async(name, max))?
875 }
876
877 #[instrument(skip(self), fields(target = %self.inner.target))]
880 pub async fn attr_history_async<T: FromValue>(&self, name: &str, max: usize) -> Result<Vec<RecordedResult<AttrRead<T>>>> {
881 let backend = self.ensure_backend().await?;
882 let res = backend.attr_history(Scalar::from(name), max).await;
883 Ok(self.check_backend(res)?.into_iter()
884 .map(|entry| entry.map(|read| read.and_then(AttrRead::parse)))
885 .collect())
886 }
887
888 pub fn attr_history<T: FromValue>(&self, name: &str, max: usize) -> Result<Vec<RecordedResult<AttrRead<T>>>> {
890 self.block_on(self.attr_history_async(name, max))?
891 }
892
893 pub async fn poll_attr_async(&self, name: &str, period: Duration) -> Result<()> {
901 self.poll_async("attribute", name, period).await
902 }
903
904 pub fn poll_attr(&self, name: &str, period: Duration) -> Result<()> {
906 self.block_on(self.poll_attr_async(name, period))?
907 }
908
909 pub async fn poll_command_async(&self, name: &str, period: Duration) -> Result<()> {
912 self.poll_async("command", name, period).await
913 }
914
915 pub fn poll_command(&self, name: &str, period: Duration) -> Result<()> {
917 self.block_on(self.poll_command_async(name, period))?
918 }
919
920 pub async fn stop_poll_attr_async(&self, name: &str) -> Result<()> {
923 self.stop_poll_async("attribute", name).await
924 }
925
926 pub fn stop_poll_attr(&self, name: &str) -> Result<()> {
928 self.block_on(self.stop_poll_attr_async(name))?
929 }
930
931 pub async fn stop_poll_command_async(&self, name: &str) -> Result<()> {
933 self.stop_poll_async("command", name).await
934 }
935
936 pub fn stop_poll_command(&self, name: &str) -> Result<()> {
938 self.block_on(self.stop_poll_command_async(name))?
939 }
940
941 pub async fn attr_poll_period_async(&self, name: &str) -> Result<Option<Duration>> {
944 Ok(status_period(&self.polling_status_async().await?, "attribute", name))
945 }
946
947 pub fn attr_poll_period(&self, name: &str) -> Result<Option<Duration>> {
950 self.block_on(self.attr_poll_period_async(name))?
951 }
952
953 pub async fn command_poll_period_async(&self, name: &str) -> Result<Option<Duration>> {
956 Ok(status_period(&self.polling_status_async().await?, "command", name))
957 }
958
959 pub fn command_poll_period(&self, name: &str) -> Result<Option<Duration>> {
962 self.block_on(self.command_poll_period_async(name))?
963 }
964
965 pub async fn polling_status_async(&self) -> Result<Vec<String>> {
968 let name = self.name_async().await?;
969 self.admin_dev_async().await?.command_async("DevPollStatus", &*name).await
970 }
971
972 pub fn polling_status(&self) -> Result<Vec<String>> {
974 self.block_on(self.polling_status_async())?
975 }
976
977 async fn poll_async(&self, kind: &str, name: &str, period: Duration) -> Result<()> {
980
981 let dev_name = self.name_async().await?;
982 let admin = self.admin_dev_async().await?;
983 let period = i32::try_from(period.as_millis()).map_err(|_| Error::new(ErrorKind::InvalidData)
984 .with_description(format!("Polling period {period:?} is too long")))?;
985
986 let input = Value::LongString(
987 Spectrum::from_slice(&[period], [1]).expect("one period"),
988 Spectrum::from(&[&*dev_name, kind, name][..]));
989
990 match admin.command_async::<_, ()>("AddObjPolling", input.clone()).await {
993 Err(err) if err.reason() == reason::API_AlreadyPolled => {
994 admin.command_async("UpdObjPollingPeriod", input).await
995 }
996 res => res,
997 }
998
999 }
1000
1001 async fn stop_poll_async(&self, kind: &str, name: &str) -> Result<()> {
1003 let dev_name = self.name_async().await?;
1004 let admin = self.admin_dev_async().await?;
1005 admin.command_async("RemObjPolling", Spectrum::<str>::from(&[&*dev_name, kind, name][..])).await
1006 }
1007
1008 pub async fn list_prop_async(&self, filter: &str) -> Result<Spectrum<str>> {
1014 let dev_name = self.name_async().await?;
1015 self.database()?.list_dev_prop_async(&dev_name, filter).await
1016 }
1017
1018 pub fn list_prop(&self, filter: &str) -> Result<Spectrum<str>> {
1020 self.block_on(self.list_prop_async(filter))?
1021 }
1022
1023 pub fn get_prop_async<Spec, Names: ToNames>(&self, names: Names) -> impl Future<Output = Result<Spec::Output>> + Send
1035 where
1036 Spec: GetPropSpec<Names>,
1037 {
1038 let names = names.to_names();
1041 async move {
1042 let dev_name = self.name_async().await?;
1043 let params = names.map(|names| prop_params(&dev_name, names.iter()));
1044 let mut reply = self.database()?.get_raw_prop_async(PropCommands::DEV, &dev_name, params).await?;
1045 Spec::collect(&mut reply)
1046 }
1047 }
1048
1049 pub fn get_prop<Spec, Names: ToNames>(&self, names: Names) -> Result<Spec::Output>
1051 where
1052 Spec: GetPropSpec<Names>,
1053 {
1054 self.block_on(self.get_prop_async::<Spec, Names>(names))?
1055 }
1056
1057 pub fn set_prop_async<'a, Spec>(&self, spec: Spec) -> impl Future<Output = Result<()>> + Send
1066 where
1067 Spec: SetPropSpec<'a>,
1068 {
1069 let params = set_prop_params(spec);
1070 async move {
1071 let dev_name = self.name_async().await?;
1072 self.database()?.prop_command_async("DbPutDeviceProperty", prop_params(&dev_name, params.iter().map(|param| &**param))).await
1073 }
1074 }
1075
1076 pub fn set_prop<'a, Spec>(&self, spec: Spec) -> Result<()>
1078 where
1079 Spec: SetPropSpec<'a>,
1080 {
1081 self.block_on(self.set_prop_async(spec))?
1082 }
1083
1084 pub fn delete_prop_async<Names: ToNamesNotAll>(&self, names: Names) -> impl Future<Output = Result<()>> + Send {
1093 let names = names.to_names_not_all();
1094 async move {
1095 let dev_name = self.name_async().await?;
1096 self.database()?.prop_command_async("DbDeleteDeviceProperty", prop_params(&dev_name, names.iter())).await
1097 }
1098 }
1099
1100 pub fn delete_prop<Names: ToNamesNotAll>(&self, names: Names) -> Result<()> {
1102 self.block_on(self.delete_prop_async(names))?
1103 }
1104
1105 pub fn subscribe_attr_value<T: FromValue>(&self, name: &str, event: AttrValueEvent) -> DevChannel<AttrRead<T>> {
1113 DevChannel::new(self.clone(), name, ChannelKind::Value(event), ChannelEvents::poll_value)
1114 }
1115
1116 pub fn subscribe_attr_ready(&self, name: &str) -> DevChannel<AttrReady> {
1120 DevChannel::new(self.clone(), name, ChannelKind::Ready, ChannelEvents::poll_ready)
1121 }
1122
1123 pub fn subscribe_attr_config(&self, name: &str) -> DevChannel<AttrInfo> {
1126 DevChannel::new(self.clone(), name, ChannelKind::Config, ChannelEvents::poll_config)
1127 }
1128
1129}
1130
1131pub(crate) fn status_period(status: &[impl AsRef<str>], kind: &str, name: &str) -> Option<Duration> {
1138 let prefix = format!("Polled {kind} name = ");
1139 status.iter().find_map(|entry| {
1140 let mut lines = entry.as_ref().lines();
1141 let polled = lines.next()?.strip_prefix(&prefix)?;
1142 if !polled.trim().eq_ignore_ascii_case(name) {
1143 return None;
1144 }
1145 lines.find_map(|line| line.strip_prefix("Polling period (mS) = "))
1146 .and_then(|ms| ms.trim().parse().ok())
1147 .map(Duration::from_millis)
1148 })
1149}
1150
1151impl Clone for DevProxy {
1152 fn clone(&self) -> Self {
1153 Self {
1154 inner: Box::new(ProxyInner {
1155 backend: self.inner.backend.clone(),
1156 target: self.inner.target.clone(),
1157 source: self.inner.source,
1158 timeout: self.inner.timeout,
1159 }),
1160 state: self.state.try_read().ok()
1161 .map(|guard| RwLock::new((*guard).clone()))
1162 .unwrap_or_else(|| RwLock::new(ProxyState {
1163 backend: None,
1164 database_host_and_port: None,
1165 })),
1166 }
1167 }
1168}
1169
1170impl PartialEq for DevProxy {
1171 #[inline]
1172 fn eq(&self, other: &Self) -> bool {
1173 Arc::ptr_eq(&self.inner.backend, &other.inner.backend) && self.inner.target == other.inner.target
1174 }
1175}
1176
1177impl Eq for DevProxy { }
1178
1179impl Hash for DevProxy {
1180 #[inline]
1181 fn hash<H: Hasher>(&self, state: &mut H) {
1182 Arc::as_ptr(&self.inner.backend).hash(state);
1183 self.inner.target.hash(state);
1184 }
1185}
1186
1187impl fmt::Display for ProxyTarget {
1190 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1191 match self {
1192 ProxyTarget::DatabaseFromEnv =>
1193 write!(f, "db"),
1194 ProxyTarget::DatabaseAt { host, port } =>
1195 write!(f, "db@{host}:{port}"),
1196 ProxyTarget::RegularFromDatabase { name, .. } =>
1197 write!(f, "{name}"),
1198 ProxyTarget::RegularAt { host, port, name } =>
1199 write!(f, "{name}@{host}:{port}"),
1200 }
1201 }
1202}
1203
1204pub trait ToNames: Sized {
1219
1220 fn iter_names(&self) -> Option<impl Iterator<Item = &str>>;
1224
1225 #[inline]
1227 fn to_names(self) -> Option<Spectrum<str>> {
1228 self.iter_names().map(names_buf)
1229 }
1230
1231}
1232
1233fn names_buf<'a>(names: impl Iterator<Item = &'a str>) -> Spectrum<str> {
1235 Spectrum::from_strs(names).expect("names of more than 4 GiB")
1236}
1237
1238pub trait ToNamesNotAll: ToNames {
1243
1244 #[inline]
1246 fn iter_names_not_all(&self) -> impl Iterator<Item = &str> {
1247 self.iter_names().expect("should not return the 'all' marker")
1248 }
1249
1250 #[inline]
1252 fn to_names_not_all(self) -> Spectrum<str> {
1253 self.to_names().expect("should not return the 'all' marker")
1254 }
1255
1256}
1257
1258#[derive(Debug, Clone, PartialEq)]
1266pub struct Named<T> {
1267 pub name: Scalar<str>,
1269 pub value: T,
1271}
1272
1273impl<T> Named<T> {
1274
1275 #[inline]
1276 pub fn new(name: Scalar<str>, value: T) -> Self {
1277 Self { name, value }
1278 }
1279
1280 #[inline]
1282 pub fn map<U>(self, f: impl FnOnce(T) -> U) -> Named<U> {
1283 Named { name: self.name, value: f(self.value) }
1284 }
1285
1286}
1287
1288pub type NamedResult<T> = Named<Result<T>>;
1290
1291#[derive(Debug, Clone, PartialEq)]
1295pub struct Recorded<T> {
1296 pub time: SystemTime,
1298 pub value: T,
1300}
1301
1302impl<T> Recorded<T> {
1303
1304 #[inline]
1305 pub fn new(time: SystemTime, value: T) -> Self {
1306 Self { time, value }
1307 }
1308
1309 #[inline]
1311 pub fn map<U>(self, f: impl FnOnce(T) -> U) -> Recorded<U> {
1312 Recorded { time: self.time, value: f(self.value) }
1313 }
1314
1315}
1316
1317pub type RecordedResult<T> = Recorded<Result<T>>;
1319
1320pub struct All;
1323impl ToNames for All {
1324 #[inline]
1325 fn iter_names(&self) -> Option<impl Iterator<Item = &str>> {
1326 None::<std::iter::Empty<&str>>
1327 }
1328}
1329
1330impl ToNames for &str {
1331 #[inline]
1332 fn iter_names(&self) -> Option<impl Iterator<Item = &str>> {
1333 Some(std::iter::once(*self))
1334 }
1335}
1336impl ToNamesNotAll for &str { }
1337
1338impl ToNames for &[&str] {
1339 #[inline]
1340 fn iter_names(&self) -> Option<impl Iterator<Item = &str>> {
1341 Some(self.iter().copied())
1342 }
1343}
1344impl ToNamesNotAll for &[&str] { }
1345
1346impl<const N: usize> ToNames for [&str; N] {
1347 #[inline]
1348 fn iter_names(&self) -> Option<impl Iterator<Item = &str>> {
1349 Some(self.iter().copied())
1350 }
1351}
1352impl<const N: usize> ToNamesNotAll for [&str; N] { }
1353
1354impl ToNames for Spectrum<str> {
1355 #[inline]
1356 fn iter_names(&self) -> Option<impl Iterator<Item = &str>> {
1357 Some(self.iter())
1358 }
1359 #[inline]
1360 fn to_names(self) -> Option<Spectrum<str>> {
1361 Some(self)
1362 }
1363}
1364impl ToNamesNotAll for Spectrum<str> { }
1365
1366impl ToNames for &Spectrum<str> {
1367 #[inline]
1368 fn iter_names(&self) -> Option<impl Iterator<Item = &str>> {
1369 Some(self.iter())
1370 }
1371 #[inline]
1372 fn to_names(self) -> Option<Spectrum<str>> {
1373 Some(self.clone())
1374 }
1375}
1376impl ToNamesNotAll for &Spectrum<str> { }
1377
1378macro_rules! impl_to_names_tuple {
1379 ( @name: $comp:ident ) => { &str };
1380 ( @one: $comp:ident ) => { 1 };
1381 ( $( $comp:ident )* ) => {
1382 #[allow(unused_variables)]
1383 impl ToNames for ( $(impl_to_names_tuple!(@name: $comp),)* ) {
1384 #[inline]
1385 fn iter_names(&self) -> Option<impl Iterator<Item = &str>> {
1386 let ( $($comp,)* ) = *self;
1387 Some(<[&str; 0 $( + impl_to_names_tuple!(@one: $comp) )*]>::into_iter([ $($comp,)* ]))
1388 }
1389 }
1390 impl ToNamesNotAll for ( $(impl_to_names_tuple!(@name: $comp),)* ) { }
1391 };
1392}
1393
1394impl_to_names_tuple!();
1395impl_to_names_tuple!(a);
1396impl_to_names_tuple!(a b);
1397impl_to_names_tuple!(a b c);
1398impl_to_names_tuple!(a b c d);
1399impl_to_names_tuple!(a b c d e);
1400impl_to_names_tuple!(a b c d e f);
1401
1402pub trait ToNamesSpec<T>: ToNames {
1409
1410 type Output;
1412
1413 fn collect(entries: Vec<NamedResult<T>>) -> Result<Self::Output>;
1418
1419}
1420
1421impl<T> ToNamesSpec<T> for &str {
1424
1425 type Output = T;
1426
1427 fn collect(entries: Vec<NamedResult<T>>) -> Result<Self::Output> {
1428 match entries.into_iter().next() {
1429 Some(entry) => entry.value,
1430 None => Err(unexpected_len_err(1)),
1431 }
1432 }
1433
1434}
1435
1436impl<T> ToNamesSpec<T> for &[&str] {
1437
1438 type Output = Vec<Result<T>>;
1439
1440 fn collect(entries: Vec<NamedResult<T>>) -> Result<Self::Output> {
1441 Ok(entries.into_iter().map(|entry| entry.value).collect())
1442 }
1443
1444}
1445
1446impl<T> ToNamesSpec<T> for Spectrum<str> {
1447
1448 type Output = Vec<Result<T>>;
1449
1450 fn collect(entries: Vec<NamedResult<T>>) -> Result<Self::Output> {
1451 <&[&str] as ToNamesSpec<T>>::collect(entries)
1452 }
1453
1454}
1455
1456impl<T> ToNamesSpec<T> for &Spectrum<str> {
1457
1458 type Output = Vec<Result<T>>;
1459
1460 fn collect(entries: Vec<NamedResult<T>>) -> Result<Self::Output> {
1461 <&[&str] as ToNamesSpec<T>>::collect(entries)
1462 }
1463
1464}
1465
1466impl<T, const N: usize> ToNamesSpec<T> for [&str; N] {
1467
1468 type Output = [Result<T>; N];
1469
1470 fn collect(entries: Vec<NamedResult<T>>) -> Result<Self::Output> {
1471 take_array(entries.into_iter().map(|entry| entry.value))
1472 }
1473
1474}
1475
1476impl<T> ToNamesSpec<T> for All {
1477
1478 type Output = Vec<NamedResult<T>>;
1481
1482 fn collect(entries: Vec<NamedResult<T>>) -> Result<Self::Output> {
1483 Ok(entries)
1484 }
1485
1486}
1487
1488pub(crate) fn take_array<T, const N: usize>(mut entries: impl Iterator<Item = T>) -> Result<[T; N]> {
1492 let mut slots = [const { None }; N];
1493 for slot in &mut slots {
1494 *slot = Some(entries.next().ok_or_else(|| unexpected_len_err(N))?);
1495 }
1496 if entries.next().is_some() {
1497 return Err(unexpected_len_err(N));
1498 }
1499 Ok(slots.map(|slot| slot.expect("every slot is filled")))
1500}
1501
1502macro_rules! impl_to_names_spec_tuple {
1503 ( @one: $index:tt ) => { 1 };
1504 ( @name: $index:tt ) => { &str };
1505 ( @t: $index:tt ) => { T };
1506 ( @next: $index:tt, $entries:ident ) => { $entries.next().ok_or_else(|| unexpected_len_err(LEN))?.value };
1507 ( $( $index:tt )* ) => {
1508
1509 impl<T> ToNamesSpec<T> for ( $( impl_to_names_spec_tuple!(@name: $index), )* ) {
1510
1511 type Output = ( $( Result< impl_to_names_spec_tuple!(@t: $index) >,)* );
1512
1513 fn collect(entries: Vec<NamedResult<T>>) -> Result<Self::Output> {
1514 const LEN: usize = 0 $( + impl_to_names_spec_tuple!(@one: $index) )*;
1515 let mut entries = entries.into_iter();
1518 let output = ( $( impl_to_names_spec_tuple!(@next: $index, entries), )* );
1519 if entries.next().is_some() {
1520 return Err(unexpected_len_err(LEN));
1521 }
1522 Ok(output)
1523 }
1524
1525 }
1526
1527 };
1528}
1529
1530impl_to_names_spec_tuple!(0);
1531impl_to_names_spec_tuple!(0 1);
1532impl_to_names_spec_tuple!(0 1 2);
1533impl_to_names_spec_tuple!(0 1 2 3);
1534impl_to_names_spec_tuple!(0 1 2 3 4);
1535impl_to_names_spec_tuple!(0 1 2 3 4 5);
1536
1537pub trait ReadAttrSpec<Names> {
1547
1548 type Output;
1550
1551 fn collect(reads: Vec<NamedResult<AttrRead<Value>>>) -> Result<Self::Output>;
1556
1557}
1558
1559impl<T: FromValue> ReadAttrSpec<&str> for T {
1560
1561 type Output = AttrRead<T>;
1562
1563 fn collect(reads: Vec<NamedResult<AttrRead<Value>>>) -> Result<Self::Output> {
1564 match reads.into_iter().next() {
1565 Some(entry) => entry.value?.parse(),
1566 None => Err(unexpected_len_err(1)),
1567 }
1568 }
1569
1570}
1571
1572impl<T: FromValue> ReadAttrSpec<&'_ [&str]> for T {
1573
1574 type Output = Vec<Result<AttrRead<T>>>;
1575
1576 fn collect(reads: Vec<NamedResult<AttrRead<Value>>>) -> Result<Self::Output> {
1577 Ok(reads.into_iter().map(|entry| entry.value.and_then(AttrRead::parse)).collect())
1578 }
1579
1580}
1581
1582impl<T: FromValue> ReadAttrSpec<Spectrum<str>> for T {
1583
1584 type Output = Vec<Result<AttrRead<T>>>;
1585
1586 fn collect(reads: Vec<NamedResult<AttrRead<Value>>>) -> Result<Self::Output> {
1587 <T as ReadAttrSpec<&[&str]>>::collect(reads)
1588 }
1589
1590}
1591
1592impl<T: FromValue> ReadAttrSpec<&Spectrum<str>> for T {
1593
1594 type Output = Vec<Result<AttrRead<T>>>;
1595
1596 fn collect(reads: Vec<NamedResult<AttrRead<Value>>>) -> Result<Self::Output> {
1597 <T as ReadAttrSpec<&[&str]>>::collect(reads)
1598 }
1599
1600}
1601
1602impl<T: FromValue, const N: usize> ReadAttrSpec<[&str; N]> for T {
1603
1604 type Output = [Result<AttrRead<T>>; N];
1605
1606 fn collect(reads: Vec<NamedResult<AttrRead<Value>>>) -> Result<Self::Output> {
1607 take_array(reads.into_iter().map(|entry| entry.value.and_then(AttrRead::parse)))
1608 }
1609
1610}
1611
1612impl<T: FromValue> ReadAttrSpec<All> for T {
1613
1614 type Output = Vec<NamedResult<AttrRead<T>>>;
1616
1617 fn collect(reads: Vec<NamedResult<AttrRead<Value>>>) -> Result<Self::Output> {
1618 Ok(reads.into_iter().map(|entry| entry.map(|read| read.and_then(AttrRead::parse))).collect())
1619 }
1620
1621}
1622
1623macro_rules! impl_read_attr_spec_tuple {
1624 ( @one: $comp:ident ) => { 1 };
1625 ( @name: $comp:ident ) => { &'name str };
1626 ( $( $comp:ident )* ) => {
1627
1628 impl<'name, $($comp,)* > ReadAttrSpec<( $( impl_read_attr_spec_tuple!(@name: $comp), )* )> for ( $($comp,)* )
1629 where
1630 $( $comp: FromValue, )*
1631 {
1632
1633 type Output = ( $( Result<AttrRead<$comp>>,)* );
1634
1635 fn collect(reads: Vec<NamedResult<AttrRead<Value>>>) -> Result<Self::Output> {
1636 const LEN: usize = 0 $( + impl_read_attr_spec_tuple!(@one: $comp) )*;
1637 let mut reads = reads.into_iter();
1641 let output = ( $( reads.next().ok_or_else(|| unexpected_len_err(LEN))?.value.and_then(AttrRead::parse::<$comp>), )* );
1642 if reads.next().is_some() {
1643 return Err(unexpected_len_err(LEN));
1644 }
1645 Ok(output)
1646 }
1647
1648 }
1649
1650 };
1651}
1652
1653impl_read_attr_spec_tuple!(A);
1654impl_read_attr_spec_tuple!(A B);
1655impl_read_attr_spec_tuple!(A B C);
1656impl_read_attr_spec_tuple!(A B C D);
1657impl_read_attr_spec_tuple!(A B C D E);
1658impl_read_attr_spec_tuple!(A B C D E F);
1659
1660pub trait WriteAttrSpec<'a>: Sized {
1669
1670 fn into_values(self) -> Vec<Named<Value>>;
1673
1674 type Output;
1676
1677 fn collect(writes: Vec<Result<()>>) -> Result<Self::Output>;
1681
1682 type ReadNames;
1684 type ReadSpec;
1686
1687}
1688
1689impl<'a, T: IntoValue> WriteAttrSpec<'a> for (&'a str, T) {
1691
1692 #[inline]
1693 fn into_values(self) -> Vec<Named<Value>> {
1694 vec![Named::new(Scalar::from(self.0), self.1.into_value())]
1695 }
1696
1697 type Output = ();
1698
1699 fn collect(writes: Vec<Result<()>>) -> Result<Self::Output> {
1700 match writes.into_iter().next() {
1701 Some(res) => res,
1702 None => Err(unexpected_len_err(1)),
1703 }
1704 }
1705
1706 type ReadNames = &'a str;
1707 type ReadSpec = T;
1708
1709}
1710
1711impl<'a, T: IntoValue, const N: usize> WriteAttrSpec<'a> for [(&'a str, T); N] {
1713
1714 #[inline]
1715 fn into_values(self) -> Vec<Named<Value>> {
1716 named_values(self.into_iter().map(|(name, value)| (name, value.into_value())).collect())
1717 }
1718
1719 type Output = [Result<()>; N];
1720
1721 fn collect(writes: Vec<Result<()>>) -> Result<Self::Output> {
1722 take_array(writes.into_iter())
1723 }
1724
1725 type ReadNames = [&'a str; N];
1726 type ReadSpec = T;
1727
1728}
1729
1730impl<'a, T: IntoValue> WriteAttrSpec<'a> for Vec<(&'a str, T)> {
1732
1733 #[inline]
1734 fn into_values(self) -> Vec<Named<Value>> {
1735 named_values(self.into_iter().map(|(name, value)| (name, value.into_value())).collect())
1736 }
1737
1738 type Output = Vec<Result<()>>;
1739
1740 fn collect(writes: Vec<Result<()>>) -> Result<Self::Output> {
1741 Ok(writes)
1742 }
1743
1744 type ReadNames = &'a [&'a str];
1745 type ReadSpec = T;
1746
1747}
1748
1749fn named_values(values: Vec<(&str, Value)>) -> Vec<Named<Value>> {
1752 let names = names_buf(values.iter().map(|&(name, _)| name));
1753 values.into_iter()
1754 .enumerate()
1755 .map(|(index, (_, value))| Named::new(names.get_scalar([index]).expect("as many names as values"), value))
1756 .collect()
1757}
1758
1759macro_rules! impl_write_attr_spec_tuple {
1760 ( @name: $comp:ident ) => { &'a str };
1761 ( @one: $comp:ident ) => { 1 };
1762 ( @unit: $comp:ident ) => { Result<()> };
1763 ( @next: $comp:ident, $writes:ident ) => { $writes.next().ok_or_else(|| unexpected_len_err(LEN))? };
1764 ( $( $comp:ident )* ) => {
1765
1766 #[allow(non_snake_case)]
1767 impl<'a, $($comp: IntoValue, )*> WriteAttrSpec<'a> for ( $( (&'a str, $comp), )* ) {
1768
1769 #[inline]
1770 fn into_values(self) -> Vec<Named<Value>> {
1771 let ( $($comp,)* ) = self;
1772 named_values(vec![$(($comp.0, $comp.1.into_value()), )* ])
1773 }
1774
1775 type Output = ( $( impl_write_attr_spec_tuple!(@unit: $comp), )* );
1776
1777 fn collect(writes: Vec<Result<()>>) -> Result<Self::Output> {
1778 const LEN: usize = 0 $( + impl_write_attr_spec_tuple!(@one: $comp) )*;
1779 let mut writes = writes.into_iter();
1782 let output = ( $( impl_write_attr_spec_tuple!(@next: $comp, writes), )* );
1783 if writes.next().is_some() {
1784 return Err(unexpected_len_err(LEN));
1785 }
1786 Ok(output)
1787 }
1788
1789 type ReadNames = ( $(impl_write_attr_spec_tuple!(@name: $comp), )* );
1790 type ReadSpec = ( $($comp, )* );
1791
1792 }
1793
1794 };
1795}
1796
1797impl_write_attr_spec_tuple!(A);
1798impl_write_attr_spec_tuple!(A B);
1799impl_write_attr_spec_tuple!(A B C);
1800impl_write_attr_spec_tuple!(A B C D);
1801impl_write_attr_spec_tuple!(A B C D E);
1802impl_write_attr_spec_tuple!(A B C D E F);
1803
1804#[derive(Debug, Clone, PartialEq, Eq)]
1810pub struct CommandInfo {
1811 pub display_level: DisplayLevel,
1813 pub in_kind: CommandDataKind,
1815 pub out_kind: CommandDataKind,
1817 pub in_doc: Option<Cow<'static, str>>,
1819 pub out_doc: Option<Cow<'static, str>>,
1821}
1822
1823#[derive(Debug, Clone, PartialEq, Eq)]
1829pub struct AttrRead<T> {
1830 pub value: Option<T>,
1833 pub w_value: Option<T>,
1836 pub quality: AttrQuality,
1838 pub time: SystemTime,
1840}
1841
1842impl<T> AttrRead<T> {
1843
1844 pub fn new(value: T) -> Self {
1845 Self {
1846 value: Some(value),
1847 w_value: None,
1848 quality: AttrQuality::Valid,
1849 time: SystemTime::now(),
1850 }
1851 }
1852
1853 pub fn with_quality(mut self, quality: AttrQuality) -> Self {
1854 self.quality = quality;
1855 self
1856 }
1857
1858 pub fn with_time(mut self, time: SystemTime) -> Self {
1859 self.time = time;
1860 self
1861 }
1862
1863}
1864
1865impl AttrRead<Value> {
1866
1867 fn parse<T: FromValue>(self) -> Result<AttrRead<T>> {
1869 Ok(AttrRead {
1870 value: match self.value {
1871 Some(value) => Some(T::from_value(value)
1872 .map_err(|err| context_err(err, "Failed to parse read value"))?),
1873 None => None,
1874 },
1875 w_value: match self.w_value {
1876 Some(value) => Some(T::from_value(value)
1877 .map_err(|err| context_err(err, "Failed to parse write value"))?),
1878 None => None,
1879 },
1880 quality: self.quality,
1881 time: self.time,
1882 })
1883 }
1884
1885}
1886
1887#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1892pub struct AttrReady {
1893 pub data_kind: AttrDataKind,
1895 pub counter: i32,
1898}
1899
1900#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
1902pub enum AttrQuality {
1903 Valid,
1904 Invalid,
1905 Alarm,
1906 Changing,
1907 Warning,
1908}
1909
1910#[derive(Debug, Clone, PartialEq)]
1912pub struct AttrInfo {
1913 pub config: AttrConfig,
1917 pub mode: AttrMode,
1919 pub data_kind: AttrDataKind,
1921 pub data_format: AttrDataFormat,
1924 pub display_level: DisplayLevel,
1926 pub memorized: Option<AttrMemorized>,
1928}
1929
1930#[derive(Debug, Clone, PartialEq, Eq)]
1932pub enum AttrMode {
1933 Read,
1935 ReadWithWrite {
1938 write_attr_name: Cow<'static, str>,
1940 },
1941 Write,
1943 ReadWrite,
1945}
1946
1947#[derive(Debug, Clone, PartialEq, Default)]
1959pub struct AttrConfig {
1960 pub doc: Option<Cow<'static, str>>,
1963 pub label: Option<Cow<'static, str>>,
1965 pub unit: Option<Cow<'static, str>>,
1968 pub standard_unit: Option<f64>,
1971 pub display_unit: Option<Cow<'static, str>>,
1973 pub display_format: Option<Cow<'static, str>>,
1976 pub range: AttrRangeConfig,
1978 pub alarm: AttrRangeConfig,
1980 pub warning: AttrRangeConfig,
1982 pub delta: Option<AttrDeltaConfig>,
1984 pub change_event: AttrChangeEventConfig,
1986 pub periodic_event: AttrPeriodicEventConfig,
1988 pub archive_event: AttrArchiveEventConfig,
1990 pub enum_labels: Vec<Cow<'static, str>>,
1997 pub extensions: Vec<Cow<'static, str>>,
1999}
2000
2001#[derive(Debug, Clone, Copy)]
2025pub enum AttrNumber {
2026 Double(f64),
2027 Long(i64),
2028 ULong(u64),
2029}
2030
2031impl AttrNumber {
2032
2033 pub fn as_f64(self) -> f64 {
2037 match self {
2038 Self::Double(value) => value,
2039 Self::Long(value) => value as f64,
2040 Self::ULong(value) => value as f64,
2041 }
2042 }
2043
2044}
2045
2046impl PartialEq for AttrNumber {
2047 #[inline]
2048 fn eq(&self, other: &Self) -> bool {
2049 self.partial_cmp(other) == Some(Ordering::Equal)
2050 }
2051}
2052
2053impl PartialOrd for AttrNumber {
2054 fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
2055 match (*self, *other) {
2056 (Self::Long(a), Self::Long(b)) => Some(a.cmp(&b)),
2057 (Self::ULong(a), Self::ULong(b)) => Some(a.cmp(&b)),
2058 (Self::Long(a), Self::ULong(b)) => Some(i128::from(a).cmp(&i128::from(b))),
2059 (Self::ULong(a), Self::Long(b)) => Some(i128::from(a).cmp(&i128::from(b))),
2060 (a, b) => a.as_f64().partial_cmp(&b.as_f64()),
2061 }
2062 }
2063}
2064
2065impl fmt::Display for AttrNumber {
2066 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
2068 match self {
2069 Self::Double(value) => Display::fmt(value, f),
2070 Self::Long(value) => Display::fmt(value, f),
2071 Self::ULong(value) => Display::fmt(value, f),
2072 }
2073 }
2074}
2075
2076macro_rules! impl_attr_number_from {
2079 ($variant:ident: $($ty:ty),* $(,)?) => { $(
2080 impl From<$ty> for AttrNumber {
2081 #[inline]
2082 fn from(value: $ty) -> Self {
2083 Self::$variant(value.into())
2084 }
2085 }
2086 )* };
2087}
2088
2089impl_attr_number_from!(Double: f32, f64);
2090impl_attr_number_from!(Long: i8, i16, i32, i64);
2091impl_attr_number_from!(ULong: u8, u16, u32, u64);
2092
2093#[derive(Debug, Clone, Copy, PartialEq, Default)]
2097pub struct AttrRangeConfig {
2098 pub min: Option<AttrNumber>,
2099 pub max: Option<AttrNumber>,
2100}
2101
2102impl AttrRangeConfig {
2103 #[inline]
2105 pub fn is_set(&self) -> bool {
2106 self.min.is_some() || self.max.is_some()
2107 }
2108}
2109
2110#[derive(Debug, Clone, Copy, PartialEq)]
2112pub struct AttrDeltaConfig {
2113 pub value: AttrNumber,
2116 pub time: Option<Duration>,
2118}
2119
2120#[derive(Debug, Clone, Copy, PartialEq, Default)]
2123pub struct AttrChangeEventConfig {
2124 pub rel_change: Option<f64>,
2126 pub abs_change: Option<f64>,
2128}
2129
2130impl AttrChangeEventConfig {
2131 #[inline]
2133 pub fn is_set(&self) -> bool {
2134 self.rel_change.is_some() || self.abs_change.is_some()
2135 }
2136}
2137
2138#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
2139pub struct AttrPeriodicEventConfig {
2140 pub period: Option<Duration>,
2141}
2142
2143impl AttrPeriodicEventConfig {
2144 #[inline]
2146 pub fn is_set(&self) -> bool {
2147 self.period.is_some()
2148 }
2149}
2150
2151#[derive(Debug, Clone, Copy, PartialEq, Default)]
2154pub struct AttrArchiveEventConfig {
2155 pub rel_change: Option<f64>,
2156 pub abs_change: Option<f64>,
2157 pub period: Option<Duration>,
2158}
2159
2160impl AttrArchiveEventConfig {
2161 #[inline]
2163 pub fn is_set(&self) -> bool {
2164 self.rel_change.is_some() || self.abs_change.is_some() || self.period.is_some()
2165 }
2166}
2167
2168#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
2174pub enum AttrMemorized {
2175 RestoreSetPoint,
2180 RestoreAndWrite,
2183}
2184
2185#[derive(Debug, Clone, PartialEq, Eq, Default)]
2194pub struct DevInfo {
2195 pub class: String,
2197 pub server_id: String,
2200 pub server_host: String,
2202 pub server_version: i32,
2205 pub doc_url: Option<Cow<'static, str>>,
2207 pub dev_type: Option<Cow<'static, str>>,
2210 pub version_info: Vec<(Cow<'static, str>, Cow<'static, str>)>,
2214}
2215
2216#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
2222#[repr(u8)]
2223pub enum DevState {
2224 On,
2225 Off,
2226 Close,
2227 Open,
2228 Insert,
2229 Extract,
2230 Moving,
2231 Standby,
2232 Fault,
2233 Init,
2234 Running,
2235 Alarm,
2236 Disable,
2237 Unknown,
2238}
2239
2240#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
2242pub enum DevSource {
2243 Dev,
2245 Cache,
2247 CacheDev,
2249}
2250
2251#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Default)]
2252pub enum DisplayLevel {
2253 #[default]
2255 Operator,
2256 Expert,
2257}
2258
2259enum ChannelEvents {
2266 Value(BoxStream<'static, Result<AttrRead<Value>>>),
2267 Ready(BoxStream<'static, Result<AttrReady>>),
2268 Config(BoxStream<'static, Result<AttrInfo>>),
2269}
2270
2271impl ChannelEvents {
2272
2273 fn poll_value<T: FromValue>(&mut self, cx: &mut Context<'_>) -> Poll<Option<Result<AttrRead<T>>>> {
2275 let Self::Value(stream) = self else { unreachable!("a value channel subscribes to values") };
2276 stream.poll_next_unpin(cx).map(|item| item.map(|read| read.and_then(AttrRead::parse)))
2277 }
2278
2279 fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Option<Result<AttrReady>>> {
2281 let Self::Ready(stream) = self else { unreachable!("a ready channel subscribes to readiness") };
2282 stream.poll_next_unpin(cx)
2283 }
2284
2285 fn poll_config(&mut self, cx: &mut Context<'_>) -> Poll<Option<Result<AttrInfo>>> {
2287 let Self::Config(stream) = self else { unreachable!("a configuration channel subscribes to configurations") };
2288 stream.poll_next_unpin(cx)
2289 }
2290
2291}
2292
2293#[derive(Debug, Clone, Copy)]
2295enum ChannelKind {
2296 Value(AttrValueEvent),
2297 Ready,
2298 Config,
2299}
2300
2301pub struct DevChannel<T> {
2329 dev: Option<DevProxy>,
2332 name: Arc<str>,
2334 kind: ChannelKind,
2335 poll_events: fn(&mut ChannelEvents, &mut Context<'_>) -> Poll<Option<Result<T>>>,
2337 timeout: Duration,
2339 state: ChannelState,
2340}
2341
2342enum ChannelState {
2343 Subscribed(ChannelEvents),
2345 Subscribing(BoxFuture<'static, (DevProxy, Result<ChannelEvents>)>),
2348}
2349
2350impl ChannelState {
2351
2352 fn subscribing(dev: DevProxy, name: Arc<str>, kind: ChannelKind) -> Self {
2355 Self::Subscribing(Box::pin(async move {
2356 let res = async {
2357 let backend = dev.ensure_backend().await?;
2358 let admin = dev.admin_dev_async().await?;
2359 let admin = admin.ensure_backend().await?;
2360 let res = match kind {
2361 ChannelKind::Value(event) => backend.subscribe_attr_value(admin, Scalar::from(&*name), event).await.map(ChannelEvents::Value),
2362 ChannelKind::Ready => backend.subscribe_attr_ready(admin, Scalar::from(&*name)).await.map(ChannelEvents::Ready),
2363 ChannelKind::Config => backend.subscribe_attr_config(admin, Scalar::from(&*name)).await.map(ChannelEvents::Config),
2364 };
2365 dev.check_backend(res)
2366 }.await;
2367 (dev, res)
2368 }))
2369 }
2370
2371}
2372
2373impl<T> DevChannel<T> {
2374
2375 fn new(
2378 dev: DevProxy,
2379 name: &str,
2380 kind: ChannelKind,
2381 poll_events: fn(&mut ChannelEvents, &mut Context<'_>) -> Poll<Option<Result<T>>>,
2382 ) -> Self {
2383 let timeout = dev.timeout();
2384 let name = Arc::<str>::from(name);
2385 let state = ChannelState::subscribing(dev, Arc::clone(&name), kind);
2386 Self { dev: None, name, kind, poll_events, timeout, state }
2387 }
2388
2389 fn resubscribe(&mut self) {
2391 let dev = self.dev.take().expect("the device is moved into one attempt at a time");
2392 self.state = ChannelState::subscribing(dev, Arc::clone(&self.name), self.kind);
2393 }
2394
2395 fn poll_subscribed(&mut self, cx: &mut Context<'_>) -> Poll<Result<()>> {
2398
2399 let ChannelState::Subscribing(attempt) = &mut self.state else {
2400 return Poll::Ready(Ok(()));
2401 };
2402
2403 let (dev, res) = ready!(attempt.as_mut().poll(cx));
2404 let dev = &*self.dev.insert(dev);
2405
2406 Poll::Ready(match res {
2407 Ok(events) => {
2408 debug!(dev = %dev.inner.target, attr = %self.name, event = ?self.kind, "Subscribed to the events");
2409 self.state = ChannelState::Subscribed(events);
2410 Ok(())
2411 }
2412 Err(err) => {
2413 debug!(dev = %dev.inner.target, attr = %self.name, event = ?self.kind, %err, "Failed to subscribe to the events, retrying on the next poll");
2414 self.resubscribe();
2415 Err(err)
2416 }
2417 })
2418
2419 }
2420
2421 pub async fn ensure_subscribed_async(&mut self) -> Result<()> {
2425 poll_fn(|cx| self.poll_subscribed(cx)).await
2426 }
2427
2428 pub fn ensure_subscribed(&mut self) -> Result<()> {
2431 let timeout = self.timeout;
2432 block_on_timeout(self.ensure_subscribed_async(), timeout)?
2433 }
2434
2435 pub fn wait(&mut self, timeout: Duration) -> Result<T> {
2438 match block_on_timeout(self.next(), timeout)? {
2439 Some(item) => item,
2440 None => unreachable!("a device channel never ends"),
2441 }
2442 }
2443
2444}
2445
2446impl<T> Stream for DevChannel<T> {
2447
2448 type Item = Result<T>;
2449
2450 fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
2451
2452 let this = self.get_mut();
2453
2454 loop {
2455
2456 if let Err(err) = ready!(this.poll_subscribed(cx)) {
2457 return Poll::Ready(Some(Err(err)));
2458 }
2459
2460 let ChannelState::Subscribed(events) = &mut this.state else {
2461 unreachable!("subscribed once the attempt succeeded");
2462 };
2463
2464 let dev = this.dev.as_ref().expect("the device is back once subscribed");
2465
2466 match ready!((this.poll_events)(events, cx)) {
2467 Some(Err(err)) if err.kind() == ErrorKind::DevNotConnected => {
2468 debug!(dev = %dev.inner.target, attr = %this.name, event = ?this.kind, %err, "Lost the subscription to the events, subscribing again on the next poll");
2469 dev.reset_backend();
2470 this.resubscribe();
2471 return Poll::Ready(Some(Err(err)));
2472 }
2473 Some(item) => return Poll::Ready(Some(item)),
2474 None => {
2476 debug!(dev = %dev.inner.target, attr = %this.name, event = ?this.kind, "The subscription to the events ended, subscribing again");
2477 this.resubscribe();
2478 }
2479 }
2480
2481 }
2482
2483 }
2484
2485}
2486
2487impl<T> Debug for DevChannel<T> {
2488 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
2489 f.debug_struct("DevChannel")
2490 .field("dev", &self.dev)
2491 .field("name", &self.name)
2492 .finish_non_exhaustive()
2493 }
2494}
2495
2496#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
2498pub enum AttrValueEvent {
2499 Change,
2500 Periodic,
2501 Archive,
2502 User,
2503 Alarm,
2504}
2505
2506#[cold]
2510pub(crate) fn unexpected_len_err(expected: usize) -> Error {
2511 Error::new(ErrorKind::Internal)
2512 .with_description(format!("Device answered a different number of entries where {expected} were requested"))
2513}
2514
2515#[cold]
2519#[track_caller]
2520fn context_err(err: Error, description: impl Into<Cow<'static, str>>) -> Error {
2521 Error::new(err.kind())
2522 .with_description(description)
2523 .with_severity(err.severity())
2524 .with_parent(err)
2525}
2526
2527fn command_err(err: Error) -> Error {
2532 match err.kind() {
2533 ErrorKind::Reason(reason::API_CommandNotFound) => {
2534 err.with_kind(ErrorKind::CommandNotFound)
2535 }
2536 ErrorKind::Reason(reason::API_IncompatibleCmdArgumentType) => {
2537
2538 let expected_kind = expected_cpp_tango_type(err.description())
2539 .and_then(command_data_kind_from_cpp_tango);
2540
2541 let err = err.with_kind(ErrorKind::CommandArgumentKindInvalid);
2542 match expected_kind {
2543 Some(kind) => err.with_append_description(format_args!("Expected kind: {kind:?}")),
2544 None => err,
2545 }
2546
2547 }
2548 _ => err,
2549 }
2550}
2551
2552fn write_attr_err(err: Error) -> Error {
2554 match err.kind() {
2555 ErrorKind::Reason(reason::API_AttrNotFound) => {
2556 err.with_kind(ErrorKind::AttrNotFound)
2557 }
2558 ErrorKind::Reason(reason::API_AttrNotWritable) => {
2559 err.with_kind(ErrorKind::AttrNotWritable)
2560 }
2561 ErrorKind::Reason(reason::API_IncompatibleAttrDataType) => {
2562
2563 let expected_kind = expected_cpp_tango_type(err.description())
2564 .and_then(attr_data_kind_from_cpp_tango);
2565
2566 let err = err.with_kind(ErrorKind::AttrKindInvalid);
2567 match expected_kind {
2568 Some(kind) => err.with_append_description(format_args!("Expected kind: {kind:?}")),
2569 None => err,
2570 }
2571
2572 }
2573 _ => err,
2574 }
2575}
2576
2577fn expected_cpp_tango_type(description: &str) -> Option<&str> {
2590 description.split_ascii_whitespace()
2591 .find_map(|word| word.strip_prefix("Tango::"))
2592 .map(|ty| ty.trim_matches(|c: char| !c.is_ascii_alphanumeric()))
2593}
2594
2595fn command_data_kind_from_cpp_tango(cpp_type: &str) -> Option<CommandDataKind> {
2600 Some(match cpp_type {
2601 "DevVoid" => CommandDataKind::Void,
2602 "DevBoolean" => CommandDataKind::Bool,
2603 "DevUChar" => CommandDataKind::Octet,
2604 "DevShort" => CommandDataKind::Short,
2605 "DevUShort" => CommandDataKind::UShort,
2606 "DevLong" => CommandDataKind::Long,
2607 "DevULong" => CommandDataKind::ULong,
2608 "DevLong64" => CommandDataKind::Long64,
2609 "DevULong64" => CommandDataKind::ULong64,
2610 "DevFloat" => CommandDataKind::Float,
2611 "DevDouble" => CommandDataKind::Double,
2612 "DevState" => CommandDataKind::State,
2613 "DevString" |
2614 "ConstDevString" => CommandDataKind::String,
2615 "DevVarBooleanArray" => CommandDataKind::BoolArray,
2616 "DevVarCharArray" => CommandDataKind::OctetArray,
2617 "DevVarShortArray" => CommandDataKind::ShortArray,
2618 "DevVarUShortArray" => CommandDataKind::UShortArray,
2619 "DevVarLongArray" => CommandDataKind::LongArray,
2620 "DevVarULongArray" => CommandDataKind::ULongArray,
2621 "DevVarLong64Array" => CommandDataKind::Long64Array,
2622 "DevVarULong64Array" => CommandDataKind::ULong64Array,
2623 "DevVarFloatArray" => CommandDataKind::FloatArray,
2624 "DevVarDoubleArray" => CommandDataKind::DoubleArray,
2625 "DevVarStateArray" => CommandDataKind::StateArray,
2626 "DevVarStringArray" => CommandDataKind::StringArray,
2627 "DevVarLongStringArray" => CommandDataKind::LongStringArray,
2628 "DevVarDoubleStringArray" => CommandDataKind::DoubleStringArray,
2629 "DevEncoded" => CommandDataKind::Encoded,
2630 _ => return None,
2631 })
2632}
2633
2634fn attr_data_kind_from_cpp_tango(cpp_type: &str) -> Option<AttrDataKind> {
2645 Some(match cpp_type {
2646 "DevBoolean" | "DevVarBooleanArray" => AttrDataKind::Bool,
2647 "DevUChar" | "DevVarCharArray" => AttrDataKind::Octet,
2648 "DevShort" | "DevVarShortArray" => AttrDataKind::Short,
2649 "DevUShort" | "DevVarUShortArray" => AttrDataKind::UShort,
2650 "DevLong" | "DevVarLongArray" => AttrDataKind::Long,
2651 "DevULong" | "DevVarULongArray" => AttrDataKind::ULong,
2652 "DevLong64" | "DevVarLong64Array" => AttrDataKind::Long64,
2653 "DevULong64" | "DevVarULong64Array" => AttrDataKind::ULong64,
2654 "DevFloat" | "DevVarFloatArray" => AttrDataKind::Float,
2655 "DevDouble" | "DevVarDoubleArray" => AttrDataKind::Double,
2656 "DevState" | "DevVarStateArray" => AttrDataKind::State,
2657 "DevString" | "ConstDevString" | "DevVarStringArray" => AttrDataKind::String,
2658 "DevEncoded" | "DevVarEncodedArray" => AttrDataKind::Encoded,
2659 _ => return None,
2660 })
2661}
2662
2663struct DisplayHostAndPorts<'a>(&'a [(Arc<str>, u16)]);
2666
2667impl Display for DisplayHostAndPorts<'_> {
2668 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
2669 for (i, (host, port)) in self.0.iter().enumerate() {
2670 if i != 0 {
2671 f.write_str(", ")?;
2672 }
2673 write!(f, "{host}:{port}")?;
2674 }
2675 Ok(())
2676 }
2677}
2678
2679fn env_databases() -> &'static [(Arc<str>, u16)] {
2681 static CACHE: OnceLock<Box<[(Arc<str>, u16)]>> = OnceLock::new();
2682 CACHE.get_or_init(|| {
2683
2684 let mut ret = Vec::new();
2685
2686 if let Ok(var) = env::var("TANGO_HOST") {
2687 for part in var.split(',') {
2688 if let Some((host, port_raw)) = part.rsplit_once(':')
2689 && let Ok(port) = u16::from_str(port_raw) {
2690 ret.push((host.into(), port));
2691 }
2692 }
2693 }
2694
2695 ret.into_boxed_slice()
2696
2697 })
2698}
2699
2700
2701#[cfg(test)]
2702mod tests {
2703
2704 use crate::value::Scalar;
2705
2706 use super::*;
2707
2708 #[allow(dead_code)]
2711 fn props_in_send_future(proxy: &DevProxy) -> BoxFuture<'_, Result<()>> {
2712 Box::pin(async move {
2713 let _speed = proxy.get_prop_async::<f64, _>("speed").await?;
2714 let _all = proxy.get_prop_async::<Spectrum<str>, _>(All).await?;
2715 proxy.set_prop_async(("speed", 2.5)).await?;
2716 proxy.delete_prop_async(["speed", "accel"]).await
2717 })
2718 }
2719
2720 #[allow(dead_code)]
2723 fn requests_in_send_future(proxy: &DevProxy) -> BoxFuture<'_, Result<()>> {
2724 Box::pin(async move {
2725 let _: f64 = proxy.command_async("Add", 1.0).await?;
2726 let _: String = proxy.command_async("Echo", "text").await?;
2727 let _ = proxy.read_attr_async::<f64, _>("speed").await?;
2728 let _ = proxy.read_attr_async::<f64, _>(["speed", "accel"]).await?;
2729 let _ = proxy.read_attr_async::<Value, _>(All).await?;
2730 proxy.write_attr_async(("speed", 2.5)).await?;
2731 let _ = proxy.write_attr_async([("speed", 2.5), ("accel", 1.0)]).await?;
2732 let _ = proxy.write_read_attr_async(("speed", 2.5)).await?;
2733 let _ = proxy.command_info_async("Add").await?;
2734 let _ = proxy.attr_info_async(["speed"]).await?;
2735 let names = proxy.list_prop_async("*").await?;
2736 let _ = proxy.read_attr_async::<f64, _>(&names).await?;
2737 let _ = proxy.attr_info_async(names).await?;
2738 Ok(())
2739 })
2740 }
2741
2742 #[allow(dead_code)]
2745 fn props_of_any_names<'a, Names>(proxy: &'a DevProxy, names: Names) -> BoxFuture<'a, Result<<f64 as GetPropSpec<Names>>::Output>>
2746 where
2747 Names: ToNames + 'a,
2748 f64: GetPropSpec<Names>,
2749 {
2750 Box::pin(proxy.get_prop_async::<f64, Names>(names))
2751 }
2752
2753 #[test]
2754 fn expected_kind() {
2755
2756 let cmd = "Incompatible command argument type, expected type is : Tango::DevVarDoubleStringArray\0";
2759 assert_eq!(expected_cpp_tango_type(cmd), Some("DevVarDoubleStringArray"));
2760 assert_eq!(
2761 expected_cpp_tango_type(cmd).and_then(command_data_kind_from_cpp_tango),
2762 Some(CommandDataKind::DoubleStringArray),
2763 );
2764
2765 let attr = "Incompatible attribute type: expected Tango::DevVarDoubleArray (even for single value), found Tango::DevVarStringArray\0";
2768 assert_eq!(expected_cpp_tango_type(attr), Some("DevVarDoubleArray"));
2769 assert_eq!(
2770 expected_cpp_tango_type(attr).and_then(attr_data_kind_from_cpp_tango),
2771 Some(AttrDataKind::Double),
2772 );
2773
2774 assert_eq!(attr_data_kind_from_cpp_tango("DevVarEncodedArray"), Some(AttrDataKind::Encoded));
2775 assert_eq!(command_data_kind_from_cpp_tango("DevEncoded"), Some(CommandDataKind::Encoded));
2776
2777 assert_eq!(expected_cpp_tango_type("Attribute my_attr is not writable"), None);
2779 assert_eq!(command_data_kind_from_cpp_tango("DevVarEncodedArray"), None);
2780
2781 }
2782
2783 fn reads<const N: usize>(entries: [(&str, Value); N]) -> Vec<NamedResult<AttrRead<Value>>> {
2786 entries.into_iter()
2787 .map(|(name, value)| Named::new(Scalar::from(name), Ok(AttrRead::new(value))))
2788 .collect()
2789 }
2790
2791 #[test]
2794 fn single_reading() {
2795 let reads = reads([("double_scalar", Value::from(2.5))]);
2796 let read = <f64 as ReadAttrSpec<&str>>::collect(reads).unwrap();
2797 assert_eq!(read.value, Some(2.5));
2798 }
2799
2800 #[test]
2803 fn readings_in_order() {
2804 let reads = reads([("first", Value::from(1.0)), ("second", Value::from(2.0))]);
2805 let [first, second] = <f64 as ReadAttrSpec<[&str; 2]>>::collect(reads).unwrap();
2806 assert_eq!(first.unwrap().value, Some(1.0));
2807 assert_eq!(second.unwrap().value, Some(2.0));
2808 }
2809
2810 #[test]
2813 fn readings_of_their_own_types() {
2814 let reads = reads([("double_scalar", Value::from(2.5)), ("string_scalar", Value::from("hello"))]);
2815 let (double, string) = <(f64, String) as ReadAttrSpec<(&str, &str)>>::collect(reads).unwrap();
2816 assert_eq!(double.unwrap().value, Some(2.5));
2817 assert_eq!(string.unwrap().value.as_deref(), Some("hello"));
2818 }
2819
2820 #[test]
2823 fn missing_reading() {
2824 let reads = reads([("first", Value::from(1.0))]);
2825 let err = <f64 as ReadAttrSpec<[&str; 2]>>::collect(reads).unwrap_err();
2826 assert_eq!(err.kind(), ErrorKind::Internal);
2827 }
2828
2829 #[test]
2832 fn every_reading_named() {
2833 let reads = reads([("first", Value::from(1.0)), ("second", Value::from(2.0))]);
2834 let all = <Value as ReadAttrSpec<All>>::collect(reads).unwrap();
2835 let names = all.iter().map(|read| read.name.value()).collect::<Vec<_>>();
2836 assert_eq!(names, ["first", "second"]);
2837 }
2838
2839 #[test]
2842 fn write_outcomes() {
2843 let writes = vec![Ok(()), Err(Error::new(ErrorKind::AttrNotWritable))];
2844 let (first, second) = <((&str, f64), (&str, f64)) as WriteAttrSpec>::collect(writes).unwrap();
2845 assert!(first.is_ok());
2846 assert_eq!(second.unwrap_err().kind(), ErrorKind::AttrNotWritable);
2847 }
2848
2849 #[test]
2852 fn unexpected_outcome() {
2853 let writes = vec![Ok(())];
2854 let err = <((&str, f64), (&str, f64)) as WriteAttrSpec>::collect(writes).unwrap_err();
2855 assert_eq!(err.kind(), ErrorKind::Internal);
2856 }
2857
2858}