|
|
|
|
@ -1,5 +1,3 @@
|
|
|
|
|
extern crate dotenv;
|
|
|
|
|
|
|
|
|
|
use rumqttc::{MqttOptions, Client, QoS, LastWill};
|
|
|
|
|
use bytes::Bytes;
|
|
|
|
|
use std::time::Duration;
|
|
|
|
|
@ -18,9 +16,6 @@ use sha2::{Sha256, Digest};
|
|
|
|
|
use std::net::TcpStream;
|
|
|
|
|
use std::sync::mpsc;
|
|
|
|
|
|
|
|
|
|
use dotenv::dotenv;
|
|
|
|
|
use std::env;
|
|
|
|
|
|
|
|
|
|
// MQTT Topics:
|
|
|
|
|
// - division/{division_id}/{round}/{match}/score
|
|
|
|
|
// - division/{division_id}/ranking
|
|
|
|
|
@ -353,7 +348,7 @@ struct BackendMessage {
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
impl BackendMessage {
|
|
|
|
|
fn from_bytes(bytes: &Vec<u8>) -> Option<BackendMessage> {
|
|
|
|
|
fn from_bytes(bytes: Vec<u8>) -> Option<BackendMessage> {
|
|
|
|
|
if bytes.len() < 5 {
|
|
|
|
|
return None;
|
|
|
|
|
}
|
|
|
|
|
@ -413,7 +408,7 @@ impl BackendPacket {
|
|
|
|
|
};
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
fn from_bytes(bytes: &Vec<u8>) -> Option<BackendPacket> {
|
|
|
|
|
fn from_bytes(bytes: Vec<u8>) -> Option<BackendPacket> {
|
|
|
|
|
if bytes.len() < BACKEND_PACKET_HEADER_SIZE {
|
|
|
|
|
return None;
|
|
|
|
|
}
|
|
|
|
|
@ -476,7 +471,7 @@ impl ConnectMsg {
|
|
|
|
|
};
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
fn from_bytes(bytes: &Vec<u8>) -> Option<ConnectMsg> {
|
|
|
|
|
fn from_bytes(bytes: Vec<u8>) -> Option<ConnectMsg> {
|
|
|
|
|
if bytes.len() < CONNECT_MSG_LEN {
|
|
|
|
|
return None;
|
|
|
|
|
}
|
|
|
|
|
@ -518,14 +513,14 @@ struct NoticeMsg {
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
impl NoticeMsg {
|
|
|
|
|
fn from_bytes(bytes: &Vec<u8>) -> Option<NoticeMsg> {
|
|
|
|
|
fn from_bytes(bytes: Vec<u8>) -> Option<NoticeMsg> {
|
|
|
|
|
if bytes.len() < 8 {
|
|
|
|
|
return None;
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
let notice_id = u64::from_le_bytes(bytes[0..8].try_into().unwrap());
|
|
|
|
|
|
|
|
|
|
match BackendMessage::from_bytes(&(bytes[8..].to_vec())) {
|
|
|
|
|
match BackendMessage::from_bytes(bytes[8..].to_vec()) {
|
|
|
|
|
Some(message) => {
|
|
|
|
|
match message.data.notice {
|
|
|
|
|
Some(notice) => Some(NoticeMsg{
|
|
|
|
|
@ -552,11 +547,11 @@ struct TMClient {
|
|
|
|
|
username: [u8; 16],
|
|
|
|
|
connected: bool,
|
|
|
|
|
time_offset: f64,
|
|
|
|
|
offset_tick: u16,
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
const TCP_BUFFER_SIZE: usize = 10000;
|
|
|
|
|
impl TMClient {
|
|
|
|
|
fn new(uuid: [u8; 16], client_name: [u8; 32], host: String, password: String, username: [u8; 16]) -> (TMClient, TMConnection) {
|
|
|
|
|
fn new(uuid: [u8; 16], client_name: [u8; 32], password: String, username: [u8; 16]) -> (TMClient, TMConnection) {
|
|
|
|
|
let (work_tx, work_rx) = mpsc::channel();
|
|
|
|
|
let (response_tx, response_rx) = mpsc::channel();
|
|
|
|
|
let (request_tx, request_rx) = mpsc::channel();
|
|
|
|
|
@ -567,8 +562,7 @@ impl TMClient {
|
|
|
|
|
|
|
|
|
|
let connector = builder.build();
|
|
|
|
|
|
|
|
|
|
log::debug!("Connecting to TM using address {host}:5000");
|
|
|
|
|
let stream = TcpStream::connect(format!("{host}:5000")).unwrap();
|
|
|
|
|
let stream = TcpStream::connect("127.0.0.1:5000").unwrap();
|
|
|
|
|
|
|
|
|
|
let mut stream_config = connector.configure().unwrap();
|
|
|
|
|
stream_config.set_verify_hostname(false);
|
|
|
|
|
@ -576,8 +570,9 @@ impl TMClient {
|
|
|
|
|
stream_config.set_private_key_file("tm.crt", openssl::ssl::SslFiletype::PEM).unwrap();
|
|
|
|
|
stream_config.set_use_server_name_indication(false);
|
|
|
|
|
|
|
|
|
|
let stream = stream_config.connect(&host, stream).unwrap();
|
|
|
|
|
stream.get_ref().set_read_timeout(Some(Duration::from_millis(10))).expect("Failed to set read timeout on socket");
|
|
|
|
|
let stream = stream_config.connect("127.0.0.1", stream).unwrap();
|
|
|
|
|
stream.get_ref().set_read_timeout(Some(Duration::from_millis(100))).expect("Failed to set read timeout on socket");
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
return (TMClient{
|
|
|
|
|
stream,
|
|
|
|
|
@ -591,7 +586,6 @@ impl TMClient {
|
|
|
|
|
username,
|
|
|
|
|
connected: false,
|
|
|
|
|
time_offset: 0.0,
|
|
|
|
|
offset_tick: 0,
|
|
|
|
|
},
|
|
|
|
|
TMConnection{
|
|
|
|
|
time_offset: 0.0,
|
|
|
|
|
@ -617,126 +611,85 @@ impl TMClient {
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
let mut header = [0; BACKEND_PACKET_HEADER_SIZE];
|
|
|
|
|
let mut read_bytes = 0;
|
|
|
|
|
let data_size: usize;
|
|
|
|
|
match self.stream.read_exact(&mut header) {
|
|
|
|
|
Ok(()) => {
|
|
|
|
|
match BackendPacket::from_bytes(&header.to_vec()) {
|
|
|
|
|
let mut incoming = [0; TCP_BUFFER_SIZE];
|
|
|
|
|
match self.stream.read(&mut incoming) {
|
|
|
|
|
Ok(read) => {
|
|
|
|
|
let data = incoming[0..read].to_vec();
|
|
|
|
|
match BackendPacket::from_bytes(data) {
|
|
|
|
|
Some(packet) => {
|
|
|
|
|
data_size = packet.size as usize;
|
|
|
|
|
read_bytes += BACKEND_PACKET_HEADER_SIZE;
|
|
|
|
|
log::debug!("Received {} bytes ({}/{})", read_bytes, read_bytes, BACKEND_PACKET_HEADER_SIZE + data_size);
|
|
|
|
|
},
|
|
|
|
|
None => {
|
|
|
|
|
log::error!("Failed to parse BackendPacket header: {:?}", header);
|
|
|
|
|
return;
|
|
|
|
|
},
|
|
|
|
|
}
|
|
|
|
|
},
|
|
|
|
|
Err(ref error) if error.kind() == std::io::ErrorKind::WouldBlock => {
|
|
|
|
|
if self.offset_tick == 0 {
|
|
|
|
|
match self.work_queue.send(Work::Offset(self.time_offset)) {
|
|
|
|
|
Ok(_) => log::debug!("Sent time offset"),
|
|
|
|
|
Err(error) => log::error!("Offset send error: {:#?}", error),
|
|
|
|
|
}
|
|
|
|
|
self.offset_tick = 100;
|
|
|
|
|
} else {
|
|
|
|
|
self.offset_tick -= 1;
|
|
|
|
|
}
|
|
|
|
|
log::debug!("Resource temporarily unavailable, retrying later.");
|
|
|
|
|
return;
|
|
|
|
|
},
|
|
|
|
|
Err(error) => {
|
|
|
|
|
log::error!("Error reading header: {:?}", error);
|
|
|
|
|
return;
|
|
|
|
|
},
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
let mut data = vec![0; BACKEND_PACKET_HEADER_SIZE + data_size];
|
|
|
|
|
data[..header.len()].copy_from_slice(&header);
|
|
|
|
|
|
|
|
|
|
while read_bytes < BACKEND_PACKET_HEADER_SIZE + data_size {
|
|
|
|
|
match self.stream.read(&mut data[read_bytes..]) {
|
|
|
|
|
Ok(read) => {
|
|
|
|
|
read_bytes += read;
|
|
|
|
|
log::debug!("Received {} bytes ({}/{})", read, read_bytes, BACKEND_PACKET_HEADER_SIZE + data_size);
|
|
|
|
|
},
|
|
|
|
|
Err(error) => {
|
|
|
|
|
log::error!("Error reading body data: {:?}", error);
|
|
|
|
|
},
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
match BackendPacket::from_bytes(&data) {
|
|
|
|
|
Some(packet) => {
|
|
|
|
|
let offset = packet.timestamp - get_float_time();
|
|
|
|
|
self.time_offset = offset.clone();
|
|
|
|
|
self.last_seq_num = packet.seq_num;
|
|
|
|
|
match packet.msg_type {
|
|
|
|
|
// Notice Message
|
|
|
|
|
4 => {
|
|
|
|
|
match NoticeMsg::from_bytes(&packet.data) {
|
|
|
|
|
Some(notice) => {
|
|
|
|
|
log::debug!("Received notice: {:#?}", notice);
|
|
|
|
|
let ack = BackendPacket::new(packet.header, packet.timestamp, 5, self.last_seq_num+1, notice.notice_id.to_le_bytes().to_vec());
|
|
|
|
|
self.last_seq_num += 1;
|
|
|
|
|
match self.stream.write(&ack.as_bytes()) {
|
|
|
|
|
Ok(_) => log::debug!("Sent ACK for notice {}", notice.notice_id),
|
|
|
|
|
Err(error) => log::error!("ACK error: {:?}", error),
|
|
|
|
|
}
|
|
|
|
|
match self.work_queue.send(Work::Notice(Box::new(NoticeWithOffset{notice: notice.notice, offset: self.time_offset}))) {
|
|
|
|
|
Ok(_) => log::debug!("Forwarded notice to callback engine"),
|
|
|
|
|
Err(error) => log::error!("Notice forward error {:?}", error),
|
|
|
|
|
self.last_seq_num = packet.seq_num;
|
|
|
|
|
match packet.msg_type {
|
|
|
|
|
// Notice Message
|
|
|
|
|
4 => {
|
|
|
|
|
match NoticeMsg::from_bytes(packet.data.clone()) {
|
|
|
|
|
Some(notice) => {
|
|
|
|
|
log::debug!("Received notice: {:#?}", notice);
|
|
|
|
|
let ack = BackendPacket::new(packet.header, packet.timestamp, 5, self.last_seq_num+1, notice.notice_id.to_le_bytes().to_vec());
|
|
|
|
|
self.last_seq_num += 1;
|
|
|
|
|
match self.stream.write(&ack.as_bytes()) {
|
|
|
|
|
Ok(_) => log::debug!("Sent ACK for notice {}", notice.notice_id),
|
|
|
|
|
Err(error) => log::error!("ACK error: {:?}", error),
|
|
|
|
|
}
|
|
|
|
|
match self.work_queue.send(Work::Notice(Box::new(notice.notice))) {
|
|
|
|
|
Ok(_) => log::debug!("Forwarded notice to callback engine"),
|
|
|
|
|
Err(error) => log::error!("Notice forward error {:?}", error),
|
|
|
|
|
}
|
|
|
|
|
},
|
|
|
|
|
None => log::error!("Notice parse error: {:?}", packet),
|
|
|
|
|
}
|
|
|
|
|
},
|
|
|
|
|
None => log::error!("Notice parse error: {:?}", packet),
|
|
|
|
|
}
|
|
|
|
|
},
|
|
|
|
|
// Response message
|
|
|
|
|
3 => {
|
|
|
|
|
match BackendMessage::from_bytes(&packet.data) {
|
|
|
|
|
Some(message) => {
|
|
|
|
|
log::debug!("Received response: {:#?}", message);
|
|
|
|
|
|
|
|
|
|
match self.responses.send((Box::new(message), offset)) {
|
|
|
|
|
Ok(_) => log::debug!("Forwarded response to callback engine"),
|
|
|
|
|
Err(error) => log::error!("Response forward error {:?}", error),
|
|
|
|
|
// Response message
|
|
|
|
|
3 => {
|
|
|
|
|
match BackendMessage::from_bytes(packet.data.clone()) {
|
|
|
|
|
Some(message) => {
|
|
|
|
|
log::debug!("Received response: {:#?}", message);
|
|
|
|
|
let offset = packet.timestamp - get_float_time();
|
|
|
|
|
self.time_offset = offset.clone();
|
|
|
|
|
log::info!("New offset: {}", offset);
|
|
|
|
|
log::info!("Server Timetamp: {}", packet.timestamp);
|
|
|
|
|
log::info!("Local Timetamp: {}", get_float_time());
|
|
|
|
|
|
|
|
|
|
match self.responses.send((Box::new(message), offset)) {
|
|
|
|
|
Ok(_) => log::debug!("Forwarded response to callback engine"),
|
|
|
|
|
Err(error) => log::error!("Response forward error {:?}", error),
|
|
|
|
|
}
|
|
|
|
|
},
|
|
|
|
|
None => log::error!("BackendMessage parse error: {:?}", packet),
|
|
|
|
|
}
|
|
|
|
|
},
|
|
|
|
|
None => log::error!("BackendMessage parse error: {:?}", packet),
|
|
|
|
|
}
|
|
|
|
|
},
|
|
|
|
|
// Server Message
|
|
|
|
|
2 => {
|
|
|
|
|
match ConnectMsg::from_bytes(&packet.data) {
|
|
|
|
|
Some(welcome_msg) => {
|
|
|
|
|
log::debug!("Received connect message: {:#?}", welcome_msg);
|
|
|
|
|
if welcome_msg.pw_valid == 0 {
|
|
|
|
|
let connect_response = ConnectMsg::from_welcome(welcome_msg, &self.password, self.uuid, self.client_name, self.username);
|
|
|
|
|
let response = BackendPacket::new(packet.header, packet.timestamp, packet.msg_type, self.last_seq_num+1, connect_response.as_bytes());
|
|
|
|
|
match self.stream.write(&response.as_bytes()) {
|
|
|
|
|
Err(error) => log::error!("Send error: {:?}", error),
|
|
|
|
|
Ok(_) => self.last_seq_num += 1,
|
|
|
|
|
}
|
|
|
|
|
} else if welcome_msg.state_valid == 0 {
|
|
|
|
|
log::error!("pw_valid but not state_valid");
|
|
|
|
|
} else {
|
|
|
|
|
self.connected = true;
|
|
|
|
|
log::info!("Connected to TM backend!");
|
|
|
|
|
// Server Message
|
|
|
|
|
2 => {
|
|
|
|
|
match ConnectMsg::from_bytes(packet.data) {
|
|
|
|
|
Some(welcome_msg) => {
|
|
|
|
|
log::debug!("Received connect message: {:#?}", welcome_msg);
|
|
|
|
|
if welcome_msg.pw_valid == 0 {
|
|
|
|
|
let connect_response = ConnectMsg::from_welcome(welcome_msg, &self.password, self.uuid, self.client_name, self.username);
|
|
|
|
|
let response = BackendPacket::new(packet.header, packet.timestamp, packet.msg_type, self.last_seq_num+1, connect_response.as_bytes());
|
|
|
|
|
match self.stream.write(&response.as_bytes()) {
|
|
|
|
|
Err(error) => log::error!("Send error: {:?}", error),
|
|
|
|
|
Ok(_) => self.last_seq_num += 1,
|
|
|
|
|
}
|
|
|
|
|
} else if welcome_msg.state_valid == 0 {
|
|
|
|
|
log::error!("pw_valid but not state_valid");
|
|
|
|
|
} else {
|
|
|
|
|
self.connected = true;
|
|
|
|
|
log::info!("Connected to TM backend!");
|
|
|
|
|
}
|
|
|
|
|
},
|
|
|
|
|
None => log::error!("Failed to parse welcome msg"),
|
|
|
|
|
}
|
|
|
|
|
},
|
|
|
|
|
None => log::error!("Failed to parse welcome msg"),
|
|
|
|
|
_ => log::warn!("Unhandled message type: {}", packet.msg_type),
|
|
|
|
|
}
|
|
|
|
|
},
|
|
|
|
|
_ => log::warn!("Unhandled message type: {}", packet.msg_type),
|
|
|
|
|
None => {
|
|
|
|
|
log::error!("Failed to parse BackendPacket({}): {}", read, String::from_utf8_lossy(&incoming));
|
|
|
|
|
// Sleep to prevent busy loop when TM is spamming 0 length packets
|
|
|
|
|
thread::sleep(Duration::from_millis(100));
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
},
|
|
|
|
|
None => {
|
|
|
|
|
log::error!("Failed to parse BackendPacket({}): {}", read_bytes, String::from_utf8_lossy(&data));
|
|
|
|
|
// Sleep to prevent busy loop when TM is spamming 0 length packets
|
|
|
|
|
thread::sleep(Duration::from_millis(100));
|
|
|
|
|
}
|
|
|
|
|
Err(_) => {},
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
}
|
|
|
|
|
@ -1153,47 +1106,6 @@ fn on_field_assigned(notice: tm::Notice, event: &mut Event, connection: &mut TMC
|
|
|
|
|
return messages;
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
fn on_timer_reset(notice: tm::Notice, event: &mut Event, connection: &mut TMConnection) -> Vec<MQTTMessage> {
|
|
|
|
|
log::info!("TIMER_RESTART: {:#?}", ¬ice);
|
|
|
|
|
|
|
|
|
|
let mut messages = Vec::new();
|
|
|
|
|
let Some(field_time) = ¬ice.field_time else { return Vec::new() };
|
|
|
|
|
let Some(field_info) = get_field_tuple(&field_time.field) else { return Vec::new() };
|
|
|
|
|
let Some(field) = get_field(&mut event.field_sets, field_info) else { return Vec::new() };
|
|
|
|
|
let Some(current_match) = field.last_known_match else { return Vec::new() };
|
|
|
|
|
let Some(m) = get_match(&mut event.divisions, current_match) else { return Vec::new() };
|
|
|
|
|
|
|
|
|
|
m.state = MatchState{
|
|
|
|
|
state: GameState::Scheduled,
|
|
|
|
|
start: get_float_time() - connection.time_offset,
|
|
|
|
|
};
|
|
|
|
|
|
|
|
|
|
match connection.state_cancels.remove(&field_info) {
|
|
|
|
|
None => {},
|
|
|
|
|
Some(cancel_tx) => {
|
|
|
|
|
match cancel_tx.send(()) {
|
|
|
|
|
Ok(_) => {},
|
|
|
|
|
Err(_) => {},
|
|
|
|
|
}
|
|
|
|
|
},
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
let match_state_topic = current_match.topic("/state");
|
|
|
|
|
let field_state_topic = field_info.topic("/state");
|
|
|
|
|
let serialized = serde_json::to_string_pretty(&m.state).unwrap();
|
|
|
|
|
messages.push(MQTTMessage{
|
|
|
|
|
topic: match_state_topic,
|
|
|
|
|
payload: serialized.clone(),
|
|
|
|
|
});
|
|
|
|
|
|
|
|
|
|
messages.push(MQTTMessage{
|
|
|
|
|
topic: field_state_topic,
|
|
|
|
|
payload: serialized,
|
|
|
|
|
});
|
|
|
|
|
|
|
|
|
|
return messages;
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
fn on_timer_stop(notice: tm::Notice, event: &mut Event, connection: &mut TMConnection) -> Vec<MQTTMessage> {
|
|
|
|
|
let mut messages = Vec::new();
|
|
|
|
|
let Some(field_time) = ¬ice.field_time else { return Vec::new() };
|
|
|
|
|
@ -1207,16 +1119,6 @@ fn on_timer_stop(notice: tm::Notice, event: &mut Event, connection: &mut TMConne
|
|
|
|
|
start: get_float_time() - connection.time_offset,
|
|
|
|
|
};
|
|
|
|
|
|
|
|
|
|
match connection.state_cancels.remove(&field_info) {
|
|
|
|
|
None => {},
|
|
|
|
|
Some(cancel_tx) => {
|
|
|
|
|
match cancel_tx.send(()) {
|
|
|
|
|
Ok(_) => {},
|
|
|
|
|
Err(_) => {},
|
|
|
|
|
}
|
|
|
|
|
},
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
let match_state_topic = current_match.topic("/state");
|
|
|
|
|
let field_state_topic = field_info.topic("/state");
|
|
|
|
|
let serialized = serde_json::to_string_pretty(&m.state).unwrap();
|
|
|
|
|
@ -1234,7 +1136,6 @@ fn on_timer_stop(notice: tm::Notice, event: &mut Event, connection: &mut TMConne
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
fn on_timer_start(notice: tm::Notice, event: &mut Event, connection: &mut TMConnection) -> Vec<MQTTMessage> {
|
|
|
|
|
log::info!("TIMER_START: {:#?}", ¬ice);
|
|
|
|
|
let Some(field_time) = ¬ice.field_time else { return Vec::new() };
|
|
|
|
|
let Some(field_info) = get_field_tuple(&field_time.field) else { return Vec::new() };
|
|
|
|
|
let Some(field) = get_field(&mut event.field_sets, field_info) else { return Vec::new() };
|
|
|
|
|
@ -1242,6 +1143,7 @@ fn on_timer_start(notice: tm::Notice, event: &mut Event, connection: &mut TMConn
|
|
|
|
|
let Some(m) = get_match(&mut event.divisions, tuple) else { return Vec::new() };
|
|
|
|
|
let Some(block_list) = &field_time.block_list else { return Vec::new() };
|
|
|
|
|
let Some(current_block_idx) = &field_time.current_block else { return Vec::new() };
|
|
|
|
|
let Some(current_block_start) = &field_time.current_block_start else { return Vec::new() };
|
|
|
|
|
let Some(current_block_end) = &field_time.current_block_end else { return Vec::new() };
|
|
|
|
|
|
|
|
|
|
let current_block = &block_list.entries[*current_block_idx as usize];
|
|
|
|
|
@ -1251,7 +1153,7 @@ fn on_timer_start(notice: tm::Notice, event: &mut Event, connection: &mut TMConn
|
|
|
|
|
if current_block.r#type == Some(2) { //Auto
|
|
|
|
|
m.state = MatchState{
|
|
|
|
|
state: GameState::Autonomous,
|
|
|
|
|
start: *current_block_end - (current_block.seconds() as f64) - connection.time_offset,
|
|
|
|
|
start: *current_block_start - connection.time_offset,
|
|
|
|
|
};
|
|
|
|
|
let field_state_topic = field_info.topic("/state");
|
|
|
|
|
let match_state_topic = tuple.topic("/state");
|
|
|
|
|
@ -1281,7 +1183,7 @@ fn on_timer_start(notice: tm::Notice, event: &mut Event, connection: &mut TMConn
|
|
|
|
|
} else if current_block.r#type == Some(3) { //Driver
|
|
|
|
|
m.state = MatchState{
|
|
|
|
|
state: GameState::Driver,
|
|
|
|
|
start: *current_block_end - (current_block.seconds() as f64) - connection.time_offset,
|
|
|
|
|
start: *current_block_start - connection.time_offset,
|
|
|
|
|
};
|
|
|
|
|
let field_state_topic = field_info.topic("/state");
|
|
|
|
|
let match_state_topic = tuple.topic("/state");
|
|
|
|
|
@ -1313,20 +1215,12 @@ fn on_timer_start(notice: tm::Notice, event: &mut Event, connection: &mut TMConn
|
|
|
|
|
return messages;
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
struct NoticeWithOffset {
|
|
|
|
|
notice: tm::Notice,
|
|
|
|
|
offset: f64,
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
enum Work {
|
|
|
|
|
Notice(Box<NoticeWithOffset>),
|
|
|
|
|
Offset(f64),
|
|
|
|
|
Notice(Box<tm::Notice>),
|
|
|
|
|
State(StateChange),
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
fn main() {
|
|
|
|
|
dotenv().ok();
|
|
|
|
|
|
|
|
|
|
env_logger::init();
|
|
|
|
|
|
|
|
|
|
let mut callbacks: HashMap<tm::NoticeId, NoticeCallback> = HashMap::new();
|
|
|
|
|
@ -1336,11 +1230,8 @@ fn main() {
|
|
|
|
|
callbacks.insert(tm::NoticeId::NoticeFieldTimerStarted, on_timer_start);
|
|
|
|
|
callbacks.insert(tm::NoticeId::NoticeFieldMatchAssigned, on_field_assigned);
|
|
|
|
|
callbacks.insert(tm::NoticeId::NoticeFieldTimerStopped, on_timer_stop);
|
|
|
|
|
callbacks.insert(tm::NoticeId::NoticeFieldResetTimer, on_timer_reset);
|
|
|
|
|
|
|
|
|
|
let mqtt_host = env::var("MQTT_HOST").unwrap_or(String::from("localhost"));
|
|
|
|
|
log::debug!("Connecting to MQTT using address {mqtt_host}:1883");
|
|
|
|
|
let mut mqttoptions = MqttOptions::new("vex-bridge", mqtt_host, 1883);
|
|
|
|
|
let mut mqttoptions = MqttOptions::new("vex-bridge", "localhost", 1883);
|
|
|
|
|
mqttoptions.set_keep_alive(Duration::from_secs(5));
|
|
|
|
|
mqttoptions.set_last_will(LastWill{
|
|
|
|
|
topic: String::from("bridge/status"),
|
|
|
|
|
@ -1363,10 +1254,8 @@ fn main() {
|
|
|
|
|
rand::thread_rng().fill_bytes(&mut uuid);
|
|
|
|
|
let mut client_name = [0u8;32];
|
|
|
|
|
rand::thread_rng().fill_bytes(&mut client_name);
|
|
|
|
|
let tm_host = env::var("TM_HOST").unwrap_or(String::from("127.0.0.1"));
|
|
|
|
|
let username: [u8;16] = [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0];
|
|
|
|
|
let password = env::var("TM_PASSWORD").unwrap_or(String::from(""));
|
|
|
|
|
let (mut tm_client, mut tm_connection) = TMClient::new(uuid, client_name, tm_host, password, username);
|
|
|
|
|
let (mut tm_client, mut tm_connection) = TMClient::new(uuid, client_name, String::from(""), username);
|
|
|
|
|
let tm_thread = thread::spawn(move ||
|
|
|
|
|
while running {
|
|
|
|
|
tm_client.process();
|
|
|
|
|
@ -1445,15 +1334,8 @@ fn main() {
|
|
|
|
|
while running {
|
|
|
|
|
match tm_connection.work_queue.recv() {
|
|
|
|
|
Ok(work) => match work {
|
|
|
|
|
Work::Offset(offset) => {
|
|
|
|
|
let current = get_float_time() - offset;
|
|
|
|
|
client.publish("time", QoS::AtMostOnce, false, serde_json::to_vec(¤t).unwrap()).unwrap();
|
|
|
|
|
},
|
|
|
|
|
Work::Notice(package) => {
|
|
|
|
|
let current = get_float_time() - package.offset;
|
|
|
|
|
let notice = package.notice;
|
|
|
|
|
Work::Notice(notice) => {
|
|
|
|
|
let callback = callbacks.get(¬ice.id());
|
|
|
|
|
client.publish("time", QoS::AtMostOnce, false, serde_json::to_vec(¤t).unwrap()).unwrap();
|
|
|
|
|
match callback {
|
|
|
|
|
None => {
|
|
|
|
|
match notice.id {
|
|
|
|
|
@ -1462,7 +1344,7 @@ fn main() {
|
|
|
|
|
}
|
|
|
|
|
},
|
|
|
|
|
Some(callback) => {
|
|
|
|
|
let messages = callback(notice, &mut event, &mut tm_connection);
|
|
|
|
|
let messages = callback(*notice, &mut event, &mut tm_connection);
|
|
|
|
|
for message in messages {
|
|
|
|
|
let result = client.publish(message.topic, QoS::AtLeastOnce, true, message.payload);
|
|
|
|
|
match result {
|
|
|
|
|
|