spvirit_client/
client.rs

1use tokio::io::AsyncWriteExt;
2use tokio::net::TcpStream;
3use tokio::time::timeout;
4
5use crate::auth::{resolved_authnz_host, resolved_authnz_user};
6use crate::search::resolve_pv_server;
7use crate::transport::{read_packet, read_until};
8use crate::types::{PvGetError, PvGetOptions, PvGetResult};
9use spvirit_codec::epics_decode::{
10    PvaPacket, PvaPacketCommand, decode_op_response_status as codec_decode_op_response_status,
11};
12use spvirit_codec::spvd_encode::encode_pv_request;
13use spvirit_codec::spvirit_encode::encode_client_connection_validation;
14pub use spvirit_codec::spvirit_encode::{
15    encode_create_channel_request, encode_get_field_request, encode_get_request,
16    encode_monitor_request, encode_put_request,
17};
18
19pub fn build_client_validation(
20    opts: &crate::types::PvGetOptions,
21    version: u8,
22    is_be: bool,
23) -> Vec<u8> {
24    let user = resolved_authnz_user(opts);
25    let host = resolved_authnz_host(opts);
26    encode_client_connection_validation(87_040, 32_767, 0, "ca", &user, &host, version, is_be)
27}
28
29pub fn op_response_status(
30    raw: &[u8],
31    is_be: bool,
32) -> Result<Option<spvirit_codec::epics_decode::PvaStatus>, PvGetError> {
33    codec_decode_op_response_status(raw, is_be).map_err(PvGetError::Protocol)
34}
35
36pub fn ensure_status_ok(raw: &[u8], is_be: bool, step: &str) -> Result<(), PvGetError> {
37    match op_response_status(raw, is_be)? {
38        None => Ok(()),
39        Some(st) if st.code == 0 => Ok(()),
40        Some(st) => Err(PvGetError::Protocol(format!(
41            "{} failed: {}",
42            step,
43            st.message.unwrap_or_else(|| format!("code={}", st.code))
44        ))),
45    }
46}
47
48pub struct ChannelConn {
49    pub stream: TcpStream,
50    pub sid: u32,
51    pub version: u8,
52    pub is_be: bool,
53    pub server_addr: std::net::SocketAddr,
54}
55
56pub async fn establish_channel(
57    target: std::net::SocketAddr,
58    opts: &PvGetOptions,
59) -> Result<ChannelConn, PvGetError> {
60    let mut stream = timeout(opts.timeout, TcpStream::connect(target))
61        .await
62        .map_err(|_| PvGetError::Timeout("connect"))??;
63
64    let mut version = 2u8;
65    let mut is_be = false;
66
67    for _ in 0..2 {
68        if let Ok(bytes) = read_packet(&mut stream, opts.timeout).await {
69            let mut pkt = PvaPacket::new(&bytes);
70            if let Some(cmd) = pkt.decode_payload() {
71                match cmd {
72                    PvaPacketCommand::Control(payload) => {
73                        if payload.command == 2 {
74                            is_be = pkt.header.flags.is_msb;
75                        }
76                    }
77                    PvaPacketCommand::ConnectionValidation(_) => {
78                        version = pkt.header.version;
79                        is_be = pkt.header.flags.is_msb;
80                    }
81                    _ => {}
82                }
83            }
84        }
85    }
86
87    let validation = build_client_validation(opts, version, is_be);
88    stream.write_all(&validation).await?;
89
90    let _ = read_until(&mut stream, opts.timeout, |cmd| {
91        matches!(cmd, PvaPacketCommand::ConnectionValidated(_))
92    })
93    .await?;
94
95    let cid = 1u32;
96    let create = encode_create_channel_request(cid, &opts.pv_name, version, is_be);
97    stream.write_all(&create).await?;
98
99    let create_resp = read_until(&mut stream, opts.timeout, |cmd| {
100        matches!(cmd, PvaPacketCommand::CreateChannel(_))
101    })
102    .await?;
103    let mut pkt = PvaPacket::new(&create_resp);
104    let cmd = pkt.decode_payload().ok_or(PvGetError::Protocol(
105        "create_channel decode failed".to_string(),
106    ))?;
107    let sid = match cmd {
108        PvaPacketCommand::CreateChannel(payload) => {
109            if payload.status.as_ref().is_some_and(|s| s.is_error()) {
110                let detail = payload
111                    .status
112                    .as_ref()
113                    .map(ToString::to_string)
114                    .unwrap_or_default();
115                return Err(PvGetError::Protocol(format!(
116                    "create_channel error: {}",
117                    detail
118                )));
119            }
120            payload.sid
121        }
122        _ => {
123            return Err(PvGetError::Protocol(
124                "unexpected create_channel response".to_string(),
125            ));
126        }
127    };
128
129    Ok(ChannelConn {
130        stream,
131        sid,
132        version,
133        is_be,
134        server_addr: target,
135    })
136}
137
138/// Convenience wrapper: GET with no field filtering.
139pub async fn pvget(opts: &PvGetOptions) -> Result<PvGetResult, PvGetError> {
140    pvget_fields(opts, &[]).await
141}
142
143/// GET with optional field filtering.
144///
145/// If `fields` is empty, requests all fields (equivalent to `-r ""`).
146/// Otherwise, encodes a pvRequest like `field(value,alarm,timeStamp)`.
147pub async fn pvget_fields(opts: &PvGetOptions, fields: &[&str]) -> Result<PvGetResult, PvGetError> {
148    let target = resolve_pv_server(opts).await?;
149
150    let conn = establish_channel(target, opts).await?;
151    let ChannelConn {
152        mut stream,
153        sid,
154        version,
155        is_be,
156        ..
157    } = conn;
158
159    let ioid = 1u32;
160    let pv_request = if fields.is_empty() {
161        // Empty pvRequest — request all fields
162        vec![0xfd, 0x02, 0x00, 0x80, 0x00, 0x00]
163    } else {
164        encode_pv_request(fields, is_be)
165    };
166    let get_init_req = encode_get_request(sid, ioid, 0x08, &pv_request, version, is_be);
167    stream.write_all(&get_init_req).await?;
168
169    let init_resp = read_until(
170        &mut stream,
171        opts.timeout,
172        |cmd| matches!(cmd, PvaPacketCommand::Op(op) if (op.subcmd & 0x08) != 0),
173    )
174    .await?;
175    let mut pkt = PvaPacket::new(&init_resp);
176    let cmd = pkt.decode_payload().ok_or(PvGetError::Protocol(
177        "get init response decode failed".to_string(),
178    ))?;
179
180    let desc = match cmd {
181        PvaPacketCommand::Op(op) => op
182            .introspection
183            .ok_or_else(|| PvGetError::Decode("missing introspection".to_string()))?,
184        _ => {
185            return Err(PvGetError::Protocol(
186                "unexpected get init response".to_string(),
187            ));
188        }
189    };
190
191    let get_data_req = encode_get_request(sid, ioid, 0x00, &[], version, is_be);
192    stream.write_all(&get_data_req).await?;
193
194    let data_resp = read_until(
195        &mut stream,
196        opts.timeout,
197        |cmd| matches!(cmd, PvaPacketCommand::Op(op) if op.subcmd == 0x00),
198    )
199    .await?;
200    let mut pkt = PvaPacket::new(&data_resp);
201    let cmd = pkt.decode_payload().ok_or(PvGetError::Protocol(
202        "get data response decode failed".to_string(),
203    ))?;
204
205    match cmd {
206        PvaPacketCommand::Op(mut op) => {
207            op.decode_with_field_desc(&desc, is_be);
208            if let Some(value) = op.decoded_value {
209                return Ok(PvGetResult {
210                    pv_name: opts.pv_name.clone(),
211                    value,
212                    raw_pva: data_resp,
213                    raw_pvd: op.body,
214                    introspection: desc,
215                });
216            }
217            Err(PvGetError::Decode("no decoded value".to_string()))
218        }
219        _ => Err(PvGetError::Protocol(
220            "unexpected get data response".to_string(),
221        )),
222    }
223}
224
225#[cfg(test)]
226mod tests {
227    use super::*;
228    use spvirit_codec::epics_decode::{PvaPacket, PvaPacketCommand, PvaStatus};
229
230    #[test]
231    fn encode_decode_monitor_request_roundtrip() {
232        let msg =
233            encode_monitor_request(1, 2, 0x08, &[0xfd, 0x02, 0x00, 0x80, 0x00, 0x00], 2, false);
234        let mut pkt = PvaPacket::new(&msg);
235        let cmd = pkt.decode_payload().expect("decoded");
236        match cmd {
237            PvaPacketCommand::Op(op) => {
238                assert_eq!(op.command, 13);
239                assert_eq!(op.subcmd, 0x08);
240                assert_eq!(op.sid_or_cid, 1);
241                assert_eq!(op.ioid, 2);
242            }
243            other => panic!("unexpected decode: {:?}", other),
244        }
245    }
246
247    #[test]
248    fn encode_decode_put_init_roundtrip() {
249        let msg = encode_put_request(5, 6, 0x08, &[0xfd, 0x02, 0x00, 0x80, 0x00, 0x00], 2, false);
250        let mut pkt = PvaPacket::new(&msg);
251        let cmd = pkt.decode_payload().expect("decoded");
252        match cmd {
253            PvaPacketCommand::Op(op) => {
254                assert_eq!(op.command, 11);
255                assert_eq!(op.subcmd, 0x08);
256                assert_eq!(op.sid_or_cid, 5);
257                assert_eq!(op.ioid, 6);
258            }
259            other => panic!("unexpected decode: {:?}", other),
260        }
261    }
262
263    #[test]
264    fn encode_decode_get_field_request_roundtrip() {
265        let msg = encode_get_field_request(7, 1, Some("*"), 2, false);
266        let mut pkt = PvaPacket::new(&msg);
267        let cmd = pkt.decode_payload().expect("decoded");
268        match cmd {
269            PvaPacketCommand::GetField(payload) => {
270                assert!(!payload.is_server);
271                assert_eq!(payload.sid, Some(7));
272                assert_eq!(payload.ioid, Some(1));
273                assert_eq!(payload.field_name.as_deref(), Some("*"));
274            }
275            other => panic!("unexpected decode: {:?}", other),
276        }
277    }
278
279    #[test]
280    fn encode_decode_get_field_request_empty_field_roundtrip() {
281        let msg = encode_get_field_request(7, 1, None, 2, false);
282        let mut pkt = PvaPacket::new(&msg);
283        let cmd = pkt.decode_payload().expect("decoded");
284        match cmd {
285            PvaPacketCommand::GetField(payload) => {
286                assert!(!payload.is_server);
287                assert_eq!(payload.sid, Some(7));
288                assert_eq!(payload.ioid, Some(1));
289                assert_eq!(payload.field_name.as_deref(), Some(""));
290            }
291            other => panic!("unexpected decode: {:?}", other),
292        }
293    }
294
295    #[test]
296    fn pva_status_code_zero_is_not_an_error() {
297        let ok = PvaStatus {
298            code: 0,
299            message: None,
300            stack: None,
301        };
302        let err = PvaStatus {
303            code: 1,
304            message: Some("bad".to_string()),
305            stack: None,
306        };
307        assert!(!None::<&PvaStatus>.is_some_and(|s| s.is_error()));
308        assert!(!Some(&ok).is_some_and(|s| s.is_error()));
309        assert!(Some(&err).is_some_and(|s| s.is_error()));
310    }
311}
spvirit_client/client.rs

spvirit_client/
client.rs
