rstango 0.3.0

Implementation of the Tango distributed control system
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139

//! The administration device of a server, see the admin commands of cppTango.

mod common;

use std::sync::{Arc, Mutex};

use futures::executor::block_on;

use tango::dev::DevSource;
use tango::value::{Scalar, Value};
use tango::server::ClassDef;
use tango::error::Result;
use tango::testing::TestContext;

use common::{local_and_wire, test_context, test_server, exported, unexported};


/// A command sent to the admin device of the test server, as a client sends it.
fn admin_command(ctx: &TestContext, name: &str, input: Value) -> Result<Value> {
    let admin = ctx.backend_dev("dserver/Test/test").unwrap();
    block_on(admin.command(Scalar::from(name), input, DevSource::Dev))
}

fn strings(value: Value) -> Vec<String> {
    Vec::<String>::try_from(value).unwrap()
}

/// A class recording the devices it deletes.
fn class(name: &'static str, deleted: &Arc<Mutex<Vec<String>>>) -> ClassDef<()> {
    let deleted = Arc::clone(deleted);
    ClassDef::new_default(name)
        .delete(move |(), dev| deleted.lock().unwrap().push(dev.name().to_string()))
}


local_and_wire! {
    /// The classes and the devices of a server, the admin ones aside, class by class.
    fn queries(local: bool) {

        let deleted = Arc::new(Mutex::new(Vec::new()));
        let ctx = test_context(&[("b/dev/2", "B"), ("a/dev/1", "A"), ("b/dev/1", "B")]);
        let _server = ctx.serve(test_server(local)
            .class(class("A", &deleted)).unwrap()
            .class(class("B", &deleted)).unwrap())
            .unwrap();

        assert_eq!(strings(admin_command(&ctx, "QueryClass", Value::Void).unwrap()), ["A", "B"]);
        assert_eq!(strings(admin_command(&ctx, "QueryDevice", Value::Void).unwrap()),
            ["A::a/dev/1", "B::b/dev/1", "B::b/dev/2"]);

    }
}

local_and_wire! {
    /// A device restarted by name, as a client knows it: deleted, then initialized and
    /// exported again.
    fn dev_restart(local: bool) {

        let deleted = Arc::new(Mutex::new(Vec::new()));
        let ctx = test_context(&[("a/dev/1", "A")]);
        let _server = ctx.serve(test_server(local)
            .class(class("A", &deleted)).unwrap())
            .unwrap();

        admin_command(&ctx, "DevRestart", Value::from("tango://db:10000/A/dev/1#dbase=yes")).unwrap();
        assert_eq!(*deleted.lock().unwrap(), ["a/dev/1"]);
        assert_eq!(unexported(ctx.test_database()), ["a/dev/1"]);
        assert_eq!(exported(ctx.test_database()), ["dserver/Test/test", "a/dev/1", "a/dev/1"]);

        assert!(admin_command(&ctx, "DevRestart", Value::from("a/dev/2")).is_err());
        assert!(admin_command(&ctx, "DevRestart", Value::from("dserver/Test/test")).is_err());

    }
}

local_and_wire! {
    /// A server restarted: every device but the admin one removed and added again from
    /// what the database lists now, before the request is answered.
    fn restart_server(local: bool) {

        let deleted = Arc::new(Mutex::new(Vec::new()));
        let ctx = test_context(&[("a/dev/1", "A"), ("a/dev/2", "A")]);
        let server = ctx.serve(test_server(local)
            .class(class("A", &deleted)).unwrap())
            .unwrap();

        let database = ctx.database();
        database.delete_dev("a/dev/1").unwrap();
        database.add_dev("a/dev/3", "A", "Test/test").unwrap();
        admin_command(&ctx, "RestartServer", Value::Void).unwrap();
        assert_eq!(exported(ctx.test_database()).len(), 5);

        let mut deleted = deleted.lock().unwrap().clone();
        deleted.sort();
        assert_eq!(deleted, ["a/dev/1", "a/dev/2"]);
        assert_eq!(server.devs().iter().map(|dev| dev.name().to_string()).collect::<Vec<_>>(),
            ["dserver/Test/test", "a/dev/2", "a/dev/3"]);

    }
}

local_and_wire! {
    /// The negotiation of an event through the admin device, which the backend answers with
    /// where it publishes it, and under which topic. A subscription is refused, the devices
    /// of a server publishing no event yet.
    #[cfg_attr(miri, ignore = "Miri cannot run ZMQ, a C library")]
    fn zmq_event_subscription_change(local: bool) {

        let deleted = Arc::new(Mutex::new(Vec::new()));
        let ctx = test_context(&[("a/dev/1", "A")]);
        let _server = ctx.serve(test_server(local)
            .class(class("A", &deleted)).unwrap())
            .unwrap();

        let input = Value::from(["A/dev/1", "Attr", "unsubscribe", "change", "6"]);
        let Value::LongString(numbers, strings) = admin_command(&ctx, "ZmqEventSubscriptionChange", input).unwrap() else {
            panic!("Not a DevVarLongStringArray");
        };

        let numbers = numbers.iter().collect::<Vec<_>>();
        let strings = strings.iter().collect::<Vec<_>>();
        assert_eq!(numbers[1], 6);
        assert!(strings[0].starts_with("tcp://"));
        assert!(strings[1].starts_with("tcp://"));
        assert!(strings[strings.len() - 2].ends_with("/a/dev/1/attr.idl5_change"));
        assert!(strings[strings.len() - 1].ends_with("/dserver/test/test"));

        // Nothing is published, the subscription below being refused.
        let input = Value::from(["a/dev/1", "attr", "idl5_change"]);
        assert!(admin_command(&ctx, "EventConfirmSubscription", input).is_err());

        let input = Value::from(["a/dev/1", "attr", "subscribe", "change", "6"]);
        assert!(admin_command(&ctx, "ZmqEventSubscriptionChange", input).is_err());

        let input = Value::from(["a/dev/2", "attr", "unsubscribe", "change", "6"]);
        assert!(admin_command(&ctx, "ZmqEventSubscriptionChange", input).is_err());

    }
}