// Source code for the Substrate Telemetry Server.
// Copyright (C) 2021 Parity Technologies (UK) Ltd.
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU General Public License for more details.
//
// You should have received a copy of the GNU General Public License
// along with this program. If not, see .
use std::net::Ipv4Addr;
use std::sync::Arc;
use futures::{Sink, SinkExt};
use parking_lot::RwLock;
use rustc_hash::FxHashMap;
use serde::Deserialize;
use common::node_types::NodeLocation;
use tokio::sync::Semaphore;
/// The returned location is optional; it may be None if not found.
pub type Location = Option>;
/// This is responsible for taking an IP address and attempting
/// to find a geographical location from this
pub fn find_location(response_chan: R) -> flume::Sender<(Id, Ipv4Addr)>
where
R: Sink<(Id, Option>)> + Unpin + Send + Clone + 'static,
Id: Clone + Send + 'static,
{
let (tx, rx) = flume::unbounded();
// cache entries
let mut cache: FxHashMap>> = FxHashMap::default();
// Default entry for localhost
cache.insert(
Ipv4Addr::new(127, 0, 0, 1),
Some(Arc::new(NodeLocation {
latitude: 52.516_6667,
longitude: 13.4,
city: "Berlin".into(),
})),
);
// Create a locator with our cache. This is used to obtain locations.
let locator = Locator::new(cache);
// Spawn a loop to handle location requests
tokio::spawn(async move {
// Allow 4 requests at a time. acquiring a token will block while the
// number of concurrent location requests is more than this.
let semaphore = Arc::new(Semaphore::new(4));
loop {
while let Ok((id, ip_address)) = rx.recv_async().await {
let permit = semaphore.clone().acquire_owned().await.unwrap();
let mut response_chan = response_chan.clone();
let locator = locator.clone();
// Once we have acquired our permit, spawn a task to avoid
// blocking this loop so that we can handle concurrent requests.
tokio::spawn(async move {
match locator.locate(ip_address).await {
Ok(loc) => {
let _ = response_chan.send((id, loc)).await;
}
Err(e) => {
log::debug!("GET error for ip location: {:?}", e);
}
};
// ensure permit is moved into task by dropping it explicitly:
drop(permit);
});
}
}
});
tx
}
/// This struct can be used to make location requests, given
/// an IPV4 address.
#[derive(Clone)]
struct Locator {
client: reqwest::Client,
cache: Arc>>>>,
}
impl Locator {
pub fn new(cache: FxHashMap>>) -> Self {
let client = reqwest::Client::new();
Locator {
client,
cache: Arc::new(RwLock::new(cache)),
}
}
pub async fn locate(&self, ip: Ipv4Addr) -> Result