dimas 0.5.1

dimas - a framework for Distributed Multi Agent Systems
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

//! `DiMAS` observation example
//! Copyright © 2024 Stephan Kunz

use dimas::prelude::*;

#[derive(Debug)]
struct AgentProps {
	limit: u128,
	new_limit: u128,
	occupied_counter: u8,
}

/// request structure for observer and observable
#[derive(Debug, Encode, Decode)]
pub struct FibonacciRequest {
	/// limit
	pub limit: u128,
}

async fn control_response(ctx: Context<AgentProps>, response: ControlResponse) -> Result<()> {
	match response {
		ControlResponse::Accepted => {
			let limit = ctx.read()?.new_limit;
			println!("Accepted fibonacci up to {limit}");
			ctx.write()?.limit = limit;
			ctx.write()?.new_limit += 1;
		}
		ControlResponse::Declined => {
			println!("Declined fibonacci up to {}", ctx.read()?.new_limit);
			ctx.write()?.limit = 0;
			ctx.write()?.new_limit = 5;
		}
		ControlResponse::Occupied => {
			println!("Service fibonacci is occupied");
			let occupied_counter = ctx.read()?.occupied_counter + 1;
			// cancel running request whenever 5 occupied messages arrived
			if occupied_counter % 5 == 0 {
				ctx.cancel_observe("fibonacci")?;
				ctx.write()?.occupied_counter = 0;
			} else {
				ctx.write()?.occupied_counter = occupied_counter;
			}
		}
		ControlResponse::Canceled => {
			println!("Canceled fibonacci up to {}", ctx.read()?.limit);
		}
	}
	Ok(())
}

async fn response(ctx: Context<AgentProps>, response: ObservableResponse) -> Result<()> {
	match response {
		ObservableResponse::Canceled(value) => {
			let msg = Message::new(value);
			let result: Vec<u128> = msg.decode()?;

			println!("Canceled at {result:?}");
		}
		ObservableResponse::Feedback(value) => {
			let msg = Message::new(value);
			let result: Vec<u128> = msg.decode()?;
			let limit = ctx.read()?.limit;
			if result.len() <= limit as usize {
				println!("Received feedback {result:?}");
			} else {
				println!("Wrong feedback {result:?}");
			}
		}
		ObservableResponse::Finished(value) => {
			let msg = Message::new(value);
			let result: Vec<u128> = msg.decode()?;
			let limit = ctx.read()?.limit;
			if result.len() == limit as usize {
				println!("Received result {result:?}");
			} else {
				println!("Wrong result {result:?}");
			}
		}
	}
	Ok(())
}

#[dimas::main]
async fn main() -> Result<()> {
	// create & initialize agents properties
	let properties = AgentProps {
		limit: 0u128,
		new_limit: 5u128,
		occupied_counter: 0u8,
	};

	// create an agent with the properties and the prefix 'examples'
	let mut agent = Agent::new(properties)
		.prefix("examples")
		.name("observer")
		.config(&Config::default())?;

	// create the observer for fibonacci
	agent
		.observer()
		.topic("fibonacci")
		.control_callback(control_response)
		.result_callback(response)
		.add()?;

	// timer for next observation
	let interval = Duration::from_secs(5);
	agent
		.timer()
		.name("timer")
		.interval(interval)
		.callback(move |ctx| -> Result<()> {
			let limit = ctx.read()?.new_limit;
			println!("request fibonacci up to {limit}");
			let msg = FibonacciRequest { limit };
			let message = Message::encode(&msg);
			ctx.observe("fibonacci", Some(message))?;
			Ok(())
		})
		.add()?;

	// activate liveliness
	agent.liveliness(true);
	// run agent
	agent.start().await?;

	Ok(())
}