Update chirpstack_api dependency.

Author: brocaar

Cargo.lock

@@ -9,7 +9,7 @@
   publish = false
 
 [dependencies]
-  chirpstack_api = { version = "4.12", default-features = false }
+  chirpstack_api = { version = "4.13.0-test.1", default-features = false }
   serde_json = "1.0"
   zmq = "0.10"
   clap = { version = "4.5", default-features = false, features = [
@@ -28,10 +28,9 @@
   rand = "0.9"
   chrono = "0.4"
   base64 = "0.22"
-  prost = "0.13"
-  prost-types = "0.13"
   prometheus-client = "0.23"
   anyhow = "1.0"
+  thiserror = "1.0"
 
   [package.metadata.deb]
     assets = [

Cargo.toml

@@ -1,10 +1,19 @@
 use std::time::Duration;
 
 use anyhow::Result;
-use prost::Message;
+use chirpstack_api::{gw, prost::Message};
 
 use super::socket::ZMQ_CONTEXT;
 
+#[derive(thiserror::Error, Debug)]
+pub enum Error {
+    #[error("Timeout")]
+    Timeout,
+
+    #[error(transparent)]
+    Anyhow(#[from] anyhow::Error),
+}
+
 pub fn get_socket(endpoint: &str) -> Result<zmq::Socket, zmq::Error> {
     info!(
         "Creating new socket for receiving events, endpoint: {}",
@@ -19,23 +28,6 @@ pub fn get_socket(endpoint: &str) -> Result<zmq::Socket, zmq::Error> {
     Ok(sock)
 }
 
-pub enum Event {
-    // Reading event timed out.
-    Timeout,
-
-    // Error reading event.
-    Error(String),
-
-    // Unknown event.
-    Unknown(String),
-
-    // Uplink event.
-    Uplink(Box<chirpstack_api::gw::UplinkFrame>),
-
-    // Stats event.
-    Stats(Box<chirpstack_api::gw::GatewayStats>),
-}
-
 pub struct Reader<'a> {
     sub_sock: &'a zmq::Socket,
     timeout: Duration,
@@ -48,48 +40,20 @@ impl<'a> Reader<'a> {
 }
 
 impl Iterator for Reader<'_> {
-    type Item = Event;
+    type Item = Result<gw::Event, Error>;
 
-    fn next(&mut self) -> Option<Event> {
+    fn next(&mut self) -> Option<Result<gw::Event, Error>> {
         // set poller so that we can timeout
         let mut items = [self.sub_sock.as_poll_item(zmq::POLLIN)];
         zmq::poll(&mut items, self.timeout.as_millis() as i64).unwrap();
         if !items[0].is_readable() {
-            return Some(Event::Timeout);
+            return Some(Err(Error::Timeout));
         }
 
-        let msg = self.sub_sock.recv_multipart(0).unwrap();
-        match handle_message(msg) {
-            Ok(v) => Some(v),
-            Err(err) => Some(Event::Error(err.to_string())),
+        let b = self.sub_sock.recv_bytes(0).unwrap();
+        match gw::Event::decode(b.as_slice()).map_err(|e| Error::Anyhow(anyhow::Error::new(e))) {
+            Ok(v) => Some(Ok(v)),
+            Err(e) => Some(Err(e)),
         }
     }
 }
-
-fn handle_message(msg: Vec<Vec<u8>>) -> Result<Event> {
-    if msg.len() != 2 {
-        return Err(anyhow!("Event must have two frames"));
-    }
-
-    let event = String::from_utf8(msg[0].clone())?;
-
-    Ok(match event.as_str() {
-        "up" => match parse_up(&msg[1]) {
-            Ok(v) => Event::Uplink(Box::new(v)),
-            Err(err) => Event::Error(err.to_string()),
-        },
-        "stats" => match parse_stats(&msg[1]) {
-            Ok(v) => Event::Stats(Box::new(v)),
-            Err(err) => Event::Error(err.to_string()),
-        },
-        _ => Event::Unknown(event),
-    })
-}
-
-fn parse_up(msg: &[u8]) -> Result<chirpstack_api::gw::UplinkFrame> {
-    Ok(chirpstack_api::gw::UplinkFrame::decode(msg)?)
-}
-
-fn parse_stats(msg: &[u8]) -> Result<chirpstack_api::gw::GatewayStats> {
-    Ok(chirpstack_api::gw::GatewayStats::decode(msg)?)
-}

src/events.rs

@@ -4,8 +4,7 @@ use std::sync::{Arc, Mutex};
 use std::{thread, time};
 
 use anyhow::Result;
-use chirpstack_api::gw;
-use prost::Message;
+use chirpstack_api::{gw, prost::Message};
 use rand::Rng;
 
 use super::commands;
@@ -318,21 +317,17 @@ fn events_loop(state: Arc<State>, stop_receive: Receiver<signals::Signal>) {
         }
 
         match cmd {
-            events::Event::Uplink(up) => {
-                events_up(&state, *up);
-            }
-            events::Event::Stats(stats) => {
-                events_stats(&state, *stats);
-            }
-            events::Event::Timeout => {
-                continue;
-            }
-            events::Event::Error(err) => {
-                error!("Read event error, error: {}", err);
-            }
-            events::Event::Unknown(event) => {
-                warn!("Unknown event received, event: {}", event);
-            }
+            Ok(v) => match v.event {
+                Some(gw::event::Event::UplinkFrame(pl)) => events_up(&state, pl),
+                Some(gw::event::Event::GatewayStats(pl)) => events_stats(&state, pl),
+                _ => continue,
+            },
+            Err(e) => match e {
+                events::Error::Timeout => continue,
+                _ => {
+                    warn!("Read event error, error: {}", e);
+                }
+            },
         }
     }
 }
@@ -472,12 +467,13 @@ fn handle_pull_resp(state: &Arc<State>, data: &[u8]) -> Result<()> {
         }
     };
 
-    let mut buf = Vec::new();
-    pl.encode(&mut buf).unwrap();
+    let pl = gw::Command {
+        command: Some(gw::command::Command::SendDownlinkFrame(pl)),
+    };
+    let b = pl.encode_to_vec();
 
     // send 'down' command with payload
-    sock.send("down", zmq::SNDMORE).unwrap();
-    sock.send(buf, 0).unwrap();
+    sock.send(b, 0).unwrap();
 
     // set poller so that we can timeout after 100ms
     let mut items = [sock.as_poll_item(zmq::POLLIN)];

src/forwarder.rs

@@ -3,13 +3,13 @@ use std::time::Duration;
 use std::time::SystemTime;
 
 use anyhow::Result;
-use base64::{engine::general_purpose, Engine as _};
+use base64::{Engine as _, engine::general_purpose};
 use chrono::{DateTime, Utc};
 use serde::de::Error;
 use serde::{Deserialize, Deserializer, Serialize, Serializer};
 use serde_json::Value;
 
-use chirpstack_api::gw;
+use chirpstack_api::{gw, prost_types};
 
 const PROTOCOL_VERSION: u8 = 0x02;