fix: fixed publish in webservice

Signed-off-by: Jan Zachmann <50990105+JanZachmann@users.noreply.github.com>

Author: JanZachmann

Cargo.lock

@@ -8,7 +8,7 @@ license = "MIT OR Apache-2.0"
 name = "omnect-device-service"
 readme = "README.md"
 repository = "https://github.com/omnect/omnect-device-service.git"
-version = "0.41.8"
+version = "0.41.9"
 
 [dependencies]
 actix-server = { version = "2.6", default-features = false }

Cargo.toml

@@ -5,19 +5,25 @@ use crate::{
 use actix_server::ServerHandle;
 use actix_web::{App, HttpResponse, HttpServer, http::StatusCode, web};
 use anyhow::{Context, Result};
-use lazy_static::lazy_static;
 use log::{debug, error, info};
 use reqwest::{Client, header};
 use reqwest_middleware::{ClientBuilder, ClientWithMiddleware};
 use reqwest_retry::{RetryTransientMiddleware, policies::ExponentialBackoff};
 use serde::{Deserialize, Serialize};
 use serde_json::json;
-use std::{collections::HashMap, env, path::Path, str::FromStr, sync::LazyLock};
+use std::{
+    collections::HashMap,
+    env,
+    path::Path,
+    str::FromStr,
+    sync::{LazyLock, OnceLock},
+};
 use tokio::{
     sync::{Mutex, mpsc, oneshot},
     time::Duration,
 };
 
+static IS_WEBSERVICE_DISABLED: OnceLock<bool> = OnceLock::new();
 static PUBLISH_CHANNEL_MAP: LazyLock<Mutex<serde_json::Map<String, serde_json::Value>>> =
     LazyLock::new(|| Mutex::new(serde_json::Map::default()));
 static PUBLISH_STATUS_MAP: LazyLock<Mutex<serde_json::Map<String, serde_json::Value>>> =
@@ -28,7 +34,6 @@ static PUBLISH_CLIENT: LazyLock<Mutex<ClientWithMiddleware>> = LazyLock::new(||
     Mutex::new(
         ClientBuilder::new(
             Client::builder()
-                .timeout(Duration::from_secs(3))
                 .danger_accept_invalid_certs(true)
                 .build()
                 .expect("building ClientWithMiddleware failed"),
@@ -88,30 +93,27 @@ struct PublishEndpointRequest {
     endpoint: PublishEndpoint,
 }
 
-lazy_static! {
-    static ref IS_WEBSERVICE_ENABLED: bool = {
-        env::var("DISABLE_WEBSERVICE")
-            .unwrap_or("false".to_string())
-            .to_lowercase()
-            != "true"
-    };
-}
-
 pub struct WebService {
     srv_handle: ServerHandle,
 }
 
 impl WebService {
     pub async fn run(tx_request: mpsc::Sender<CommandRequest>) -> Result<Option<Self>> {
         // we only start web service feature if not explicitly disabled by 'DISABLE_WEBSERVICE="true"' env var
-        if !(*IS_WEBSERVICE_ENABLED) {
+        if *IS_WEBSERVICE_DISABLED.get_or_init(|| {
+            env::var("DISABLE_WEBSERVICE")
+                .unwrap_or("false".to_string())
+                .to_lowercase()
+                == "true"
+        }) {
             info!("WebService is disabled");
             return Ok(None);
         };
 
         info!("WebService is enabled");
 
         if matches!(Path::new(&publish_endpoints_path!()).try_exists(), Ok(true)) {
+            debug!("restore publish endpoints");
             *PUBLISH_ENDPOINTS.lock().await = from_json_file(publish_endpoints_path!())?;
         }
 
@@ -396,7 +398,7 @@ impl WebService {
 }
 
 pub async fn publish(channel: PublishChannel, value: serde_json::Value) {
-    if !(*IS_WEBSERVICE_ENABLED) {
+    if *IS_WEBSERVICE_DISABLED.wait() {
         debug!("publish: skip since feature not enabled");
         return;
     }
@@ -416,7 +418,7 @@ pub async fn publish(channel: PublishChannel, value: serde_json::Value) {
         .insert(channel.to_status_string(), value.clone());
 
     for endpoint in PUBLISH_ENDPOINTS.lock().await.values() {
-        if let Err(e) = publish_to_endpoint(&msg, endpoint).await {
+        if let Err(e) = publish_to_endpoint(&msg, endpoint) {
             error!("publish: {e:#}");
         }
     }
@@ -426,7 +428,7 @@ async fn republish_to_endpoint(endpoint: &PublishEndpoint) -> HttpResponse {
     for (channel, value) in PUBLISH_CHANNEL_MAP.lock().await.iter() {
         let msg = json!({"channel": channel, "data": value});
 
-        if let Err(e) = publish_to_endpoint(&msg, endpoint).await {
+        if let Err(e) = publish_to_endpoint(&msg, endpoint) {
             error!("republish_to_endpoint: {e:#}");
             return HttpResponse::InternalServerError().body(e.to_string());
         }
@@ -435,10 +437,12 @@ async fn republish_to_endpoint(endpoint: &PublishEndpoint) -> HttpResponse {
     HttpResponse::Ok().finish()
 }
 
-async fn publish_to_endpoint(msg: &serde_json::Value, endpoint: &PublishEndpoint) -> Result<()> {
+fn publish_to_endpoint(msg: &serde_json::Value, endpoint: &PublishEndpoint) -> Result<()> {
+    let msg = msg.to_string();
+    let url = endpoint.url.clone();
     let mut headers = header::HeaderMap::new();
 
-    info!("try to send {msg} to {}.", endpoint.url);
+    info!("try to send {msg} to {url}");
 
     for h in &endpoint.headers {
         headers.insert(
@@ -447,18 +451,21 @@ async fn publish_to_endpoint(msg: &serde_json::Value, endpoint: &PublishEndpoint
         );
     }
 
-    PUBLISH_CLIENT
-        .lock()
-        .await
-        .post(&endpoint.url)
-        .headers(headers)
-        .body(msg.to_string())
-        .send()
-        .await
-        .context("sending request failed")?
-        .error_for_status()
-        .context("response error")
-        .map(|_| ())
+    tokio::spawn(async move {
+        if let Err(e) = PUBLISH_CLIENT
+            .lock()
+            .await
+            .post(url)
+            .headers(headers)
+            .body(msg)
+            .send()
+            .await
+        {
+            error!("publish_to_endpoint: {e:#}")
+        }
+    });
+
+    Ok(())
 }
 
 async fn save_publish_endpoints() -> Result<()> {