iec104 0.4.0

A rust implementation of the IEC-60870-5-104 protocol.
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

//! Example client for IEC 60870-5-104

use std::time::Duration;

use async_trait::async_trait;
use iec104::{
	asdu::Asdu,
	client::{Client, ClientCallback, errors::ClientError},
	config::ClientConfig,
	types::{
		commands::Rcs,
		information_elements::{Dpi, Spi},
	},
};
use snafu::{ResultExt as _, Whatever, whatever};
use tokio::{
	signal::unix::{SignalKind, signal},
	time::Instant,
};
use tracing_error::ErrorLayer;
use tracing_subscriber::{
	Layer as _, filter::EnvFilter, layer::SubscriberExt as _, util::SubscriberInitExt as _,
};

#[tokio::main]
async fn main() -> Result<(), Whatever> {
	let filter = EnvFilter::from("info");
	let layer = tracing_subscriber::fmt::layer().with_filter(filter);
	tracing_subscriber::registry()
		.with(layer)
		//needed to get the tracing_error working
		.with(ErrorLayer::default().with_filter(EnvFilter::from("debug")))
		.init();

	let mut client = Client::new(ClientConfig::default(), MyCallback);
	client.connect().await.whatever_context("Failed to connect")?;
	client.start_receiving().await.whatever_context("Failed to start receiving")?;

	let mut s1 = signal(SignalKind::interrupt()).whatever_context("Failed to create signal")?;
	let mut s2 = signal(SignalKind::terminate()).whatever_context("Failed to create signal")?;

	let period = tokio::time::sleep(Duration::from_secs(1));
	tokio::pin!(period);

	let stop = tokio::time::sleep(Duration::from_secs(5));
	tokio::pin!(stop);

	let restart = tokio::time::sleep(Duration::from_secs(15));
	tokio::pin!(restart);

	loop {
		tokio::select! {
			_ = s1.recv() => {tracing::info!("SIGINT"); break;},
			_ = s2.recv() => {tracing::info!("SIGTERM"); break;},
			_ = &mut period => {
				tracing::info!("Period");
				check_error(client.send_command_rc(47, 13, Rcs::Increment, None, None, None).await)?;
				check_error(client.send_command_sp(47, 14, Spi::On, None, None, None).await)?;
				check_error(client.send_command_dp(47, 15, Dpi::On, None, None, None).await)?;
				check_error(client.send_command_bs(47, 16, 1, None).await)?;


				period.as_mut().reset(Instant::now() + Duration::from_secs(1));
			},
			_ = &mut stop => {
				tracing::info!("Stopping");
				client.stop_receiving().await.whatever_context("Failed to stop receiving")?;
				stop.as_mut().reset(Instant::now() + Duration::from_secs(3600));
			}
			_ = &mut restart => {
				tracing::info!("Restarting");
				client.start_receiving().await.whatever_context("Failed to start receiving")?;
				restart.as_mut().reset(Instant::now() + Duration::from_secs(3615));
			}
		}
	}

	tracing::info!("Disconnecting");

	Ok(())
}

/// Callback for the client
struct MyCallback;

#[async_trait]
impl ClientCallback for MyCallback {
	// Required callbacks
	async fn on_new_objects(&self, asdu: Asdu) {
		tracing::trace!("Received objects: {asdu:?}");
	}
	// Optional callbacks
	async fn on_connection_started(&self) {
		tracing::debug!("Connection started");
	}
	async fn on_connection_stopped(&self) {
		tracing::debug!("Connection stopped");
	}
	async fn on_error(&self, error: iec104::error::Error) {
		tracing::debug!("Error: {error}");
	}
	async fn on_reconnecting(&self) {
		tracing::debug!("Reconnecting");
	}
}

/// Check the error to see if it is a critical error
fn check_error(r: Result<(), ClientError>) -> Result<(), Whatever> {
	if let Err(e) = r {
		match e {
			ClientError::NoWriteChannel { .. } => {
				whatever!("There is no channel to send commands");
			}
			_ => {
				tracing::error!("Error sending ASDU: {e}");
				Ok(())
			}
		}
	} else {
		Ok(())
	}
}