fix busy cpu loop

Makefile

@@ -4,9 +4,12 @@ build-linux-on-osx:
 build: build-linux-on-osx
 
 run:
-	RUST_BACKTRACE=1 cargo run -- \
+	cargo run --release -- \
         --udp-bind=127.0.0.1:1113 \
         --http-bind=127.0.0.1:8080 \
+        --token foo\
+        --agent-key foo\
+        --account-url foo\
 	    --debug
 
 example_php_client:

openmetrics_udpserver/src/processor.rs

@@ -40,7 +40,8 @@ impl Processor {
         let regex_allowed_chars = Regex::new(r"^[^a-zA-Z_:]|[^a-zA-Z0-9_:]")
             .expect("Unable to compile metrics naming regex, should not happen");
         loop {
-            match receiver.try_recv() {
+
+            match receiver.recv().await {
                 Ok(inbound_metric) => {
                     let metric_name = regex_allowed_chars
                         .replace_all(&inbound_metric.name.replace('.', "_"), "")
@@ -75,11 +76,9 @@ impl Processor {
                         }
                     }
                 }
-                Err(TryRecvError::Empty | TryRecvError::Lagged(_)) => {
-                    yield_now().await;
-                }
-                Err(TryRecvError::Closed) => {
-                    panic!("All metric senders were dropped, should not happen")
+                Err(e) => {
+                    eprintln!("processor recv error {:#?}, investigate!", e);
+                    ::tokio::time::sleep(Duration::from_millis(300)).await;
                 }
             }
         }

openmetrics_udpserver/src/serverdensity/aggregator.rs

@@ -132,7 +132,7 @@ impl ServerDensityAggregator {
             loop {
                 let mut i = 0;
 
-                match receiver.try_recv() {
+                match receiver.recv().await {
                     Ok(metric) => {
                         let metric_name = regex.replace_all(&metric.name, "").trim().to_string();
 
@@ -164,11 +164,9 @@ impl ServerDensityAggregator {
                             );
                         }
                     }
-                    Err(TryRecvError::Closed) => {
-                        panic!("channel disconnected, should never happen.");
-                    }
-                    Err(TryRecvError::Empty | TryRecvError::Lagged(_)) => {
-                        break;
+                    Err(e) => {
+                        eprintln!("aggregator recv error {:#?}, investigate!", e);
+                        ::tokio::time::sleep(Duration::from_millis(300)).await;
                     }
                 };
             }

openmetrics_udpserver/src/udp_server.rs

@@ -24,19 +24,18 @@ impl UdpServer {
             .await
             .expect("Unable to bind UDP Server");
         loop {
-            if udp_socket.readable().await.is_ok() {
-                let mut buf = [0; 300];
-                if let Ok(read_bytes) = udp_socket.try_recv(&mut buf) {
-                    match self.decode_buffer(&buf, read_bytes) {
-                        Ok(inbound_metric) => {
-                            if let Err(err) = self.metric_sender.send(inbound_metric) {
-                                eprintln!("Unable to process inbound metric: {}", err);
-                            }
-                        }
-                        Err(err) => {
-                            eprintln!("could not decode message from socket: {}", err);
+            let mut buf = [0; 300];
+            if let Ok(read_bytes) = udp_socket.recv(&mut buf).await {
+                match self.decode_buffer(&buf, read_bytes) {
+                    Ok(inbound_metric) => {
+                        if let Err(err) = self.metric_sender.send(inbound_metric) {
+                            eprintln!("Unable to process inbound metric: {}", err);
                         }
                     }
+                    Err(err) => {
+                        // it could be, that we are so fast that we read a part of the message, may we need to improve this code.
+                        eprintln!("could not decode message from socket: {}", err);
+                    }
                 }
             }
         }