1mod orb;
4mod tango;
5mod zmq;
6
7use std::sync::{Arc, OnceLock};
8use std::time::SystemTime;
9use std::future::Future;
10use std::borrow::Cow;
11use std::pin::Pin;
12use std::task;
13
14use futures::future::{ready, BoxFuture};
15use futures::stream::BoxStream;
16use tracing::{debug, trace, warn};
17use uuid::Uuid;
18
19use crate::dev::{AttrConfig, AttrInfo, AttrQuality, AttrRead, AttrValueEvent, CommandInfo, DevInfo, DevSource, DevState, Named, NamedResult};
20use crate::backend::{Backend, BackendDev, BackendServer, BackendServerConfig};
21use crate::error::{Error, ErrorKind, Result};
22use crate::value::{Scalar, Spectrum, Value};
23use crate::future::resolve;
24
25use orb::client::{Client, ClientObject, ClientRequest, ClientReply};
26use orb::server::{Server, ServerObject, ServerRequest, ServerReply};
27use orb::cdr::{CdrWrite, CdrRead, CdrReader, CdrWriter};
28use orb::giop::ReplyStatus;
29use zmq::{Consumer, Supplier};
30use tango::{
31 tango_repo_id,
32 ClientIdent,
33 NamedError,
34 write_command_data, read_command_data,
35 read_attr, write_attr,
36 read_command_info, read_attr_info,
37 read_dev_info, write_dev_info,
38 write_command_info, write_attr_info,
39};
40
41
42const ALL_ATTRIBUTES: &str = "All attributes";
44const ALL_ATTRIBUTES_3: &str = "All attributes_3";
46
47static CLIENT_ID: ClientIdent = ClientIdent::Java {
49 main_class: Cow::Borrowed("rstango"),
50 uuid: Uuid::nil(),
51};
52
53#[derive(Debug)]
59pub struct CorbaBackend {
60 client: OnceLock<Client>,
62 consumer: Arc<OnceLock<Consumer>>,
65}
66
67impl CorbaBackend {
68
69 pub fn new() -> Arc<Self> {
70 Arc::new(Self {
71 client: OnceLock::new(),
72 consumer: Arc::new(OnceLock::new()),
73 })
74 }
75
76 fn ensure_client(&self) -> Result<&Client> {
77
78 if let Some(client) = self.client.get() {
79 return Ok(client);
80 }
81
82 let client = Client::new()?;
83 Ok(self.client.get_or_init(|| client))
84
85 }
86
87 async fn check_device(&self, object: ClientObject) -> Result<Arc<dyn BackendDev>> {
89
90 let mut versions_it = DevVersion::ALL.into_iter().rev();
94 let version = loop {
95
96 let Some(version) = versions_it.next() else {
97 return Err(Error::new(ErrorKind::DevNotConnected)
98 .with_append_description(format_args!("Object {object:?}")));
99 };
100
101 let reply = object.request(b"_is_a", &version.repo_id()).await.map_err(corba_err)?;
102 let (mut reader, exception_id) = reply.read().map_err(corba_err)?;
103
104 if let Some(exception_id) = exception_id.as_deref() {
105 return Err(unexpected_user_exception_err(exception_id));
106 }
107
108 if reader.read_bool()? {
109 break version;
110 }
111
112 };
113
114 debug!(?version, "Connected to the device over CORBA");
115
116 Ok(Arc::new(CorbaDevBackend {
117 object,
118 version,
119 consumer: Arc::clone(&self.consumer),
120 }))
121
122 }
123
124}
125
126impl Backend for CorbaBackend {
127
128 fn connect(&self, itr: &str) -> BoxFuture<'_, Result<Arc<dyn BackendDev>>> {
129 if let Some(_corbaloc) = itr.strip_prefix("corbaloc:iiop:") {
130 Box::pin(ready(Err(Error::not_implemented()
131 .with_append_description("Scheme corbaloc:iiop:"))))
132 } else if let Some(ior) = itr.strip_prefix("IOR:") {
133 let ior = ior.to_string();
134 Box::pin(async move {
135 let client = self.ensure_client()?;
136 let object = client.connect_object_ior(&ior, &DevVersion::REPO_ID).await.map_err(corba_err)?;
137 self.check_device(object).await
138 })
139 } else {
140 Box::pin(ready(Err(Error::new(ErrorKind::NotHandled))))
141 }
142 }
143
144 fn connect_at(&self, host: &str, port: u16, name: &str) -> BoxFuture<'_, Result<Arc<dyn BackendDev>>> {
145 let host = host.to_string();
146 let object: Arc<[u8]> = Arc::from(name.to_ascii_lowercase().into_bytes());
147 Box::pin(async move {
148 let client = self.ensure_client()?;
149 let object = client.connect_object_iiop(&host, port, None, None, object).await.map_err(corba_err)?;
150 self.check_device(object).await
151 })
152 }
153
154 fn serve<'a>(&'a self, config: &'a BackendServerConfig) -> BoxFuture<'a, Result<Arc<dyn BackendServer>>> {
155 Box::pin(async move {
156
157 let host = config.host.as_deref().unwrap_or("0.0.0.0");
160 let port = config.port.unwrap_or(0);
161
162 let addrs = resolve(host, port).await?;
163
164 let mut last_err = None;
167 for addr in addrs {
168 match Server::bind(addr) {
169 Ok(server) => {
170
171 debug!(addr = %server.local_addr(), addrs = ?server.addrs(), "Serving devices over CORBA");
172
173 let supplier = config.admin_name.as_deref()
174 .map(|admin_name| Supplier::new(
175 admin_name,
176 config.tango_host.as_ref().map(|(host, port)| (&**host, *port)),
177 &server.host(),
178 server.local_addr().port(),
179 server.local_addr().ip(),
180 server.addrs(),
181 ));
182
183 return Ok(Arc::new(CorbaServerBackend { server, supplier }) as Arc<dyn BackendServer>);
184
185 }
186 Err(err) => last_err = Some(err),
187 }
188 }
189
190 Err(last_err.unwrap_or_else(|| Error::new(ErrorKind::Corba)
191 .with_append_description(format_args!("No address to listen on for {host}:{port}"))))
192
193 })
194 }
195
196}
197
198#[derive(Debug)]
204struct CorbaDevBackend {
205 object: ClientObject,
206 version: DevVersion,
207 consumer: Arc<OnceLock<Consumer>>,
209}
210
211impl CorbaDevBackend {
212
213 fn request(&self, operation: &[u8], input: &dyn CdrWrite) -> CorbaDevRequest {
219 CorbaDevRequest { inner: self.object.request(operation, input) }
220 }
221
222 fn consumer(&self) -> Result<&Consumer> {
226
227 if let Some(consumer) = self.consumer.get() {
228 return Ok(consumer);
229 }
230
231 let consumer = Consumer::start()?;
232 Ok(self.consumer.get_or_init(|| consumer))
233
234 }
235
236}
237
238impl BackendDev for CorbaDevBackend {
239
240 fn name(&self) -> BoxFuture<'_, Result<Arc<str>>> {
241 let request = self.request(b"_get_name", &());
242 Box::pin(async move {
243 let reply = request.await?;
244 Ok(Arc::from(reply.read::<String>()?))
245 })
246 }
247
248 fn admin_dev_name(&self) -> BoxFuture<'_, Result<Arc<str>>> {
249 let request = self.request(b"_get_adm_name", &());
250 Box::pin(async move {
251 let reply = request.await?;
252 Ok(Arc::from(reply.read::<String>()?))
253 })
254 }
255
256 fn description(&self) -> BoxFuture<'_, Result<Arc<str>>> {
257 let request = self.request(b"_get_description", &());
258 Box::pin(async move {
259 let reply = request.await?;
260 Ok(Arc::from(reply.read::<String>()?))
261 })
262 }
263
264 fn ping(&self) -> BoxFuture<'_, Result<()>> {
265 let request = self.request(b"ping", &());
266 Box::pin(async move {
267 let reply = request.await?;
268 reply.read::<()>()
269 })
270 }
271
272 fn info(&self) -> BoxFuture<'_, Result<DevInfo>> {
273
274 let (operation, info_version): (&[u8], DevVersion) = match self.version {
283 DevVersion::V1 |
284 DevVersion::V2 => (b"info", DevVersion::V1),
285 DevVersion::V3 |
286 DevVersion::V4 |
287 DevVersion::V5 => (b"info_3", DevVersion::V3),
288 DevVersion::V6 => (b"info_6", DevVersion::V6),
289 };
290
291 let request = self.request(operation, &());
292
293 Box::pin(async move {
294 let reply = request.await?;
295 let mut reader = reply.read_simple()?;
296 read_dev_info(&mut reader, info_version)
297 })
298
299 }
300
301 fn state(&self) -> BoxFuture<'_, Result<DevState>> {
302 let request = self.request(b"_get_state", &());
303 Box::pin(async move {
304 let reply = request.await?;
305 reply.read::<DevState>()
306 })
307 }
308
309 fn status(&self) -> BoxFuture<'_, Result<String>> {
310 let request = self.request(b"_get_status", &());
311 Box::pin(async move {
312 let reply = request.await?;
313 reply.read::<String>()
314 })
315 }
316
317 fn black_box(&self, max: Option<usize>) -> BoxFuture<'_, Result<Vec<String>>> {
318 let max = max.and_then(|n| i32::try_from(n).ok()).unwrap_or(i32::MAX);
319 let request = self.request(b"black_box", &max);
320 Box::pin(async move {
321 let reply = request.await?;
322 reply.read::<Vec<String>>()
323 })
324 }
325
326 fn command(&self, name: Scalar<str>, input: Value, source: DevSource) -> BoxFuture<'_, Result<Value>> {
327
328 struct Input<'a> {
334 version: DevVersion,
335 name: &'a str,
336 input: Value,
337 source: DevSource,
338 }
339
340 impl CdrWrite for Input<'_> {
341 fn write(&self, buf: &mut CdrWriter) -> Result<()> {
342 buf.write_string(self.name)?;
343 write_command_data(buf, &self.input)?;
344 if self.version >= DevVersion::V2 {
345 buf.write(&self.source)?;
346 }
347 if self.version >= DevVersion::V4 {
348 buf.write(&CLIENT_ID)?;
349 }
350 Ok(())
351 }
352 }
353
354 let operation: &[u8] = match self.version {
355 DevVersion::V1 => b"command_inout",
356 DevVersion::V2 |
357 DevVersion::V3 => b"command_inout_2",
358 DevVersion::V4 |
359 DevVersion::V5 |
360 DevVersion::V6 => b"command_inout_4",
361 };
362
363 let request = self.request(operation, &Input { version: self.version, name: name.value(), input, source });
364
365 Box::pin(async move {
366 let reply = request.await?;
367 read_command_data(&mut reply.read_simple()?)
368 })
369
370 }
371
372 fn read_attr(&self, names: Option<Spectrum<str>>, source: DevSource) -> BoxFuture<'_, Result<Vec<NamedResult<AttrRead<Value>>>>> {
373
374 struct Input<'a> {
382 version: DevVersion,
383 names: &'a dyn CdrWrite,
384 source: DevSource,
385 }
386
387 impl CdrWrite for Input<'_> {
388 fn write(&self, buf: &mut CdrWriter) -> Result<()> {
389 buf.write(&self.names)?;
390 if self.version >= DevVersion::V2 {
391 buf.write(&self.source)?;
392 }
393 if self.version >= DevVersion::V4 {
394 buf.write(&CLIENT_ID)?;
395 }
396 Ok(())
397 }
398 }
399
400 if let Some(names) = &names
401 && let Err(err) = check_no_all_marker(names) {
402 return Box::pin(ready(Err(err)));
403 }
404
405 let operation: &[u8] = match self.version {
406 DevVersion::V1 => b"read_attributes",
407 DevVersion::V2 => b"read_attributes_2",
408 DevVersion::V3 => b"read_attributes_3",
409 DevVersion::V4 => b"read_attributes_4",
410 DevVersion::V5 |
411 DevVersion::V6 => b"read_attributes_5",
412 };
413
414 let version = self.version;
415 let names = match &names {
416 Some(names) => names as &dyn CdrWrite,
417 None => &[ALL_ATTRIBUTES],
418 };
419
420 let request = self.request(operation, &Input { version, names, source });
421
422 Box::pin(async move {
423 let reply = request.await?;
424 read_attr_list(&mut reply.read_simple()?, version)
425 })
426
427 }
428
429 fn write_attr(&self, values: Vec<Named<Value>>) -> BoxFuture<'_, Result<Vec<Result<()>>>> {
430
431 struct Input<'a> {
440 version: DevVersion,
443 values: &'a [Named<Value>],
444 }
445
446 impl CdrWrite for Input<'_> {
447 fn write(&self, buf: &mut CdrWriter) -> Result<()> {
448 buf.write_sequence_len(self.values.len())?;
449 for Named { name, value } in self.values {
450 write_attr(buf, self.version, name.value(), &Ok(AttrRead {
451 value: None,
452 w_value: Some(value.clone()),
453 quality: AttrQuality::Valid,
454 time: SystemTime::UNIX_EPOCH,
455 }))?;
456 }
457 if self.version >= DevVersion::V4 {
458 buf.write(&CLIENT_ID)?;
459 }
460 Ok(())
461 }
462 }
463
464 let (operation, input_version): (&[u8], DevVersion) = match self.version {
470 DevVersion::V1 |
471 DevVersion::V2 => (b"write_attributes", DevVersion::V1),
472 DevVersion::V3 => (b"write_attributes_3", DevVersion::V1),
473 DevVersion::V4 |
474 DevVersion::V5 |
475 DevVersion::V6 => (b"write_attributes_4", DevVersion::V4),
476 };
477
478 let request = self.request(operation, &Input { version: input_version, values: &values });
479
480 let count = values.len();
481
482 Box::pin(async move {
483
484 let reply = request.await?;
485 let (mut reader, exception_id) = reply.read_raw()?;
486
487 let Some(exception_id) = exception_id else {
488 return Ok((0..count).map(|_| Ok(())).collect());
489 };
490
491 if exception_id.as_ref() != tango_repo_id!("MultiDevFailed") {
492 return Err(unexpected_user_exception_err(&exception_id));
493 }
494
495 let mut results: Vec<Result<()>> = (0..count).map(|_| Ok(())).collect();
499 for failure in reader.read::<Vec<NamedError>>()? {
500 if let Some(result) = results.get_mut(failure.index) {
501 *result = Err(failure.err.unwrap_or_else(|| Error::new(ErrorKind::Internal)
502 .with_description("The device reported a failed attribute write holding no error")));
503 }
504 }
505
506 Ok(results)
507
508 })
509
510 }
511
512 fn write_read_attr(&self, values: Vec<Named<Value>>) -> BoxFuture<'_, Result<Vec<NamedResult<AttrRead<Value>>>>> {
513
514 struct Input<'a> {
523 version: DevVersion,
524 values: &'a [Named<Value>],
525 }
526
527 impl CdrWrite for Input<'_> {
528 fn write(&self, buf: &mut CdrWriter) -> Result<()> {
529
530 buf.write_sequence_len(self.values.len())?;
534 for Named { name, value } in self.values {
535 write_attr(buf, DevVersion::V4, name.value(), &Ok(AttrRead {
536 value: None,
537 w_value: Some(value.clone()),
538 quality: AttrQuality::Valid,
539 time: SystemTime::UNIX_EPOCH,
540 }))?;
541 }
542
543 if self.version >= DevVersion::V5 {
547 buf.write_sequence_len(self.values.len())?;
548 for write in self.values {
549 buf.write_string(write.name.value())?;
550 }
551 }
552
553 buf.write(&CLIENT_ID)?;
554 Ok(())
555
556 }
557 }
558
559 let (operation, reply_version): (&[u8], DevVersion) = match self.version {
564 DevVersion::V1 |
565 DevVersion::V2 |
566 DevVersion::V3 => {
567 return Box::pin(ready(Err(Error::not_implemented()
568 .with_append_description("Writing and reading attributes back in a single request needs a device of IDL v4 or later")
569 .with_append_description(format_args!("This one is {:?}", self.version)))));
570 }
571 DevVersion::V4 => (b"write_read_attributes_4", DevVersion::V4),
572 DevVersion::V5 |
573 DevVersion::V6 => (b"write_read_attributes_5", DevVersion::V5),
574 };
575
576 let request = self.request(operation, &Input { version: self.version, values: &values });
577
578 let names = values.into_iter().map(|write| write.name).collect::<Vec<_>>();
581
582 Box::pin(async move {
583
584 let reply = request.await?;
585 let (mut reader, exception_id) = reply.read_raw()?;
586
587 let Some(exception_id) = exception_id else {
588 return read_attr_list(&mut reader, reply_version);
589 };
590
591 if exception_id.as_ref() != tango_repo_id!("MultiDevFailed") {
592 return Err(unexpected_user_exception_err(&exception_id));
593 }
594
595 let mut failures = reader.read::<Vec<NamedError>>()?;
599 let entries = names.into_iter().enumerate().map(|(index, name)| {
600 let err = failures.iter_mut()
601 .find(|failure| failure.index == index)
602 .and_then(|failure| failure.err.take())
603 .unwrap_or_else(|| Error::new(ErrorKind::Internal)
604 .with_description("The device wrote this attribute but read none of them back, another one having failed"));
605 Named::new(name, Err(err))
606 }).collect();
607
608 Ok(entries)
609
610 })
611
612 }
613
614 fn command_info(&self, names: Option<Spectrum<str>>) -> BoxFuture<'_, Result<Vec<NamedResult<CommandInfo>>>> {
615
616 let (query_operation, list_operation, info_version): (&[u8], &[u8], DevVersion) = match self.version {
631 DevVersion::V1 => (b"command_query", b"command_list_query", DevVersion::V1),
632 _ => (b"command_query_2", b"command_list_query_2", DevVersion::V2),
633 };
634
635 if let Some(names) = names {
636
637 let requests = (0..names.len())
641 .map(|index| names.get_scalar([index]).expect("a name at every index"))
642 .map(|name| {
643 let request = self.request(query_operation, &name.value());
644 (name, request)
645 })
646 .collect::<Vec<_>>();
647
648 Box::pin(async move {
649
650 let mut entries = Vec::with_capacity(requests.len());
651
652 for (name, request) in requests {
653
654 let res: Result<CommandInfo> = async {
655 let reply = request.await?;
656 let mut reader = reply.read_simple()?;
657 read_command_info(&mut reader, info_version)?.1
658 }.await;
659
660 match res {
666 Err(err) if matches!(err.kind(), ErrorKind::TimedOut | ErrorKind::DevNotConnected) => {
667 return Err(err);
668 }
669 res => entries.push(Named::new(name, res)),
670 }
671
672 }
673
674 Ok(entries)
675
676 })
677
678 } else {
679
680 let request = self.request(list_operation, &());
682
683 Box::pin(async move {
684 let reply = request.await?;
685 let mut reader = reply.read_simple()?;
686 let len = reader.read_sequence_len()?;
687 let mut entries = Vec::with_capacity(len);
688 for _ in 0..len {
689 let (name, info) = read_command_info(&mut reader, info_version)?;
690 entries.push(Named::new(Scalar::<str>::from(&*name), info));
691 }
692 Ok(entries)
693 })
694
695 }
696
697 }
698
699 fn set_attr_config(&self, configs: Vec<Named<AttrConfig>>) -> BoxFuture<'_, Result<Vec<Result<()>>>> {
700
701 struct Input<'a> {
715 version: DevVersion,
717 infos: &'a [(Scalar<str>, AttrInfo)],
718 with_client: bool,
720 }
721
722 impl CdrWrite for Input<'_> {
723 fn write(&self, buf: &mut CdrWriter) -> Result<()> {
724 buf.write_sequence_len(self.infos.len())?;
725 for (name, info) in self.infos {
726 write_attr_info(buf, self.version, name.value(), info)?;
727 }
728 if self.with_client {
729 buf.write(&CLIENT_ID)?;
730 }
731 Ok(())
732 }
733 }
734
735 let (operation, config_version, with_client): (&'static [u8], DevVersion, bool) = match self.version {
738 DevVersion::V1 |
739 DevVersion::V2 => (b"set_attribute_config", DevVersion::V1, false),
740 DevVersion::V3 => (b"set_attribute_config_3", DevVersion::V3, false),
741 DevVersion::V4 => (b"set_attribute_config_4", DevVersion::V3, true),
742 DevVersion::V5 |
743 DevVersion::V6 => (b"set_attribute_config_5", DevVersion::V5, true),
744 };
745
746 let names = configs.iter().map(|config| config.name.value()).collect::<Vec<_>>();
747 let current = self.attr_info(Some(Spectrum::<str>::from(&names[..])));
748 let count = configs.len();
749
750 Box::pin(async move {
751
752 let mut infos = Vec::with_capacity(count);
755 for (Named { name, value: config }, current) in configs.into_iter().zip(current.await?) {
756 let mut info = current.value?;
757 info.config = config;
758 infos.push((name, info));
759 }
760
761 let reply = self.request(operation, &Input { version: config_version, infos: &infos, with_client }).await?;
766 let (mut reader, exception_id) = reply.inner.read().map_err(corba_err)?;
767 match exception_id.as_deref() {
768 None => Ok((0..count).map(|_| Ok(())).collect()),
769 Some(tango_repo_id!("DevFailed")) => {
770 let err = reader.read::<Error>()?;
771 match count {
772 1 => Ok(vec![Err(err)]),
773 _ => Err(err),
774 }
775 }
776 Some(exception_id) => Err(unexpected_user_exception_err(exception_id)),
777 }
778
779 })
780
781 }
782
783 fn attr_info(&self, names: Option<Spectrum<str>>) -> BoxFuture<'_, Result<Vec<NamedResult<AttrInfo>>>> {
784
785 let (operation, config_version): (&[u8], DevVersion) = match self.version {
799 DevVersion::V1 => (b"get_attribute_config", DevVersion::V1),
800 DevVersion::V2 => (b"get_attribute_config_2", DevVersion::V2),
801 DevVersion::V3 |
802 DevVersion::V4 => (b"get_attribute_config_3", DevVersion::V3),
803 DevVersion::V5 |
804 DevVersion::V6 => (b"get_attribute_config_5", DevVersion::V5),
805 };
806
807 if let Some(names) = &names
808 && let Err(err) = check_no_all_marker(names) {
809 return Box::pin(ready(Err(err)));
810 }
811
812 let request = match &names {
816 Some(names) => self.request(operation, names),
817 None if config_version >= DevVersion::V3 => self.request(operation, &[ALL_ATTRIBUTES_3]),
820 None => self.request(operation, &[ALL_ATTRIBUTES]),
821 };
822
823 let expected = names;
825
826 Box::pin(async move {
827
828 let reply = request.await?;
829 let mut reader = reply.read_simple()?;
830 let len = reader.read_sequence_len()?;
831 let mut entries = Vec::with_capacity(len);
832
833 for index in 0..len {
834
835 let (name, info) = read_attr_info(&mut reader, config_version)?;
836
837 if let Some(expected) = expected.as_ref().and_then(|names| names.get([index]))
841 && !name.eq_ignore_ascii_case(expected) {
842 return Err(Error::new(ErrorKind::InvalidData)
843 .with_description("The device answered with the configuration of another attribute")
844 .with_append_description(format_args!("Got {name:?} where {expected:?} was requested")));
845 }
846
847 entries.push(Named::new(Scalar::<str>::from(&*name), info));
848
849 }
850
851 Ok(entries)
852
853 })
854
855 }
856
857 fn subscribe_attr_value<'a>(&'a self, admin: Arc<dyn BackendDev>, name: &'a str, event: AttrValueEvent) -> BoxFuture<'a, Result<BoxStream<'static, Result<AttrRead<Value>>>>> {
858 match self.consumer() {
859 Ok(consumer) => zmq::subscribe_attr_value(consumer, self.name(), admin, name, event),
860 Err(err) => Box::pin(ready(Err(err))),
861 }
862 }
863
864 fn subscribe_attr_ready<'a>(&'a self, admin: Arc<dyn BackendDev>, name: &'a str) -> BoxFuture<'a, Result<BoxStream<'static, Result<()>>>> {
865 match self.consumer() {
866 Ok(consumer) => zmq::subscribe_attr_ready(consumer, self.name(), admin, name),
867 Err(err) => Box::pin(ready(Err(err))),
868 }
869 }
870
871 fn subscribe_attr_conf<'a>(&'a self, admin: Arc<dyn BackendDev>, name: &'a str) -> BoxFuture<'a, Result<BoxStream<'static, Result<AttrInfo>>>> {
872 match self.consumer() {
873 Ok(consumer) => zmq::subscribe_attr_conf(consumer, self.name(), admin, name),
874 Err(err) => Box::pin(ready(Err(err))),
875 }
876 }
877
878 fn command_history<'a>(&'a self, _name: &'a str, _max: usize) -> BoxFuture<'a, Result<Vec<(SystemTime, Result<Value>)>>> {
879 Box::pin(ready(Err(Error::not_implemented()
880 .with_append_description("The recorded history of a command"))))
881 }
882
883 fn attr_history<'a>(&'a self, _name: &'a str, _max: usize) -> BoxFuture<'a, Result<Vec<Result<AttrRead<Value>>>>> {
884 Box::pin(ready(Err(Error::not_implemented()
885 .with_append_description("The recorded history of an attribute"))))
886 }
887
888}
889
890struct CorbaDevRequest {
892 inner: ClientRequest,
893}
894
895impl Future for CorbaDevRequest {
896 type Output = Result<CorbaDevReply>;
897 fn poll(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> task::Poll<Self::Output> {
898 Pin::new(&mut self.get_mut().inner).poll(cx)
899 .map(|inner| inner
900 .map(|inner| CorbaDevReply { inner })
901 .map_err(corba_err))
902 }
903}
904
905fn corba_err(err: Error) -> Error {
911 if err.kind() != ErrorKind::Corba {
912 return err;
913 }
914 Error::new(ErrorKind::DevNotConnected).with_parent(err)
915}
916
917#[derive(Debug)]
919struct CorbaDevReply {
920 inner: ClientReply,
921}
922
923impl CorbaDevReply {
924
925 fn read_raw(&self) -> Result<(CdrReader<'_>, Option<Cow<'_, str>>)> {
931 let (mut reader, exception_id) = self.inner.read().map_err(corba_err)?;
932 if let Some(tango_repo_id!("DevFailed")) = exception_id.as_deref() {
933 Err(match reader.read::<Error>() {
934 Ok(err) => err,
935 Err(err) => err,
936 })
937 } else {
938 Ok((reader, exception_id))
939 }
940 }
941
942 fn read_simple(&self) -> Result<CdrReader<'_>> {
946 let (reader, exception_id) = self.read_raw()?;
947 if let Some(exception_id) = exception_id.as_deref() {
948 Err(unexpected_user_exception_err(exception_id))
949 } else {
950 Ok(reader)
951 }
952 }
953
954 fn read<'a, O: CdrRead<'a>>(&'a self) -> Result<O> {
956 self.read_simple()?.read()
957 }
958
959}
960
961fn read_attr_list(reader: &mut CdrReader<'_>, version: DevVersion) -> Result<Vec<NamedResult<AttrRead<Value>>>> {
964 let len = reader.read_sequence_len()?;
965 let mut entries = Vec::with_capacity(len);
966 for _ in 0..len {
967 entries.push(read_attr(reader, version)?);
968 }
969 Ok(entries)
970}
971
972const SERVED_VERSION: DevVersion = DevVersion::V6;
983
984const LIB_VERSION: i32 = 1000;
988
989const SERVED_REPO_IDS: [&str; SERVED_VERSION.number() as usize] = {
993 let mut ids = [""; SERVED_VERSION.number() as usize];
994 let mut index = 0;
995 while index < ids.len() {
996 ids[index] = DevVersion::REPO_ID[SERVED_VERSION as usize - index];
997 index += 1;
998 }
999 ids
1000};
1001
1002#[derive(Debug)]
1004struct CorbaServerBackend {
1005 server: Server<CorbaServerObject>,
1009 supplier: Option<Supplier>,
1013}
1014
1015impl CorbaServerBackend {
1016
1017 fn supplier(&self) -> Result<&Supplier> {
1019 self.supplier.as_ref().ok_or_else(|| Error::not_implemented()
1020 .with_append_description("This server publishes no event over ZMQ"))
1021 }
1022
1023 fn served(&self, name: &str) -> Option<Arc<dyn BackendDev>> {
1026 self.server.get(name.as_bytes()).map(|object| Arc::clone(&object.dev))
1027 }
1028
1029 fn is_served(&self, name: &str) -> bool {
1032 self.server.is_registered(name.as_bytes())
1033 }
1034
1035}
1036
1037impl BackendServer for CorbaServerBackend {
1038
1039 fn add(&self, name: &str, dev: Arc<dyn BackendDev>) -> Result<Arc<str>> {
1040
1041 let name = Arc::<str>::from(name.to_ascii_lowercase());
1046 let itr = self.server.register(Arc::new(CorbaServerObject {
1047 name: Arc::clone(&name),
1048 dev,
1049 })).map(Arc::<str>::from);
1050
1051 if itr.is_ok() {
1052 debug!(dev = %name, "Serving the device over CORBA");
1053 }
1054
1055 itr
1056
1057 }
1058
1059 fn remove(&self, name: &str) -> bool {
1060 let id = name.to_ascii_lowercase().into_bytes();
1061 self.server.deregister(&id)
1062 }
1063
1064 fn host(&self) -> Arc<str> {
1065 self.server.host()
1066 }
1067
1068 fn port(&self) -> u16 {
1069 self.server.local_addr().port()
1070 }
1071
1072 fn version(&self) -> i32 {
1073 SERVED_VERSION.number() as i32
1074 }
1075
1076 fn stop(&self) {
1077 if let Some(supplier) = &self.supplier {
1078 supplier.stop();
1079 }
1080 self.server.stop();
1081 }
1082
1083 fn wait(&self) -> BoxFuture<'_, ()> {
1084 Box::pin(self.server.wait())
1085 }
1086
1087 fn zmq_event_subscription_change(&self, args: Spectrum<str>) -> BoxFuture<'_, Result<(Spectrum<i32>, Spectrum<str>)>> {
1088 Box::pin(async move {
1089 let supplier = self.supplier()?;
1090 supplier.subscription_change(&args, |name| self.served(name), SERVED_VERSION.number()).await
1091 })
1092 }
1093
1094 fn event_confirm_subscription(&self, args: Spectrum<str>) -> BoxFuture<'_, Result<()>> {
1095 Box::pin(async move {
1096 let supplier = self.supplier()?;
1097 supplier.confirm_subscription(&args, |name| self.is_served(name)).await
1098 })
1099 }
1100
1101}
1102
1103#[derive(Debug)]
1107struct CorbaServerObject {
1108 name: Arc<str>,
1111 dev: Arc<dyn BackendDev>,
1113}
1114
1115impl ServerObject for CorbaServerObject {
1116
1117 type RequestFuture = BoxFuture<'static, Result<ServerReply>>;
1118
1119 fn id(&self) -> &[u8] {
1120 self.name.as_bytes()
1121 }
1122
1123 fn repo_ids(&self) -> &[&str] {
1124 &SERVED_REPO_IDS
1125 }
1126
1127 fn request(self: Arc<Self>, request: ServerRequest) -> Self::RequestFuture {
1128 Box::pin(async move { self.handle_request(&request).await })
1131 }
1132
1133}
1134
1135fn answer(request: &ServerRequest, res: Result<ServerReply>) -> Result<ServerReply> {
1142
1143 let err = match res {
1144 Ok(reply) => return Ok(reply),
1145 Err(err) => err,
1146 };
1147
1148 let mut reply = request.reply_user_exception(tango_repo_id!("DevFailed"))?;
1149 reply.write(&err)?;
1150
1151 Ok(reply)
1152
1153}
1154
1155impl CorbaServerObject {
1156
1157 async fn handle_request(&self, request: &ServerRequest) -> Result<ServerReply> {
1160
1161 let operation = request.operation();
1162 let res = self.dispatch(operation, request).await;
1163
1164 if let Err(err) = &res {
1165 trace!(dev = %self.name, operation = %String::from_utf8_lossy(operation), %err, "Failed to run an operation");
1166 }
1167
1168 answer(request, res)
1169
1170 }
1171
1172 async fn dispatch<'d>(&self, operation: &[u8], request: &'d ServerRequest) -> Result<ServerReply> {
1178
1179 let mut input = request.read();
1180
1181 let mut reply = request.reply(ReplyStatus::NoException)?;
1185 let output = &mut reply;
1186
1187 match operation {
1188 b"_get_name" => output.write_string(&self.dev.name().await?)?,
1189 b"_get_adm_name" => output.write_string(&self.dev.admin_dev_name().await?)?,
1190 b"_get_description" => output.write_string(&self.dev.description().await?)?,
1191 b"_get_state" => output.write(&self.dev.state().await?)?,
1192 b"_get_status" => output.write_string(&self.dev.status().await?)?,
1193 b"ping" => self.dev.ping().await?,
1194
1195 b"info" | b"info_3" | b"info_6" => {
1200 let version = match operation {
1201 b"info" => DevVersion::V1,
1202 b"info_3" => DevVersion::V3,
1203 _ => DevVersion::V6,
1204 };
1205 write_dev_info(output, version, &self.dev.info().await?)?;
1206 }
1207 b"black_box" => {
1208
1209 let max = input.read_long()?;
1212 let entries = self.dev.black_box(usize::try_from(max).ok()).await?;
1213 output.write(&entries)?;
1214
1215 }
1216 b"command_inout" |
1217 b"command_inout_2" |
1218 b"command_inout_4" => {
1219
1220 let name = input.read::<Scalar<str>>()?;
1229 let arg = read_command_data(&mut input)?;
1230
1231 let source = if operation == b"command_inout" {
1234 DevSource::Dev
1235 } else {
1236 input.read()?
1237 };
1238
1239 let answer = self.dev.command(name, arg, source).await?;
1240 write_command_data(output, &answer)?;
1241
1242 }
1243 b"read_attributes" |
1244 b"read_attributes_2" |
1245 b"read_attributes_3" |
1246 b"read_attributes_4" |
1247 b"read_attributes_5" => {
1248
1249 let version = match operation {
1260 b"read_attributes" => DevVersion::V1,
1261 b"read_attributes_2" => DevVersion::V2,
1262 b"read_attributes_3" => DevVersion::V3,
1263 b"read_attributes_4" => DevVersion::V4,
1264 _ => DevVersion::V5,
1265 };
1266
1267 let names = input.read::<Spectrum<str>>()?;
1268 let source = if version >= DevVersion::V2 {
1269 input.read()?
1270 } else {
1271 DevSource::Dev
1272 };
1273
1274 let readings = self.dev.read_attr(all_attributes(names), source).await?;
1275
1276 output.write_sequence_len(readings.len())?;
1277 for Named { name, value: read } in &readings {
1278 write_attr(output, version, name.value(), read)?;
1279 }
1280
1281 }
1282 b"write_attributes" |
1283 b"write_attributes_3" |
1284 b"write_attributes_4" => {
1285
1286 let version = if operation == b"write_attributes_4" {
1296 DevVersion::V4
1297 } else {
1298 DevVersion::V1
1299 };
1300
1301 let values = read_attr_values(&mut input, version)?;
1302 let names = values.iter().map(|write| write.name.clone()).collect::<Vec<_>>();
1303
1304 let results = self.dev.write_attr(values).await?;
1305
1306 let mut failures = results.into_iter()
1311 .enumerate()
1312 .filter_map(|(index, res)| res.err().map(|err| (index, err)))
1313 .collect::<Vec<_>>();
1314
1315 if !failures.is_empty() {
1316
1317 if version < DevVersion::V4 {
1321 return Err(failures.swap_remove(0).1);
1322 }
1323
1324 *output = request.reply_user_exception(tango_repo_id!("MultiDevFailed"))?;
1327 output.write_sequence_len(failures.len())?;
1328 for (index, err) in failures {
1329 output.write(&NamedError {
1330 name: Cow::Borrowed(names.get(index).map(|name| name.value()).unwrap_or("")),
1331 index,
1332 err: Some(err),
1333 })?;
1334 }
1335
1336 }
1337
1338 }
1339 b"write_read_attributes_4" |
1340 b"write_read_attributes_5" => {
1341
1342 let reply_version = if operation == b"write_read_attributes_5" {
1351 DevVersion::V5
1352 } else {
1353 DevVersion::V4
1354 };
1355
1356 let values = read_attr_values(&mut input, DevVersion::V4)?;
1357
1358 let readings = self.dev.write_read_attr(values).await?;
1362
1363 output.write_sequence_len(readings.len())?;
1364 for Named { name, value: read } in &readings {
1365 write_attr(output, reply_version, name.value(), read)?;
1366 }
1367
1368 }
1369 b"command_query" |
1370 b"command_query_2" => {
1371
1372 let version = if operation == b"command_query_2" {
1377 DevVersion::V2
1378 } else {
1379 DevVersion::V1
1380 };
1381
1382 let name = input.read::<Scalar<str>>()?;
1383 let names = name.clone().reshape([1]).expect("one name");
1384
1385 let infos = self.dev.command_info(Some(names)).await?;
1388
1389 let Some(Named { name, value: info }) = infos.into_iter().next() else {
1390 return Err(Error::new(ErrorKind::CommandNotFound)
1391 .with_append_description(format_args!("Command {name:?}")));
1392 };
1393
1394 write_command_info(output, version, name.value(), &info?)?;
1395
1396 }
1397 b"command_list_query" |
1398 b"command_list_query_2" => {
1399
1400 let version = if operation == b"command_list_query_2" {
1405 DevVersion::V2
1406 } else {
1407 DevVersion::V1
1408 };
1409
1410 let infos = self.dev.command_info(None).await?;
1411
1412 let out = &mut **output;
1415 let len = SequenceLen::start(out)?;
1416 let mut count = 0;
1417
1418 for Named { name, value: info } in infos {
1419
1420 let info = match info {
1423 Ok(info) => info,
1424 Err(err) => {
1425 warn!(command = %name.value(), %err, "Failed to describe the command, leaving it out of the listing");
1426 continue;
1427 }
1428 };
1429
1430 write_command_info(out, version, name.value(), &info)?;
1431 count += 1;
1432
1433 }
1434
1435 len.finish(out, count)?;
1436
1437 }
1438 b"set_attribute_config" |
1439 b"set_attribute_config_3" |
1440 b"set_attribute_config_4" |
1441 b"set_attribute_config_5" => {
1442
1443 let version = match operation {
1455 b"set_attribute_config" => DevVersion::V1,
1456 b"set_attribute_config_5" => DevVersion::V5,
1457 _ => DevVersion::V3,
1458 };
1459
1460 let len = input.read_sequence_len()?;
1461 let mut configs = Vec::with_capacity(len);
1462 for _ in 0..len {
1463 let (name, info) = read_attr_info(&mut input, version)?;
1464 configs.push(Named::new(Scalar::from(&*name), info?.config));
1465 }
1466
1467 let results = self.dev.set_attr_config(configs).await?;
1470 if let Some(err) = results.into_iter().find_map(Result::err) {
1471 return Err(err);
1472 }
1473
1474 }
1475 b"get_attribute_config" |
1476 b"get_attribute_config_2" |
1477 b"get_attribute_config_3" |
1478 b"get_attribute_config_5" => {
1479
1480 let version = match operation {
1489 b"get_attribute_config" => DevVersion::V1,
1490 b"get_attribute_config_2" => DevVersion::V2,
1491 b"get_attribute_config_3" => DevVersion::V3,
1492 _ => DevVersion::V5,
1493 };
1494
1495 let names = input.read::<Spectrum<str>>()?;
1496 let requested = all_attributes(names);
1497 let named = requested.is_some();
1498
1499 let infos = self.dev.attr_info(requested).await?;
1500
1501 let out = &mut **output;
1502 let len = SequenceLen::start(out)?;
1503 let mut count = 0;
1504
1505 for Named { name, value: info } in infos {
1506
1507 let info = match info {
1512 Ok(info) => info,
1513 Err(err) if named => return Err(err),
1514 Err(err) => {
1515 warn!(attr = %name.value(), %err, "Failed to describe the attribute, leaving it out of the listing");
1516 continue;
1517 }
1518 };
1519
1520 write_attr_info(out, version, name.value(), &info)?;
1521 count += 1;
1522
1523 }
1524
1525 len.finish(out, count)?;
1526
1527 }
1528 b"command_inout_history_2" |
1542 b"command_inout_history_4" |
1543 b"read_attribute_history_2" |
1544 b"read_attribute_history_3" |
1545 b"read_attribute_history_4" |
1546 b"read_attribute_history_5" |
1547 b"get_pipe_config_5" |
1548 b"set_pipe_config_5" |
1549 b"read_pipe_5" |
1550 b"write_pipe_5" |
1551 b"write_read_pipe_5" => {
1552 return Err(Error::not_implemented()
1553 .with_append_description(format_args!("Operation {:?}, in this server", String::from_utf8_lossy(operation))));
1554 }
1555 _ => {
1556 return Err(Error::new(ErrorKind::NotHandled));
1557 }
1558 }
1559
1560 Ok(reply)
1561
1562 }
1563
1564}
1565
1566fn all_attributes(names: Spectrum<str>) -> Option<Spectrum<str>> {
1570 if has_all_marker(&names) {
1571 None
1572 } else {
1573 Some(names)
1574 }
1575}
1576
1577fn has_all_marker(names: &Spectrum<str>) -> bool {
1580 names.iter().any(|name| name == ALL_ATTRIBUTES || name == ALL_ATTRIBUTES_3)
1581}
1582
1583fn check_no_all_marker(names: &Spectrum<str>) -> Result<()> {
1586 if has_all_marker(names) {
1587 return Err(Error::new(ErrorKind::AttrNameInvalid)
1588 .with_append_description("Given attribute names contains the 'all attributes' marker")
1589 .with_append_description(format_args!("{names:?}")));
1590 }
1591 Ok(())
1592}
1593
1594fn read_attr_values(input: &mut CdrReader<'_>, version: DevVersion) -> Result<Vec<Named<Value>>> {
1598
1599 let len = input.read_sequence_len()?;
1600 let mut values = Vec::with_capacity(len);
1601
1602 for _ in 0..len {
1603
1604 let Named { name, value: read } = read_attr(input, version)?;
1605
1606 let value = read?.w_value.ok_or_else(|| Error::new(ErrorKind::InvalidData)
1609 .with_description(format!("The value to write of the attribute {:?} is missing", name.value())))?;
1610
1611 values.push(Named::new(name, value));
1612
1613 }
1614
1615 Ok(values)
1616
1617}
1618
1619struct SequenceLen(usize);
1623
1624impl SequenceLen {
1625
1626 fn start(buf: &mut CdrWriter) -> Result<Self> {
1628 let pos = buf.position();
1629 buf.write_sequence_len(0)?;
1630 Ok(Self(pos))
1631 }
1632
1633 fn finish(self, buf: &mut CdrWriter, len: usize) -> Result<()> {
1636 let end = buf.position();
1637 buf.set_position(self.0);
1638 buf.write_sequence_len(len)?;
1639 buf.set_position(end);
1640 Ok(())
1641 }
1642
1643}
1644
1645#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
1647pub(crate) enum DevVersion {
1648 V1,
1649 V2,
1650 V3,
1651 V4,
1652 V5,
1653 V6,
1654}
1655
1656impl DevVersion {
1657
1658 pub const ALL: [Self; 6] = [
1660 Self::V1,
1661 Self::V2,
1662 Self::V3,
1663 Self::V4,
1664 Self::V5,
1665 Self::V6,
1666 ];
1667
1668 const REPO_ID: [&str; 6] = [
1669 tango_repo_id!("Device"),
1670 tango_repo_id!("Device_2"),
1671 tango_repo_id!("Device_3"),
1672 tango_repo_id!("Device_4"),
1673 tango_repo_id!("Device_5"),
1674 tango_repo_id!("Device_6"),
1675 ];
1676
1677 #[inline]
1678 pub const fn repo_id(&self) -> &'static str {
1679 Self::REPO_ID[*self as usize]
1680 }
1681
1682 #[inline]
1685 pub const fn number(&self) -> u32 {
1686 *self as u32 + 1
1687 }
1688
1689 #[inline]
1692 pub const fn from_number(number: u32) -> Option<Self> {
1693 match number {
1694 1..=6 => Some(Self::ALL[number as usize - 1]),
1695 _ => None,
1696 }
1697 }
1698
1699}
1700
1701fn unexpected_user_exception_err<'a>(exception_id: &'a str) -> Error {
1702 Error::new(ErrorKind::Internal)
1703 .with_description(format!("Unexpected user exception: {exception_id}"))
1704}
1705
1706#[cfg(test)]
1707mod tests {
1708
1709 use std::net::TcpListener;
1710 use std::time::Duration;
1711
1712 use crate::future::block_on_timeout;
1713
1714 use super::*;
1715
1716 #[test]
1719 fn connect_refused() {
1720
1721 let backend = CorbaBackend::new();
1722 let port = TcpListener::bind("127.0.0.1:0").unwrap().local_addr().unwrap().port();
1723
1724 let err = block_on_timeout(backend.connect_at("127.0.0.1", port, "test/dev/1"), Duration::from_secs(5))
1725 .unwrap()
1726 .unwrap_err();
1727
1728 assert_eq!(err.kind(), ErrorKind::DevNotConnected);
1729 assert_eq!(err.parent().unwrap().kind(), ErrorKind::Corba);
1730
1731 }
1732
1733}