Migrated project into separate directories

weather-service/Cargo.toml

@@ -0,0 +1,26 @@
+[package]
+name = "weather-service"
+version = "0.1.0"
+edition = "2021"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+actix-web = "4.4.0"
+actix-rt = "2.9.0"
+chrono = { version = "0.4.28", features = ["serde"] }
+dotenv = "0.15.0"
+diesel = { version = "1.4", features = ["postgres", "r2d2", "uuid", "chrono"] }
+diesel_migrations = "1.4"
+env_logger = "0.10.0"
+lazy_static = "1.4.0"
+listenfd = "1.0.1"
+quick-xml = { version = "0.30.0", features = ["serialize"] }
+r2d2 = "0.8.10"
+reqwest = "0.11.20"
+serde = {version = "1.0.188", features = ["derive"]}
+serde_json = "1.0.105"
+tokio = { version = "1.32.0", features = ["macros", "rt"] }
+uuid = { version = "1.4.1", features = ["serde", "v4"] }
+log = "0.4.20"
+actix-cors = "0.6.4"

weather-service/Makefile

@@ -0,0 +1,29 @@
+#!make
+
+include .env
+
+SHELL := /bin/bash
+
+.PHONY: help build start stop lint
+
+help: ## This info
+	@echo
+	@cat Makefile | grep -E '^[a-zA-Z\/_-]+:.*?## .*$$' | awk 'BEGIN {FS = ":.*?## "}; {printf "\033[36m%-30s\033[0m %s\n", $$1, $$2}'
+	@echo
+
+up:
+	docker compose up -d
+
+down:
+	docker compose down
+
+connect:
+	docker exec -it aviation_weather_db psql -U postgres
+
+lint:  ## Run the linter
+	npm run lint
+
+clean-db:  ## Remove database and Cargo packages
+	docker exec -i ${DATABASE_HOST} sh -c 'PGPASSWORD=${DATABASE_PASSWORD} psql -U ${DATABASE_USER} -d postgres -c "DROP DATABASE IF EXISTS \"${DATABASE_NAME}\";"'  || true
+
+	

weather-service/diesel.toml

@@ -0,0 +1,5 @@
+# For documentation on how to configure this file,
+# see diesel.rs/guides/configuring-diesel-cli
+
+[print_schema]
+file = "src/schema.rs"

weather-service/docker-compose.yml

@@ -0,0 +1,22 @@
+version: '3'
+
+services:
+  db:
+    image: postgres:latest
+    container_name: weather.db
+    env_file:
+      - .env
+    environment:
+      POSTGRES_USER: ${DATABASE_USER}
+      POSTGRES_PASSWORD: ${DATABASE_PASSWORD}
+      POSTGRES_DB: ${DATABASE_NAME}
+    volumes:
+     - db_weather:/var/lib/postgresql/data
+     - db_logs_weather:/var/log
+    ports:
+      - "5433:5432"
+    restart: unless-stopped
+
+volumes:
+  db_weather:
+  db_logs_weather:

weather-service/migrations/create_airports/down.sql

@@ -0,0 +1 @@
+DROP TABLE airports;

weather-service/migrations/create_airports/up.sql

@@ -0,0 +1,7 @@
+CREATE TABLE IF NOT EXISTS airports (
+    id SERIAL PRIMARY KEY,
+    full_name TEXT NOT NULL,
+    icao TEXT NOT NULL,
+    latitude DOUBLE PRECISION NOT NULL,
+    longitude DOUBLE PRECISION NOT NULL
+)

weather-service/migrations/create_metars/down.sql

@@ -0,0 +1 @@
+DROP TABLE metars;

weather-service/migrations/create_metars/up.sql

@@ -0,0 +1,24 @@
+CREATE TABLE IF NOT EXISTS metars (
+    id SERIAL PRIMARY KEY,
+    icao TEXT NOT NULL,
+    raw_text TEXT NOT NULL,
+    station_id TEXT NOT NULL,
+    observation_time TEXT NOT NULL,
+    latitude INTEGER NOT NULL,
+    longitude INTEGER NOT NULL,
+    temp_c DOUBLE PRECISION NOT NULL,
+    dewpoint_c DOUBLE PRECISION NOT NULL,
+    wind_dir_degrees INTEGER NOT NULL,
+    wind_speed_kt INTEGER NOT NULL,
+    visibility_statute_mi TEXT NOT NULL,
+    altim_in_hg DOUBLE PRECISION NOT NULL,
+    sea_level_pressure_mb DOUBLE PRECISION,
+    wx_string TEXT,
+    flight_category TEXT,
+    three_hr_pressure_tendency_mb DOUBLE PRECISION,
+    metar_type TEXT,
+    max_t_c DOUBLE PRECISION,
+    min_t_c DOUBLE PRECISION,
+    precip_in DOUBLE PRECISION,
+    elevation_m INTEGER
+)

weather-service/migrations/diesel_initial_setup/down.sql

@@ -0,0 +1,2 @@
+DROP FUNCTION IF EXISTS diesel_manage_updated_at(_tbl regclass);
+DROP FUNCTION IF EXISTS diesel_set_updated_at();

weather-service/migrations/diesel_initial_setup/up.sql

@@ -0,0 +1,18 @@
+CREATE OR REPLACE FUNCTION diesel_manage_updated_at(_tbl regclass) RETURNS VOID AS $$
+BEGIN
+    EXECUTE format('CREATE TRIGGER set_updated_at BEFORE UPDATE ON %s
+                    FOR EACH ROW EXECUTE PROCEDURE diesel_set_updated_at()', _tbl);
+END;
+$$ LANGUAGE plpgsql;
+
+CREATE OR REPLACE FUNCTION diesel_set_updated_at() RETURNS trigger AS $$
+BEGIN
+    IF (
+        NEW IS DISTINCT FROM OLD AND
+        NEW.updated_at IS NOT DISTINCT FROM OLD.updated_at
+    ) THEN
+        NEW.updated_at := current_timestamp;
+    END IF;
+    RETURN NEW;
+END;
+$$ LANGUAGE plpgsql;

weather-service/src/airports/mod.rs

@@ -0,0 +1,5 @@
+mod model;
+mod routes;
+
+pub use model::*;
+pub use routes::init_routes;

weather-service/src/airports/model.rs

@@ -0,0 +1,61 @@
+use crate::db;
+use crate::error_handler::CustomError;
+use crate::schema::airports;
+use diesel::prelude::*;
+use serde::{Deserialize, Serialize};
+
+#[derive(Serialize, Deserialize, AsChangeset, Insertable)]
+#[table_name = "airports"]
+pub struct Airport {
+  pub full_name: String,
+  pub icao: String,
+  pub latitude: f64,
+  pub longitude: f64,
+}
+
+#[derive(Serialize, Deserialize, Queryable)]
+pub struct Airports {
+  pub id: i32,
+  pub full_name: String,
+  pub icao: String,
+  pub latitude: f64,
+  pub longitude: f64,
+}
+
+impl Airports {
+  pub fn find_all() -> Result<Vec<Self>, CustomError> {
+    let conn = db::connection()?;
+    let airports = airports::table.load::<Airports>(&conn)?;
+    Ok(airports)
+  }
+
+  pub fn find(id: i32) -> Result<Self, CustomError> {
+    let conn = db::connection()?;
+    let airport = airports::table.filter(airports::id.eq(id)).first(&conn)?;
+    Ok(airport)
+}
+
+pub fn create(airport: Airport) -> Result<Self, CustomError> {
+    let conn = db::connection()?;
+    let airport = Airport::from(airport);
+    let airport = diesel::insert_into(airports::table)
+        .values(airport)
+        .get_result(&conn)?;
+    Ok(airport)
+}
+
+pub fn update(id: i32, airport: Airport) -> Result<Self, CustomError> {
+    let conn = db::connection()?;
+    let airport = diesel::update(airports::table)
+        .filter(airports::id.eq(id))
+        .set(airport)
+        .get_result(&conn)?;
+    Ok(airport)
+}
+
+pub fn delete(id: i32) -> Result<usize, CustomError> {
+    let conn = db::connection()?;
+    let res = diesel::delete(airports::table.filter(airports::id.eq(id))).execute(&conn)?;
+    Ok(res)
+}
+}

weather-service/src/airports/routes.rs

@@ -0,0 +1,73 @@
+use crate::airports::{Airport, Airports};
+use crate::error_handler::CustomError;
+use actix_web::{delete, get, post, put, web, HttpResponse};
+use log::error;
+use serde_json::json;
+
+#[get("/airports")]
+async fn find_all() -> Result<HttpResponse, CustomError> {
+    let airports = match web::block(|| Airports::find_all()).await.unwrap() {
+        Ok(a) => a,
+        Err(err) => {
+            error!("{}", err);
+            return Err(err);
+        }
+    };
+    Ok(HttpResponse::Ok().json(airports))
+}
+
+#[get("/airports/{id}")]
+async fn find(id: web::Path<i32>) -> Result<HttpResponse, CustomError> {
+    let airport = match Airports::find(id.into_inner()) {
+        Ok(a) => a,
+        Err(err) => {
+            error!("{}", err);
+            return Err(err);
+        }
+    };
+    Ok(HttpResponse::Ok().json(airport))
+}
+
+#[post("/airports")]
+async fn create(airport: web::Json<Airport>) -> Result<HttpResponse, CustomError> {
+    let airport = match Airports::create(airport.into_inner()) {
+        Ok(a) => a,
+        Err(err) => {
+            error!("{}", err);
+            return Err(err);
+        }
+    };
+    Ok(HttpResponse::Ok().json(airport))
+}
+
+#[put("/airports/{id}")]
+async fn update(id: web::Path<i32>, airport: web::Json<Airport>) -> Result<HttpResponse, CustomError> {
+    let airport = match Airports::update(id.into_inner(), airport.into_inner()) {
+        Ok(a) => a,
+        Err(err) => {
+            error!("{}", err);
+            return Err(err);
+        }
+    };
+    Ok(HttpResponse::Ok().json(airport))
+}
+
+#[delete("/airports/{id}")]
+async fn delete(id: web::Path<i32>) -> Result<HttpResponse, CustomError> {
+    let deleted_airport = match Airports::delete(id.into_inner()) {
+        Ok(a) => a,
+        Err(err) => {
+            error!("{}", err);
+            return Err(err);
+        }
+    };
+    Ok(HttpResponse::Ok().json(json!({ "deleted": deleted_airport })))
+}
+
+pub fn init_routes(config: &mut web::ServiceConfig) {
+  config.service(find_all);
+  config.service(find);
+  config.service(create);
+  config.service(update);
+  config.service(delete);
+}

weather-service/src/db.rs

@@ -0,0 +1,37 @@
+use crate::error_handler::CustomError;
+use diesel::pg::PgConnection;
+use diesel::r2d2::ConnectionManager;
+use lazy_static::lazy_static;
+use log::{error, info};
+use r2d2;
+use std::env;
+
+type Pool = r2d2::Pool<ConnectionManager<PgConnection>>;
+pub type DbConnection = r2d2::PooledConnection<ConnectionManager<PgConnection>>;
+
+diesel_migrations::embed_migrations!();
+
+lazy_static! {
+    static ref POOL: Pool = {
+        let username = env::var("DATABASE_USER").expect("Database username is not set");
+        let password = env::var("DATABASE_PASSWORD").expect("Database password is not set");
+        let name = env::var("DATABASE_NAME").expect("Database name is not set");
+        let url = format!("postgres://{}:{}@localhost:5433/{}", username, password, name);
+        let manager = ConnectionManager::<PgConnection>::new(url);
+        Pool::new(manager).expect("Failed to create db pool")
+    };
+}
+
+pub fn init() {
+    lazy_static::initialize(&POOL);
+    let conn = connection().expect("Failed to get db connection");
+    match embedded_migrations::run(&conn) {
+        Ok(_) => info!("Database initialized"),
+        Err(err) => error!("Failed to initialize database; {}", err),
+    };
+}
+
+pub fn connection() -> Result<DbConnection, CustomError> {
+    POOL.get()
+        .map_err(|e| CustomError::new(500, format!("Failed getting db connection: {}", e)))
+}

weather-service/src/error_handler.rs

@@ -0,0 +1,55 @@
+use actix_web::http::StatusCode;
+use actix_web::{HttpResponse, ResponseError};
+use diesel::result::Error as DieselError;
+use serde::{Deserialize, Serialize};
+use serde_json::json;
+use std::fmt;
+
+#[derive(Debug, Deserialize, Serialize)]
+pub struct CustomError {
+    pub error_status_code: u16,
+    pub error_message: String,
+}
+
+impl CustomError {
+    pub fn new(error_status_code: u16, error_message: String) -> CustomError {
+        CustomError {
+            error_status_code,
+            error_message,
+        }
+    }
+}
+
+impl fmt::Display for CustomError {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        f.write_str(self.error_message.as_str())
+    }
+}
+
+impl From<DieselError> for CustomError {
+    fn from(error: DieselError) -> CustomError {
+        match error {
+            DieselError::DatabaseError(_, err) => CustomError::new(409, err.message().to_string()),
+            DieselError::NotFound => {
+                CustomError::new(404, "The airport record was not found".to_string())
+            }
+            err => CustomError::new(500, format!("Unknown Diesel error: {}", err)),
+        }
+    }
+}
+
+impl ResponseError for CustomError {
+    fn error_response(&self) -> HttpResponse {
+        let status_code = match StatusCode::from_u16(self.error_status_code) {
+            Ok(status_code) => status_code,
+            Err(_) => StatusCode::INTERNAL_SERVER_ERROR,
+        };
+
+        let error_message = match status_code.as_u16() < 500 {
+            true => self.error_message.clone(),
+            false => "Internal server error".to_string(),
+        };
+
+        HttpResponse::build(status_code).json(json!({ "message": error_message }))
+    }
+}

weather-service/src/lib.rs.bk

@@ -0,0 +1,162 @@
+use std::error::Error;
+use std::fmt;
+use log::warn;
+use std::io::BufRead;
+use quick_xml::{Reader, events::{Event, BytesStart}, Writer, de::Deserializer};
+use serde::Deserialize;
+
+pub struct Airport {
+    pub name: String,
+    pub icao: String
+}
+
+impl Airport {
+    pub fn new(name: String, icao: String) -> Airport {
+        Airport { name, icao }
+    }
+}
+
+#[derive(Debug)]
+pub struct WeatherError(pub String);
+
+impl fmt::Display for WeatherError {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        write!(f, "{}", self.0)
+    }
+}
+
+impl Error for WeatherError {}
+
+#[derive(Deserialize, Debug)]
+pub struct Metar {
+    pub raw_text: String,
+    pub station_id: String,
+    pub observation_time: String,
+    pub latitude: f32,
+    pub longitude: f32,
+    pub temp_c: f32,
+    pub dewpoint_c: f32,
+    pub wind_dir_degrees: i32,
+    pub wind_speed_kt: i32,
+    pub visibility_statute_mi: String,
+    pub altim_in_hg: f32,
+    pub sea_level_pressure_mb: Option<f32>,
+    pub quality_control_flags: Option<QualityControlFlags>,
+    pub wx_string: Option<String>,
+    // pub sky_con dition: Option<Vec<String>>, // TODO work on attributes
+    pub flight_category: String,
+    pub three_hr_pressure_tendency_mb: Option<f32>,
+    pub metar_type: String,
+    #[serde(rename = "maxT_c")]
+    pub max_t_c: Option<f32>,
+    #[serde(rename = "minT_c")]
+    pub min_t_c: Option<f32>,
+    pub precip_in: Option<f32>,
+    pub elevation_m: i32
+}
+
+#[derive(Deserialize, Debug)]
+pub struct QualityControlFlags {
+    pub auto: Option<bool>,
+    pub auto_station: Option<bool>
+}
+
+impl Metar {
+    pub fn parse(input: String) -> Result<Vec<Metar>, WeatherError> {
+        if input.is_empty() {
+            return Err(WeatherError("Input is empty".to_string()))
+        }
+
+        let mut reader = Reader::from_str(&input);
+        let mut buf = Vec::new();
+        let mut junk_buf: Vec<u8> = Vec::new();
+
+        loop {
+            match reader.read_event_into(&mut buf) {
+                Err(e) => panic!("Error at position: {}: {:?}", reader.buffer_position(), e),
+                Ok(Event::Eof) => break,
+                Ok(Event::Start(e)) => {
+                    match e.name().as_ref() {
+                        b"METAR" => {
+                            let metar_bytes = Metar::read_to_end_into_buffer(&mut reader, &e, &mut junk_buf).unwrap();
+                            let str = std::str::from_utf8(&metar_bytes).unwrap();
+                            let mut deserializer = Deserializer::from_str(str);
+                            let metar = Metar::deserialize(&mut deserializer).unwrap();
+                            println!("{:#?}", metar);
+                        },
+                        _ => ()
+                    }
+                },
+                _ => ()
+            }
+        }
+
+        return Ok(vec![])
+    }
+
+    // https://capnfabs.net/posts/parsing-huge-xml-quickxml-rust-serde/
+    pub fn read_to_end_into_buffer<R: BufRead>(reader: &mut Reader<R>, start_tag: &BytesStart, junk_buf: &mut Vec<u8>) -> Result<Vec<u8>, quick_xml::Error> {
+        let mut depth = 0;
+        let mut output_buf: Vec<u8> = Vec::new();
+        let mut w = Writer::new(&mut output_buf);
+        let tag_name = start_tag.name();
+        w.write_event(Event::Start(start_tag.clone()))?;
+        loop {
+            junk_buf.clear();
+            let event = reader.read_event_into(junk_buf)?;
+            w.write_event(&event)?;
+      
+            match event {
+                Event::Start(e) if e.name() == tag_name => depth += 1,
+                Event::End(e) if e.name() == tag_name => {
+                    if depth == 0 {
+                        return Ok(output_buf);
+                    }
+                    depth -= 1;
+                }
+                Event::Eof => {
+                    panic!("EOF")
+                }
+                _ => {}
+            }
+        }
+    }
+}
+
+pub struct Weather {
+    pub base_url: String
+}
+
+impl Weather {
+    pub async fn metar(&mut self, airports: Vec<Airport>) -> Vec<Metar> {
+        let mut station_icaos: Vec<&str> = vec![];
+        for station in airports.iter() {
+            station_icaos.push(&station.icao);
+        }
+        let station_string = station_icaos.join(",");
+        let url = format!("{}/metar.php?ids={}&format=xml", self.base_url, station_string);
+        
+        let metars: Vec<Metar> = match reqwest::get(url).await {
+            Ok(r) => match r.text().await {
+                Ok(r) => {
+                    match Metar::parse(r) {
+                        Ok(m) => m,
+                        Err(err) => {
+                            warn!("{}", err);
+                            vec![]
+                        }
+                    }
+                },
+                Err(err) => {
+                    warn!("Unable to parse METAR request: {}", err);
+                    vec![]
+                }
+            },
+            Err(err) => {
+                warn!("Unable to get METAR request: {}", err);
+                vec![]
+            }
+        };
+        return metars;
+    }
+}

weather-service/src/main.rs

@@ -0,0 +1,51 @@
+#[macro_use]
+extern crate diesel;
+#[macro_use]
+extern crate diesel_migrations;
+
+use actix_cors::Cors;
+use actix_web::{App, HttpServer, middleware::Logger};
+use dotenv::dotenv;
+use listenfd::ListenFd;
+use log::debug;
+use std::env;
+
+mod airports;
+mod db;
+mod error_handler;
+mod metars;
+mod schema;
+
+#[actix_rt::main]
+async fn main() -> std::io::Result<()> {
+    dotenv().ok();
+    if std::env::var_os("RUST_LOG").is_none() {
+        std::env::set_var("RUST_LOG", "info,actix=info,diesel_migrations=warn,reqwest=warn,hyper=warn");
+    }
+    env_logger::init();
+    db::init();
+
+    let mut listenfd = ListenFd::from_env();
+    let mut server = HttpServer::new(|| {
+        let cors = Cors::default()
+            .allow_any_origin()
+            .allow_any_method()
+            .allow_any_header();
+        App::new()
+        .configure(airports::init_routes)
+        .configure(metars::init_routes)
+        .wrap(cors)
+        .wrap(Logger::default())
+    });
+
+    server = match listenfd.take_tcp_listener(0)? {
+        Some(listener) => server.listen(listener)?,
+        None => {
+            let host = env::var("HOST").expect("Please set host in .env");
+            let port = env::var("PORT").expect("Please set port in .env");
+            debug!("Binding server to {}:{}", host, port);
+            server.bind(format!("{}:{}", host, port))?
+        }
+    };
+    server.run().await
+}

weather-service/src/metars/mod.rs

@@ -0,0 +1,5 @@
+mod model;
+mod routes;
+
+pub use model::*;
+pub use routes::init_routes;

weather-service/src/metars/model.rs

@@ -0,0 +1,162 @@
+use crate::error_handler::CustomError;
+use crate::schema::metars;
+use log::warn;
+use std::io::BufRead;
+use quick_xml::{Reader, events::{Event, BytesStart}, Writer, de::Deserializer};
+use serde::{Deserialize, Serialize};
+
+#[derive(Serialize, Deserialize)]
+pub struct QualityControlFlags {
+    pub auto: Option<bool>,
+    pub auto_station: Option<bool>
+}
+
+#[derive(Serialize, Deserialize, AsChangeset, Insertable)]
+#[table_name = "metars"]
+pub struct Metar {
+    pub icao: String,
+    pub raw_text: String,
+    pub station_id: String,
+    pub observation_time: String,
+    pub latitude: f64,
+    pub longitude: f64,
+    pub temp_c: f64,
+    pub dewpoint_c: f64,
+    pub wind_dir_degrees: i32,
+    pub wind_speed_kt: i32,
+    pub visibility_statute_mi: String,
+    pub altim_in_hg: f64,
+    pub sea_level_pressure_mb: Option<f64>,
+    // pub quality_control_flags: Option<QualityControlFlags>,
+    pub wx_string: Option<String>,
+    // pub sky_con dition: Option<Vec<String>>, // TODO work on attributes
+    pub flight_category: String,
+    pub three_hr_pressure_tendency_mb: Option<f64>,
+    pub metar_type: String,
+    #[serde(rename = "maxT_c")]
+    pub max_t_c: Option<f64>,
+    #[serde(rename = "minT_c")]
+    pub min_t_c: Option<f64>,
+    pub precip_in: Option<f64>,
+    pub elevation_m: i32
+}
+
+#[derive(Serialize, Deserialize, Queryable)]
+pub struct Metars {
+    // pub id: i32,
+    // pub icao: String,
+    pub raw_text: String,
+    pub station_id: String,
+    pub observation_time: String,
+    pub latitude: f64,
+    pub longitude: f64,
+    pub temp_c: f64,
+    pub dewpoint_c: f64,
+    pub wind_dir_degrees: i32,
+    pub wind_speed_kt: i32,
+    pub visibility_statute_mi: String,
+    pub altim_in_hg: f64,
+    pub sea_level_pressure_mb: Option<f64>,
+    pub quality_control_flags: Option<QualityControlFlags>,
+    pub wx_string: Option<String>,
+    // pub sky_condition: Option<Vec<String>>, // TODO work on attributes
+    pub flight_category: String,
+    pub three_hr_pressure_tendency_mb: Option<f64>,
+    pub metar_type: String,
+    #[serde(rename = "maxT_c")]
+    pub max_t_c: Option<f64>,
+    #[serde(rename = "minT_c")]
+    pub min_t_c: Option<f64>,
+    pub precip_in: Option<f64>,
+    pub elevation_m: i32
+}
+
+impl Metars {
+    pub async fn get_all(icaos: String) -> Result<Vec<Self>, CustomError> {
+        // let station_string = station_icaos.join(",");
+        let url = format!("https://beta.aviationweather.gov/cgi-bin/data/metar.php?ids={}&format=xml", icaos);
+        let metars: Vec<Metars> = match reqwest::get(url).await {
+            Ok(r) => match r.text().await {
+                Ok(r) => {
+                    match Metars::parse(r) {
+                        Ok(m) => m,
+                        Err(err) => {
+                            warn!("{}", err);
+                            vec![]
+                        }
+                    }
+                },
+                Err(err) => {
+                    warn!("Unable to parse METAR request: {}", err);
+                    vec![]
+                }
+            },
+            Err(err) => {
+                warn!("Unable to get METAR request: {}", err);
+                vec![]
+            }
+        };
+        Ok(metars)
+    }
+
+    pub fn parse(input: String) -> Result<Vec<Self>, CustomError> {
+        if input.is_empty() {
+            return Err(CustomError::new(500, "Input is empty".to_string()))
+        }
+
+        let mut reader = Reader::from_str(&input);
+        let mut buf = Vec::new();
+        let mut junk_buf: Vec<u8> = Vec::new();
+        let mut metars: Vec<Self> = vec![];
+
+        loop {
+            match reader.read_event_into(&mut buf) {
+                Err(e) => panic!("Error at position: {}: {:?}", reader.buffer_position(), e),
+                Ok(Event::Eof) => break,
+                Ok(Event::Start(e)) => {
+                    match e.name().as_ref() {
+                        b"METAR" => {
+                            let metar_bytes = Metars::read_to_end_into_buffer(&mut reader, &e, &mut junk_buf).unwrap();
+                            let str = std::str::from_utf8(&metar_bytes).unwrap();
+                            let mut deserializer = Deserializer::from_str(str);
+                            let metar = Metars::deserialize(&mut deserializer).unwrap();
+                            metars.push(metar);
+                        },
+                        _ => ()
+                    }
+                },
+                _ => ()
+            }
+        }
+
+        return Ok(metars)
+    }
+
+    // https://capnfabs.net/posts/parsing-huge-xml-quickxml-rust-serde/
+    pub fn read_to_end_into_buffer<R: BufRead>(reader: &mut Reader<R>, start_tag: &BytesStart, junk_buf: &mut Vec<u8>) -> Result<Vec<u8>, quick_xml::Error> {
+        let mut depth = 0;
+        let mut output_buf: Vec<u8> = Vec::new();
+        let mut w = Writer::new(&mut output_buf);
+        let tag_name = start_tag.name();
+        w.write_event(Event::Start(start_tag.clone()))?;
+        loop {
+            junk_buf.clear();
+            let event = reader.read_event_into(junk_buf)?;
+            w.write_event(&event)?;
+      
+            match event {
+                Event::Start(e) if e.name() == tag_name => depth += 1,
+                Event::End(e) if e.name() == tag_name => {
+                    if depth == 0 {
+                        return Ok(output_buf);
+                    }
+                    depth -= 1;
+                }
+                Event::Eof => {
+                    panic!("EOF")
+                }
+                _ => {}
+            }
+        }
+    }
+}

weather-service/src/metars/routes.rs

@@ -0,0 +1,29 @@
+use crate::error_handler::CustomError;
+use crate::metars::Metars;
+use actix_web::{get, web, HttpResponse, Responder};
+
+// #[get("metars/{ids}")]
+// async fn get_all(ids: web::Path<String>) -> Result<HttpResponse, CustomError> {
+//     let airports = web::block(|| Ok::<_, CustomError>(async {Metars::get_all(ids.into_inner()).await}))
+//         .await
+//         .unwrap()
+//         .unwrap()
+//         .await
+//         .unwrap();
+//     Ok(HttpResponse::Ok().json(airports))
+// }
+
+#[get("metars/{ids}")]
+async fn get_all(ids: web::Path<String>) -> impl Responder {
+    let airports = web::block(|| Ok::<_, CustomError>(async {Metars::get_all(ids.into_inner()).await}))
+        .await
+        .unwrap()
+        .unwrap()
+        .await
+        .unwrap();
+    HttpResponse::Ok().json(airports)
+}
+
+pub fn init_routes(config: &mut web::ServiceConfig) {
+    config.service(get_all);
+}

weather-service/src/schema.rs

@@ -0,0 +1,36 @@
+diesel::table! {
+  airports (id) {
+    id -> Integer,
+    full_name -> Text,
+    icao -> Text,
+    latitude -> Double,
+    longitude -> Double,
+  }
+}
+
+diesel::table! {
+  metars (id) {
+    id -> Integer,
+    icao -> Text,
+    raw_text -> Text,
+    station_id -> Text,
+    observation_time -> Text,
+    latitude -> Double,
+    longitude -> Double,
+    temp_c -> Double,
+    dewpoint_c -> Double,
+    wind_dir_degrees -> Integer,
+    wind_speed_kt -> Integer,
+    visibility_statute_mi -> Text,
+    altim_in_hg -> Double,
+    sea_level_pressure_mb -> Nullable<Double>,
+    wx_string -> Nullable<Text>,
+    flight_category -> Text,
+    three_hr_pressure_tendency_mb -> Nullable<Double>,
+    metar_type -> Text,
+    max_t_c -> Nullable<Double>,
+    min_t_c -> Nullable<Double>,
+    precip_in -> Nullable<Double>,
+    elevation_m -> Integer,
+  }
+}