ExternalVendorCode/chirpstack
1
0
Fork 0
mirror of https://github.com/chirpstack/chirpstack.git synced 2025-03-13 15:56:51 +00:00
Code Issues Actions 2 Packages Projects Releases Wiki Activity

Re-use http clients within integrations.

Browse Source 
This reduces CPU load, especially in case when HTTPS is being used
as connections will be pooled by the http client.
This commit is contained in:
Orne Brocaar 2025-01-30 19:17:44 +00:00
parent 99613014ad
commit 32a9aadbae
10 changed files with 160 additions and 51 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
chirpstack/src/integration/aws_sns.rs
View File

@ -1,9 +1,12 @@
use std::collections::HashMap;
use std::sync::OnceLock;
use std::time::Duration;
use anyhow::Result;
use async_trait::async_trait;
use base64::{engine::general_purpose, Engine as _};
use prost::Message;
use reqwest::Client;
use tracing::{info, trace};
use super::Integration as IntegrationTrait;
@ -11,13 +14,25 @@ use crate::storage::application::AwsSnsConfiguration;
use chirpstack_api::api::Encoding;
use chirpstack_api::integration;
static CLIENT: OnceLock<Client> = OnceLock::new();
fn get_client() -> Client {
CLIENT
.get_or_init(|| {
Client::builder()
.timeout(Duration::from_secs(5))
.build()
.unwrap()
})
.clone()
}
pub struct Integration {
json: bool,
access_key_id: String,
secret_access_key: String,
region: String,
topic_arn: String,
client: reqwest::Client,
}
impl Integration {
@ -35,7 +50,6 @@ impl Integration {
access_key_id: conf.access_key_id.clone(),
secret_access_key: conf.secret_access_key.clone(),
region: conf.region.clone(),
client: reqwest::Client::new(),
})
}
@ -97,7 +111,7 @@ impl Integration {
headers.insert(reqwest::header::AUTHORIZATION, s.parse()?);
self.client
get_client()
.post(url)
.headers(headers)
.body(body)

20
chirpstack/src/integration/azure_service_bus.rs
View File

@ -1,4 +1,5 @@
use std::collections::HashMap;
use std::sync::OnceLock;
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use anyhow::Result;
@ -16,8 +17,20 @@ use crate::storage::application::AzureServiceBusConfiguration;
use chirpstack_api::api::Encoding;
use chirpstack_api::integration;
static CLIENT: OnceLock<Client> = OnceLock::new();
fn get_client() -> Client {
CLIENT
.get_or_init(|| {
Client::builder()
.timeout(Duration::from_secs(5))
.build()
.unwrap()
})
.clone()
}
pub struct Integration {
timeout: Duration,
json: bool,
uri: String,
key_name: String,
@ -31,7 +44,6 @@ impl Integration {
let kv = parse_connection_string(&conf.connection_string);
Ok(Integration {
timeout: Duration::from_secs(5),
json: match Encoding::try_from(conf.encoding)
.map_err(|_| anyhow!("Invalid encoding"))?
{
@ -65,7 +77,6 @@ impl Integration {
&(SystemTime::now() + Duration::from_secs(60 * 5)),
)?;
let client = Client::builder().timeout(self.timeout).build()?;
let mut headers = HeaderMap::new();
headers.insert(AUTHORIZATION, token.parse()?);
@ -89,7 +100,7 @@ impl Integration {
);
info!(event = %event, dev_eui = %dev_eui, "Publishing event");
let res = client
let res = get_client()
.post(format!("{}/messages", self.uri))
.body(pl.to_string())
.headers(headers)
@ -307,7 +318,6 @@ pub mod test {
let server = MockServer::start();
let i = Integration {
timeout: Duration::from_secs(5),
json: true,
uri: server.url(""),
key_name: "key-name".to_string(),

19
chirpstack/src/integration/gcp_pub_sub.rs
View File

@ -1,4 +1,5 @@
use std::collections::HashMap;
use std::sync::OnceLock;
use std::time::Duration;
use anyhow::{Context, Result};
@ -16,12 +17,24 @@ use crate::storage::application::GcpPubSubConfiguration;
use chirpstack_api::api::Encoding;
use chirpstack_api::integration;
static CLIENT: OnceLock<Client> = OnceLock::new();
fn get_client() -> Client {
CLIENT
.get_or_init(|| {
Client::builder()
.timeout(Duration::from_secs(5))
.build()
.unwrap()
})
.clone()
}
pub struct Integration {
json: bool,
project_id: String,
topic_name: String,
service_account: gcp_auth::CustomServiceAccount,
timeout: Duration,
}
#[derive(Serialize)]
@ -57,7 +70,6 @@ impl Integration {
project_id: conf.project_id.clone(),
topic_name: conf.topic_name.clone(),
service_account,
timeout: Duration::from_secs(5),
})
}
@ -93,7 +105,6 @@ impl Integration {
.await
.context("Get GCP bearer token")?;
let client = Client::builder().timeout(self.timeout).build()?;
let mut headers = HeaderMap::new();
headers.insert(CONTENT_TYPE, "application/json".parse().unwrap());
headers.insert(
@ -101,7 +112,7 @@ impl Integration {
format!("Bearer {}", token.as_str()).parse().unwrap(),
);
let res = client
let res = get_client()
.post(format!(
"https://pubsub.googleapis.com/v1/{}:publish",
topic

20
chirpstack/src/integration/http.rs
View File

@ -1,4 +1,5 @@
use std::collections::HashMap;
use std::sync::OnceLock;
use std::time::Duration;
use anyhow::Result;
@ -12,8 +13,20 @@ use super::Integration as IntegrationTrait;
use crate::storage::application::HttpConfiguration;
use chirpstack_api::integration;
static CLIENT: OnceLock<Client> = OnceLock::new();
fn get_client() -> Client {
CLIENT
.get_or_init(|| {
Client::builder()
.timeout(Duration::from_secs(5))
.build()
.unwrap()
})
.clone()
}
pub struct Integration {
timeout: Duration,
endpoints: Vec<String>,
headers: HashMap<String, String>,
json: bool,
@ -24,7 +37,6 @@ impl Integration {
trace!("Initializing http integration");
Integration {
timeout: Duration::from_secs(5),
headers: conf.headers.clone(),
json: conf.json,
endpoints: conf
@ -36,7 +48,6 @@ impl Integration {
}
async fn post_event(&self, event: &str, b: Vec<u8>) -> Result<()> {
let client = Client::builder().timeout(self.timeout).build()?;
let mut headers = HeaderMap::new();
for (k, v) in &self.headers {
@ -51,7 +62,7 @@ impl Integration {
for url in &self.endpoints {
info!(event = %event, url = %url, "Posting event");
let res = client
let res = get_client()
.post(url)
.body(b.clone())
.query(&[("event", event)])
@ -214,7 +225,6 @@ pub mod test {
let server = MockServer::start();
let i = Integration {
timeout: Duration::from_secs(5),
endpoints: vec![server.url("/")],
headers: [("Foo".to_string(), "Bar".to_string())]
.iter()

22
chirpstack/src/integration/ifttt.rs
View File

@ -1,4 +1,5 @@
use std::collections::HashMap;
use std::sync::OnceLock;
use std::time::Duration;
use anyhow::Result;
@ -13,6 +14,19 @@ use crate::codec;
use crate::storage::application::IftttConfiguration;
use chirpstack_api::integration;
static CLIENT: OnceLock<Client> = OnceLock::new();
fn get_client() -> Client {
CLIENT
.get_or_init(|| {
Client::builder()
.timeout(Duration::from_secs(5))
.build()
.unwrap()
})
.clone()
}
#[derive(Serialize, Deserialize)]
struct Values {
#[serde(skip_serializing_if = "String::is_empty")]
@ -63,12 +77,16 @@ impl Integration {
format!("{}/trigger/{}/with/key/{}", self.server, event, self.key)
};
let client = Client::builder().timeout(Duration::from_secs(5)).build()?;
let mut headers = HeaderMap::new();
headers.insert(CONTENT_TYPE, "application/json".parse().unwrap());
info!(event = %event, "Sending event to IFTTT");
let res = client.post(url).json(&v).headers(headers).send().await?;
let res = get_client()
.post(url)
.json(&v)
.headers(headers)
.send()
.await?;
match res.error_for_status() {
Ok(_) => Ok(()),
Err(e) => {

22
chirpstack/src/integration/influxdb.rs
View File

@ -1,5 +1,6 @@
use std::collections::HashMap;
use std::fmt;
use std::sync::OnceLock;
use std::time::Duration;
use anyhow::Result;
@ -13,8 +14,20 @@ use crate::storage::application::InfluxDbConfiguration;
use chirpstack_api::api::{InfluxDbPrecision, InfluxDbVersion};
use chirpstack_api::integration;
static CLIENT: OnceLock<Client> = OnceLock::new();
fn get_client() -> Client {
CLIENT
.get_or_init(|| {
Client::builder()
.timeout(Duration::from_secs(5))
.build()
.unwrap()
})
.clone()
}
pub struct Integration {
timeout: Duration,
endpoint: String,
version: InfluxDbVersion,
@ -36,7 +49,6 @@ impl Integration {
trace!("Initializing InfluxDB integration");
Ok(Integration {
timeout: Duration::from_secs(5),
endpoint: conf.endpoint.clone(),
version: InfluxDbVersion::try_from(conf.version)
.map_err(|_| anyhow!("Invalid version"))?,
@ -66,8 +78,6 @@ impl Integration {
measurements.sort();
let body = measurements.join("\n");
let client = Client::builder().timeout(self.timeout).build()?;
let mut headers = HeaderMap::new();
headers.insert(CONTENT_TYPE, "text/plain".parse().unwrap());
if self.version == InfluxDbVersion::Influxdb2 {
@ -87,7 +97,7 @@ impl Integration {
}
}
let mut req = client
let mut req = get_client()
.post(&self.endpoint)
.body(body)
.query(&query)
@ -477,7 +487,6 @@ pub mod test {
let server = MockServer::start();
let i = Integration {
timeout: Duration::from_secs(5),
endpoint: server.url("/write"),
version: InfluxDbVersion::Influxdb1,
db: "testdb".into(),
@ -832,7 +841,6 @@ device_uplink,application_name=test-app,dev_eui=0102030405060708,device_name=tes
let server = MockServer::start();
let i = Integration {
timeout: Duration::from_secs(5),
endpoint: server.url("/write"),
version: InfluxDbVersion::Influxdb2,
db: "".into(),

25
chirpstack/src/integration/loracloud/client.rs
View File

@ -1,4 +1,5 @@
use std::fmt;
use std::sync::OnceLock;
use std::time::Duration;
use anyhow::Result;
@ -13,6 +14,19 @@ use crate::gpstime::ToGpsTime;
use crate::uplink::helpers;
use lrwn::EUI64;
static CLIENT: OnceLock<Client> = OnceLock::new();
fn get_client() -> Client {
CLIENT
.get_or_init(|| {
Client::builder()
.timeout(Duration::from_secs(5))
.build()
.unwrap()
})
.clone()
}
#[derive(Error, Debug)]
pub enum Error {
#[error("No location")]
@ -25,7 +39,6 @@ pub enum Error {
pub struct ApiClient {
uri: String,
token: String,
timeout: Duration,
}
impl ApiClient {
@ -33,7 +46,6 @@ impl ApiClient {
ApiClient {
uri: uri.to_string(),
token: token.to_string(),
timeout: Duration::from_secs(5),
}
}
@ -117,7 +129,6 @@ impl ApiClient {
pub async fn uplink_send(&self, req: &UplinkRequest) -> Result<UplinkResponse> {
let endpoint = format!("{}/api/v1/device/send", self.uri);
let client = Client::builder().timeout(self.timeout).build()?;
let mut headers = HeaderMap::new();
headers.insert(CONTENT_TYPE, "application/json".parse().unwrap());
headers.insert(
@ -125,7 +136,7 @@ impl ApiClient {
self.token.parse()?,
);
let res = client
let res = get_client()
.post(endpoint)
.headers(headers)
.json(req)
@ -138,7 +149,6 @@ impl ApiClient {
async fn request(&self, endpoint: &str, body: &str) -> Result<Response> {
let endpoint = format!("{}{}", self.uri, endpoint);
let client = Client::builder().timeout(self.timeout).build()?;
let mut headers = HeaderMap::new();
headers.insert(CONTENT_TYPE, "application/json".parse().unwrap());
headers.insert(
@ -146,7 +156,7 @@ impl ApiClient {
self.token.parse()?,
);
let res = client
let res = get_client()
.post(endpoint)
.body(body.to_string())
.headers(headers)
@ -160,7 +170,6 @@ impl ApiClient {
async fn v3_request(&self, endpoint: &str, body: &str) -> Result<V3Response> {
let endpoint = format!("{}{}", self.uri, endpoint);
let client = Client::builder().timeout(self.timeout).build()?;
let mut headers = HeaderMap::new();
headers.insert(CONTENT_TYPE, "application/json".parse().unwrap());
headers.insert(
@ -168,7 +177,7 @@ impl ApiClient {
self.token.parse()?,
);
let res = client
let res = get_client()
.post(endpoint)
.body(body.to_string())
.headers(headers)

20
chirpstack/src/integration/mydevices.rs
View File

@ -1,4 +1,5 @@
use std::collections::HashMap;
use std::sync::OnceLock;
use std::time::Duration;
use anyhow::Result;
@ -13,6 +14,19 @@ use super::Integration as IntegrationTrait;
use crate::storage::application::MyDevicesConfiguration;
use chirpstack_api::integration;
static CLIENT: OnceLock<Client> = OnceLock::new();
fn get_client() -> Client {
CLIENT
.get_or_init(|| {
Client::builder()
.timeout(Duration::from_secs(5))
.build()
.unwrap()
})
.clone()
}
#[derive(Serialize)]
struct UplinkPayload {
#[serde(rename = "correlationID")]
@ -88,7 +102,6 @@ struct Location {
}
pub struct Integration {
timeout: Duration,
endpoint: String,
}
@ -96,7 +109,6 @@ impl Integration {
pub fn new(conf: &MyDevicesConfiguration) -> Integration {
trace!("Initializing myDevices integration");
Integration {
timeout: Duration::from_secs(5),
endpoint: conf.endpoint.clone(),
}
}
@ -120,11 +132,10 @@ impl IntegrationTrait for Integration {
let pl = UplinkPayload::from_uplink_event(pl);
let b = serde_json::to_string(&pl)?;
let client = Client::builder().timeout(self.timeout).build()?;
let mut headers = HeaderMap::new();
headers.insert(CONTENT_TYPE, "application/json".parse().unwrap());
let req = client
let req = get_client()
.post(&self.endpoint)
.body(b)
.headers(headers)
@ -204,7 +215,6 @@ pub mod test {
let server = MockServer::start();
let i = Integration {
timeout: Duration::from_secs(5),
endpoint: server.url("/"),
};

20
chirpstack/src/integration/pilot_things.rs
View File

@ -1,4 +1,5 @@
use std::collections::HashMap;
use std::sync::OnceLock;
use std::time::Duration;
use anyhow::Result;
@ -12,8 +13,20 @@ use super::Integration as IntegrationTrait;
use crate::storage::application::PilotThingsConfiguration;
use chirpstack_api::integration;
static CLIENT: OnceLock<Client> = OnceLock::new();
fn get_client() -> Client {
CLIENT
.get_or_init(|| {
Client::builder()
.timeout(Duration::from_secs(5))
.build()
.unwrap()
})
.clone()
}
pub struct Integration {
timeout: Duration,
server: String,
token: String,
}
@ -23,7 +36,6 @@ impl Integration {
trace!("Initializing Pilot Things integration");
Integration {
timeout: Duration::from_secs(5),
server: conf.server.clone(),
token: conf.token.clone(),
}
@ -45,11 +57,10 @@ impl IntegrationTrait for Integration {
let pl = UplinkPayload::from_uplink_event(pl);
let b = serde_json::to_string(&pl)?;
let client = Client::builder().timeout(self.timeout).build()?;
let mut headers = HeaderMap::new();
headers.insert(CONTENT_TYPE, "application/json".parse().unwrap());
let res = client
let res = get_client()
.post(endpoint)
.body(b)
.query(&[("token", self.token.clone())])
@ -182,7 +193,6 @@ pub mod test {
let server = MockServer::start();
let i = Integration {
timeout: Duration::from_secs(5),
server: server.url(""),
token: "foo-token".into(),
};

23
chirpstack/src/integration/thingsboard.rs
View File

@ -1,4 +1,5 @@
use std::collections::{BTreeMap, HashMap};
use std::sync::OnceLock;
use std::time::Duration;
use anyhow::Result;
@ -12,9 +13,21 @@ use super::Integration as IntegrationTrait;
use crate::storage::application::ThingsBoardConfiguration;
use chirpstack_api::integration;
static CLIENT: OnceLock<Client> = OnceLock::new();
fn get_client() -> Client {
CLIENT
.get_or_init(|| {
Client::builder()
.timeout(Duration::from_secs(5))
.build()
.unwrap()
})
.clone()
}
pub struct Integration {
server: String,
timeout: Duration,
}
impl Integration {
@ -22,7 +35,6 @@ impl Integration {
trace!("Initializing ThingsBoard integration");
Integration {
timeout: Duration::from_secs(5),
server: conf.server.clone(),
}
}
@ -39,11 +51,10 @@ impl Integration {
let endpoint = format!("{}/api/v1/{}/attributes", self.server, access_token);
let b = serde_json::to_string(&attributes)?;
let client = Client::builder().timeout(self.timeout).build()?;
let mut headers = HeaderMap::new();
headers.insert(CONTENT_TYPE, "application/json".parse().unwrap());
let res = client
let res = get_client()
.post(endpoint)
.body(b)
.headers(headers)
@ -65,11 +76,10 @@ impl Integration {
let endpoint = format!("{}/api/v1/{}/telemetry", self.server, access_token);
let b = serde_json::to_string(&telemetry)?;
let client = Client::builder().timeout(self.timeout).build()?;
let mut headers = HeaderMap::new();
headers.insert(CONTENT_TYPE, "application/json".parse().unwrap());
let res = client
let res = get_client()
.post(endpoint)
.body(b)
.headers(headers)
@ -323,7 +333,6 @@ pub mod test {
let i = Integration {
server: server.url(""),
timeout: Duration::from_secs(5),
};
let mut vars: HashMap<String, String> = HashMap::new();