Skip to main content

tango/corba/
mod.rs

1//! Version 1 of the backend, the one currently implemented by every Tango implementation.
2
3mod 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
42/// Special name for reading all attributes.
43const ALL_ATTRIBUTES: &str = "All attributes";
44/// Special name for reading all attributes on IDL v3+.
45const ALL_ATTRIBUTES_3: &str = "All attributes_3";
46
47/// Static client identifier.
48static CLIENT_ID: ClientIdent = ClientIdent::Java { 
49    main_class: Cow::Borrowed("rstango"), 
50    uuid: Uuid::nil(),
51};
52
53// =========================== //
54//        Common backend       //
55// =========================== //
56
57/// A pure rust CORBA backend.
58#[derive(Debug)]
59pub struct CorbaBackend {
60    /// The ORB client.
61    client: OnceLock<Client>,
62    /// The event consumer, shared with every device connected through this backend, and
63    /// only started by the first subscription.
64    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    /// Internal async function to check if a device is 
88    async fn check_device(&self, object: ClientObject) -> Result<Arc<dyn BackendDev>> {
89
90        // Iterate over the versions to find the largest one supported. We iterate in
91        // reverse order because nowadays it's much more likely to encounter a recent
92        // version than an old one!
93        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            // The port the server is bound to is what the references it hands out then
158            // point at, see [`CorbaServerBackend::build_ior`].
159            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            // A name is bound at the first of its addresses that can be, as the standard
165            // library does.
166            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// =========================== //
199//        Client backend       //
200// =========================== //
201
202/// Our internal backend to device proxies.
203#[derive(Debug)]
204struct CorbaDevBackend {
205    object: ClientObject,
206    version: DevVersion,
207    /// The event consumer of the backend, see [`Self::consumer`].
208    consumer: Arc<OnceLock<Consumer>>,
209}
210
211impl CorbaDevBackend {
212
213    /// Send a request to this device, over the object's own connection.
214    ///
215    /// The request goes out as soon as this returns, and not once the future it gives
216    /// back is polled, see [`ClientObject::request`]: whatever was borrowed to build
217    /// `input` need not outlive this call, only the future it returns has to be awaited.
218    fn request(&self, operation: &[u8], input: &dyn CdrWrite) -> CorbaDevRequest {
219        CorbaDevRequest { inner: self.object.request(operation, input) }
220    }
221
222    /// The event consumer of the backend, started if this is the first subscription. Two
223    /// first subscriptions at once may both start one, the one that is not kept stopping
224    /// as soon as it is dropped.
225    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        // IDL variants:
275        // DevInfo info()
276        // DevInfo_3 info_3()
277        // DevInfo_6 info_6()
278        //
279        // The second member is the version of the structure the info is read in, which is
280        // the one of the operation and not the one of the device: v2 answers with the v1
281        // structure, and v4 and v5 with the v3 one.
282        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        // IDL variants:
329        // any command_inout(in string command, in any argin)
330        // any command_inout_2(in string command, in any argin, in DevSource source)
331        // any command_inout_4(in string command, in any argin, in DevSource source, in ClntIdent cl_ident)
332
333        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        // IDL variants:
375        // AttributeValueList read_attributes(in DevVarStringArray names)
376        // AttributeValueList read_attributes_2(in DevVarStringArray names, in DevSource source)
377        // AttributeValueList_3 read_attributes_3(in DevVarStringArray names, in DevSource source)
378        // AttributeValueList_4 read_attributes_4(in DevVarStringArray names, in DevSource source, in ClntIdent cl_ident)
379        // AttributeValueList_5 read_attributes_5(in DevVarStringArray names, in DevSource source, in ClntIdent cl_ident)
380
381        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        // IDL variants:
432        // void write_attributes(in AttributeValueList values)
433        // void write_attributes_3(in AttributeValueList values)
434        // void write_attributes_4(in AttributeValueList_4 values, in ClntIdent cl_ident)
435        //
436        // Note that v2 has no variant of its own and reuses the v1 one, and that v3 only
437        // adds MultiDevFailed to the raised exceptions, the argument staying the same.
438
439        struct Input<'a> {
440            /// The version of the structure the values go out in, which is the one of the
441            /// operation and *not* the one of the device, see below.
442            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        // The second member is the version of the structure the values are written in,
465        // which is the one of the operation and not the one of the device: v3 takes the v1
466        // structure like v1 and v2 do, and v5 and v6 have no write operation of their own
467        // so they take the v4 one. Sending the structure of the device version instead
468        // makes the server fail to unmarshal the request entirely.
469        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            // A `MultiDevFailed` only names the attributes that failed to write, the
496            // ones behind them in the list having gone through: unlike a `DevFailed`,
497            // this is per-value and not the request as a whole.
498            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        // IDL variants:
515        // AttributeValueList_4 write_read_attributes_4(in AttributeValueList_4 values, in ClntIdent cl_ident)
516        // AttributeValueList_5 write_read_attributes_5(in AttributeValueList_4 values, in DevVarStringArray r_names, in ClntIdent cl_ident)
517        //
518        // Note that this operation only exists from v4 onward, and that v5 is the first
519        // one able to read back attributes other than the written ones. Both variants take
520        // the v4 structure for what they write, only what they answer with differs.
521
522        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                // Hardcoded v4 and not the device's version: both operations take the very
531                // same AttributeValueList_4, v5 having never added a list of its own for
532                // what it writes, only for what it answers with.
533                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                // Only v5 onward has this second list, of the attributes to read back, and
544                // we have no way of telling which ones the caller wants apart from the
545                // written ones, so it reads those back.
546                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        // The second member is the version the *reply* is read with, which is the one of
560        // the operation and not the one of the device: v4 answers with its own list, v5
561        // with the one it added, and v6 reuses the v5 operation and therefore its list.
562        // So a v6 device is read back as a v5 one here, unlike everywhere else.
563        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        // Only needed if the write raises `MultiDevFailed`, to name the failure of every
579        // value nothing more specific was reported for.
580        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            // A `MultiDevFailed` here means none of the values was read back at all, so
596            // every one of them is reported as a failure, the ones it does not name
597            // getting a generic error of their own.
598            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        // IDL variants:
617        // DevCmdInfoList command_list_query()
618        // DevCmdInfo command_query(in string command)
619        // DevCmdInfoList_2 command_list_query_2()
620        // DevCmdInfo_2 command_query_2(in string command)
621        //
622        // Note that these take no name list, so querying a subset means either one
623        // request per name or filtering the whole list afterward. We do the former, which
624        // is what cppTango does too. The v2 variants only add the display level to the
625        // returned structures, and no version after them has one of its own.
626
627        // The second member is the version of the structure the infos are read in, which
628        // is the one of the operation and not the one of the device, every version from
629        // v2 onward answering with the v2 structure.
630        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            // One request per name, sent right away and all in flight together, each
638            // answering with the info of that very command, so there is no order to
639            // check here, see [`ClientObject::request`].
640            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                    // The request of a single command failing, an unknown command name
661                    // being the usual reason, only fails that one command: it is
662                    // reported for it and the following ones are still awaited, so
663                    // that the caller gets exactly one answer for each name it gave.
664                    // Only a failure that makes the rest pointless aborts the batch.
665                    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            // List everything!
681            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        // IDL variants:
702        // void set_attribute_config(in AttributeConfigList new_conf)
703        // void set_attribute_config_3(in AttributeConfigList_3 new_conf)
704        // void set_attribute_config_4(in AttributeConfigList_3 new_conf, in ClntIdent cl_ident)
705        // void set_attribute_config_5(in AttributeConfigList_5 new_conf, in ClntIdent cl_ident)
706        //
707        // Note that v2 has no variant of its own and reuses the v1 one, as v4 does with the
708        // v3 structure. The structure sent is the whole one a configuration is read in,
709        // what the attribute is included: its type above all, which is what its limits are
710        // written as. That part is never changed by the device, so it is taken from the
711        // attribute itself, asked for first, which is what a client of cppTango does as
712        // well, reading a configuration before sending it back changed.
713
714        struct Input<'a> {
715            /// The version of the structure the configurations go out in.
716            version: DevVersion,
717            infos: &'a [(Scalar<str>, AttrInfo)],
718            /// Whether the operation takes the identification of the client after them.
719            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        // The second member is the version of the structure the configurations are
736        // written in, which is the one of the operation and not the one of the device.
737        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            // An attribute that cannot be described fails the whole call, which is what
753            // the device would do of it anyway, a single DevFailed being all it raises.
754            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            // The device raises a single DevFailed for the first attribute it failed to
762            // change, the ones before having been changed and the ones behind not even
763            // tried, which it does not tell apart: only a single attribute is known to be
764            // the one it is about, and the call fails as a whole otherwise.
765            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        // IDL variants:
786        // AttributeConfigList get_attribute_config(in DevVarStringArray names)
787        // AttributeConfigList_2 get_attribute_config_2(in DevVarStringArray names)
788        // AttributeConfigList_3 get_attribute_config_3(in DevVarStringArray names)
789        // AttributeConfigList_5 get_attribute_config_5(in DevVarStringArray names)
790        //
791        // Note that v4 has no variant of its own and reuses the v3 one, and that the
792        // 'all attributes' marker to pass for reading them all is version-dependent
793        // here, ALL_ATTRIBUTES_3 from v3 onward and ALL_ATTRIBUTES before.
794
795        // The second member is the version of the structure the configurations are read
796        // in, which is the one of the operation and not the one of the device, a v4 device
797        // answering with the v3 structure and a v6 one with the v5 structure.
798        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        // Unlike the commands, a single request carries every name, so an unknown one
813        // fails the whole request, which is the device raising and not something we could
814        // report for that one attribute only.
815        let request = match &names {
816            Some(names) => self.request(operation, names),
817            // The marker itself is what changed in v3 and not the operation taking it, so
818            // it follows the version of the structure like the operation does.
819            None if config_version >= DevVersion::V3 => self.request(operation, &[ALL_ATTRIBUTES_3]),
820            None => self.request(operation, &[ALL_ATTRIBUTES]),
821        };
822
823        // Kept so that the order can still be checked once the reply comes back.
824        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                // When the attributes have been named, the device answers with them in the
838                // order they were asked for, which the caller is allowed to rely on to tell
839                // one answer from the next, so it is checked here.
840                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
890/// TODO doc
891struct 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
905/// The failure of a CORBA call to a device, which is taken as the device not being
906/// reachable, or no longer: whatever it is, the connection to it is not one to go on with.
907/// The CORBA failure becomes the parent of a [`ErrorKind::DevNotConnected`], which is what
908/// tells a caller to find the device again, see [`crate::dev::DevProxy`]. What the device
909/// itself answered with, a `DevFailed` above all, is left as it is.
910fn 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/// TODO doc
918#[derive(Debug)]
919struct CorbaDevReply {
920    inner: ClientReply,
921}
922
923impl CorbaDevReply {
924
925    /// Read this reply, decoding the `DevFailed` every operation of the IDL may raise on
926    /// its own into an [`Err`] of its own: nothing that reads a reply afterward has to
927    /// deal with that one. Whichever other user exception turned up, if any, is left for
928    /// the caller to make sense of, see [`Self::read_simple`] for operations that raise
929    /// nothing else.
930    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    /// Same as [`Self::read`], for an operation that raises nothing but `DevFailed`:
943    /// whichever other user exception turned up is a failure of its own, nothing here
944    /// ever expecting one.
945    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    /// Same as [`Self::read_simple`], decoding the answer straight out of the reader.
955    fn read<'a, O: CdrRead<'a>>(&'a self) -> Result<O> {
956        self.read_simple()?.read()
957    }
958
959}
960
961/// Read a list of attribute readings, one entry per attribute, in the structure of that
962/// version.
963fn 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
972// =========================== //
973//       Server backend        //
974// =========================== //
975
976/// The IDL version the objects served by this backend speak, which is what a client is
977/// told they are and therefore which operations it sends them.
978///
979/// Six, the latest, whose only operation of its own, `info_6`, is answered: a client of an
980/// older release still reaches the objects, which announce every earlier interface as well
981/// and answer `_is_a` for each of them, see [`SERVED_REPO_IDS`].
982const SERVED_VERSION: DevVersion = DevVersion::V6;
983
984/// The Tango library release this backend tells its clients it is, in the encoding cppTango
985/// uses, `major * 100 + minor * 10 + patch`: the first cppTango release whose devices are
986/// of IDL v6, as the ones it serves are, see [`SERVED_VERSION`].
987const LIB_VERSION: i32 = 1000;
988
989/// The interfaces a served device announces, most derived first, which is every version of
990/// the Tango IDL up to the one this backend speaks, each of them deriving from the previous
991/// one, see [`Object::repo_ids`].
992const 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/// Internal structure implementing [`BackendServer`]
1003#[derive(Debug)]
1004struct CorbaServerBackend {
1005    /// The CORBA server itself, which is where the devices are served from and what the
1006    /// references handed out for them are built by. It serves from the moment it is bound
1007    /// and needs nothing run for it to, see [`Server::bind`].
1008    server: Server<CorbaServerObject>,
1009    /// What the events of the devices are published with, none when the server was not
1010    /// given the name of its admin device, which names them, see
1011    /// [`BackendServerConfig::admin_name`].
1012    supplier: Option<Supplier>,
1013}
1014
1015impl CorbaServerBackend {
1016
1017    /// The supplier of the events, failing if there is none, see [`Self::supplier`].
1018    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    /// The device of that name, which must be lowercased, if it is served here, see
1024    /// [`CorbaServerObject::name`].
1025    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    /// Whether a device of that name, which must be lowercased, is served here, see
1030    /// [`Self::served`].
1031    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        // Tango addresses a device by its own name, lowercased, which is both what the
1042        // reference the server gives back carries and what makes a plain
1043        // 'corbaloc:iiop:host:port/dev/name' reach it. Registering it is all there is to
1044        // it, the server answering from the moment it is registered.
1045        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/// One served device, seen as the CORBA object a client sends its requests to: this is the
1104/// mirror of [`CorbaDevBackend`], turning the operations of the Tango IDL into calls of the
1105/// device backend instead of the other way around.
1106#[derive(Debug)]
1107struct CorbaServerObject {
1108    /// The name of the device, lowercased, which is also its object key, see
1109    /// [`ServerObject::id`].
1110    name: Arc<str>,
1111    /// The device itself, whatever it is: this backend only knows it through the trait.
1112    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        // Which operation this is for rides in the request and is read back out of the
1129        // octets it arrived as, so nothing of it is copied here to outlive the call.
1130        Box::pin(async move { self.handle_request(&request).await })
1131    }
1132
1133}
1134
1135/// What an operation came to, turned into what goes back to the client.
1136///
1137/// Whatever it failed with is answered as the `DevFailed` every operation of the IDL
1138/// raises, whatever its kind: a device fails with any kind it likes, and what it says is
1139/// what a Tango client reads. Only a reply that could not be written at all is given back
1140/// to the backend, which turns it into the CORBA system exception that says what it is.
1141fn 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    /// Answer one request of this object, which is what the backend drives, see
1158    /// [`ServerObject::request`]. A failure is answered as a `DevFailed`, see [`answer`].
1159    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    /// Run one operation of the Tango IDL, reading its arguments out of the octets of the
1173    /// request and writing what it answers straight into the ones of the reply.
1174    ///
1175    /// Failing with [`ErrorKind::NotHandled`] is an operation this object does not have,
1176    /// which the client is told about like any other failure, see [`answer`].
1177    async fn dispatch<'d>(&self, operation: &[u8], request: &'d ServerRequest) -> Result<ServerReply> {
1178
1179        let mut input = request.read();
1180
1181        // The answer is written straight into the octets that go out, so the reply is
1182        // started before the operation runs and is what it writes into. One that raises
1183        // afterwards drops this and starts another, see [`Self::handle_request`].
1184        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            // IDL variants:
1196            // DevInfo info()
1197            // DevInfo_3 info_3()
1198            // DevInfo_6 info_6()
1199            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                // 'n' is a long and not a sequence length, a negative one meaning that the
1210                // client wants the whole of it.
1211                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                // IDL variants:
1221                // any command_inout(in string command, in any argin)
1222                // any command_inout_2(in string command, in any argin, in DevSource source)
1223                // any command_inout_4(in string command, in any argin, in DevSource source, in ClntIdent cl_ident)
1224                //
1225                // The identity of the client is read by none of them: this implementation has
1226                // nowhere to report it and nothing follows it in the request.
1227
1228                let name = input.read::<Scalar<str>>()?;
1229                let arg = read_command_data(&mut input)?;
1230
1231                // v1 has no source of its own, and the device is what a client that cannot
1232                // ask for anything else gets.
1233                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                // IDL variants:
1250                // AttributeValueList read_attributes(in DevVarStringArray names)
1251                // AttributeValueList read_attributes_2(in DevVarStringArray names, in DevSource source)
1252                // AttributeValueList_3 read_attributes_3(in DevVarStringArray names, in DevSource source)
1253                // AttributeValueList_4 read_attributes_4(in DevVarStringArray names, in DevSource source, in ClntIdent cl_ident)
1254                // AttributeValueList_5 read_attributes_5(in DevVarStringArray names, in DevSource source, in ClntIdent cl_ident)
1255
1256                // The version of the structure the values go out in, which is the one of
1257                // the operation and not the one of this object: a client that speaks an
1258                // older version is answered in the one it asked in.
1259                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                // IDL variants:
1287                // void write_attributes(in AttributeValueList values)
1288                // void write_attributes_3(in AttributeValueList values)
1289                // void write_attributes_4(in AttributeValueList_4 values, in ClntIdent cl_ident)
1290                //
1291                // Note that v2 has no variant of its own and reuses the v1 one, and that v3
1292                // only adds MultiDevFailed to the raised exceptions, the argument staying the
1293                // same.
1294
1295                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                // One outcome per value and in the order they were given, so which
1307                // attribute a failure is of is where it lies and not anything the backend
1308                // says: only the index is kept, and the names given are what the name of it
1309                // is looked up in below, see [`BackendDev::write_attr`].
1310                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                    // v1 has no exception for some of the attributes having failed, so the
1318                    // first failure is what the whole call failed with, and the client is
1319                    // not told which one of them it was.
1320                    if version < DevVersion::V4 {
1321                        return Err(failures.swap_remove(0).1);
1322                    }
1323
1324                    // Which starts the reply over as that exception, dropping whatever
1325                    // of a normal answer had been written into it.
1326                    *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                // IDL variants:
1343                // AttributeValueList_4 write_read_attributes_4(in AttributeValueList_4 values, in ClntIdent cl_ident)
1344                // AttributeValueList_5 write_read_attributes_5(in AttributeValueList_4 values, in DevVarStringArray r_names, in ClntIdent cl_ident)
1345                //
1346                // Both take the very same v4 list of values, only what they answer with
1347                // differs. The list of the attributes v5 lets a client read back other than
1348                // the written ones is read and ignored: the backend reads back what it wrote.
1349
1350                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                // A failure of one of the attributes is carried by its own entry of the
1359                // answer, which has a place for one, so there is no MultiDevFailed to
1360                // raise here unlike above.
1361                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                // IDL variants:
1373                // DevCmdInfo command_query(in string command)
1374                // DevCmdInfo_2 command_query_2(in string command)
1375
1376                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                // One command was asked about, so the first entry is the answer and
1386                // anything behind it is a backend that answered more than it was asked.
1387                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                // IDL variants:
1401                // DevCmdInfoList command_list_query()
1402                // DevCmdInfoList_2 command_list_query_2()
1403
1404                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                // The commands that cannot be described are left out, so how many go out
1413                // is only known once they have all been written, see [`SequenceLen`].
1414                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                    // One command that cannot be described is not worth failing the whole
1421                    // listing for, so it is simply left out of it.
1422                    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                // IDL variants:
1444                // void set_attribute_config(in AttributeConfigList new_conf)
1445                // void set_attribute_config_3(in AttributeConfigList_3 new_conf)
1446                // void set_attribute_config_4(in AttributeConfigList_3 new_conf, in ClntIdent cl_ident)
1447                // void set_attribute_config_5(in AttributeConfigList_5 new_conf, in ClntIdent cl_ident)
1448                //
1449                // Note that the configuration of v2 is never sent, v2 reusing the v1 operation,
1450                // and that the client identification is left unread. What an attribute *is*,
1451                // which the structure carries along with its configuration, is left out, the
1452                // configuration being all a client may change of it.
1453
1454                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                // Every operation raises a single DevFailed, so the first attribute that
1468                // failed is what the whole call fails with, the others having been changed.
1469                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                // IDL variants:
1481                // AttributeConfigList get_attribute_config(in DevVarStringArray names)
1482                // AttributeConfigList_2 get_attribute_config_2(in DevVarStringArray names)
1483                // AttributeConfigList_3 get_attribute_config_3(in DevVarStringArray names)
1484                // AttributeConfigList_5 get_attribute_config_5(in DevVarStringArray names)
1485                //
1486                // Note that v4 has no variant of its own and reuses the v3 one.
1487
1488                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                    // An attribute the client named and that cannot be described is what
1508                    // the whole request fails with, the client having no place to be told
1509                    // about that one only. One that it did not name, the listing being of
1510                    // every attribute, is simply left out.
1511                    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            // The operations the IDL declares and that this implementation has nothing to
1529            // answer with, either because the device backend has no such notion — the
1530            // polling history, the pipes and the static information of a device — or
1531            // because it is read-only, the configuration of an attribute being given by
1532            // its definition and not by a client.
1533            //
1534            // These are told apart from an operation that does not exist at all, so that the
1535            // DevFailed a client is answered with says that this server does not implement
1536            // them, and not that they should never have been sent.
1537            // TODO: The operations below are declared by the IDL, and the device backend
1538            // has what it takes to answer them, but the structures they travel in are not
1539            // marshalled yet: the run-length encoded history of v4 and v5 above all. See
1540            // `BackendDev::command_history` and `attr_history`.
1541            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
1566/// The names of a listing operation as the backend takes them, which is [`None`] for every
1567/// object of the device: 'all attributes' is a name of its own in the IDL and not the name
1568/// of an attribute, and the marker to use for it changed in v3, so both are recognized.
1569fn all_attributes(names: Spectrum<str>) -> Option<Spectrum<str>> {
1570    if has_all_marker(&names) {
1571        None
1572    } else {
1573        Some(names)
1574    }
1575}
1576
1577/// Whether the 'all attributes' marker is among these names, of either version, see
1578/// [`all_attributes`].
1579fn has_all_marker(names: &Spectrum<str>) -> bool {
1580    names.iter().any(|name| name == ALL_ATTRIBUTES || name == ALL_ATTRIBUTES_3)
1581}
1582
1583/// Refuse names given to a request that hold the 'all attributes' marker, which would ask
1584/// a device for every attribute where specific ones were meant.
1585fn 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
1594/// Read the list of values of an attribute writing operation, out of the octets it was
1595/// received in: only what the client wrote is kept of every entry, which is the only thing
1596/// those operations carry a value for.
1597fn 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        // A client that sends an attribute to write with no value in it is one whose
1607        // request makes no sense, there being nothing to write.
1608        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
1619/// The length of a sequence, kept for once its entries are written and how many there are
1620/// is known, which is how a listing that leaves some entries out goes out without being
1621/// filtered first, see [`CdrWriter::set_position`].
1622struct SequenceLen(usize);
1623
1624impl SequenceLen {
1625
1626    /// Leave the room for the length behind, where the sequence starts.
1627    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    /// Fill the length in, now that it is known, and carry on writing where the entries
1634    /// stopped.
1635    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/// Describe the Tango version of the device interface.
1646#[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    /// An array with all IDL versions in order.
1659    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    /// The IDL release number this version is known as in the Tango protocol, which every
1683    /// place the version travels as a number uses, one-based unlike the discriminant.
1684    #[inline]
1685    pub const fn number(&self) -> u32 {
1686        *self as u32 + 1
1687    }
1688
1689    /// The version of that IDL release number, see [`Self::number`], none for a release
1690    /// this crate does not know.
1691    #[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    /// A device whose server is not there is not connected, the CORBA failure being the
1717    /// parent.
1718    #[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}