Second set of commits for realtime support

src/gtfs/realtime.rs

@@ -1,15 +1,14 @@
 /***
  * Implementation of GTFS-R polling 
   */
-
-use std::{str::FromStr, time::Duration};
-
 use reqwest::{blocking::Client, header::{HeaderName, HeaderValue}};
+use std::{collections::HashMap, str::FromStr, time::Duration};
+use serde_json::{Value, de::Read};
 
-use crate::gtfs::structs::{Arrival, Error, Gtfs};
+use crate::gtfs::{self, structs::{Arrival, Error, Gtfs}};
 
 impl Gtfs<'_> {
-    fn realtime_update (&self, arrivals: Vec<Arrival<'_>>) -> Result<bool, Error>{
+    fn realtime_update (&self, arrivals: Vec<Arrival<'_>>) -> Result<(), gtfs::structs::Error>{
 
         // Poll GTFS-R API
         let client = Client::builder()
@@ -17,20 +16,52 @@ impl Gtfs<'_> {
             .connect_timeout(Duration::from_secs(5))
             .build()?;
         
-        let mut response = client
-            .get(self.preferences.realtime_url)
-            .header("x-api-key", self.preferences.realtime_api_key)
+        let response = client
+            .get(&self.preferences.realtime_url)
+            .header("x-api-key", &self.preferences.realtime_api_key)
             .send()?;
 
         // Parse response
-        let response = json::rdr 
+        let realtime_info: gtfs_realtime::FeedMessage  
+            = serde_json::from_reader(response)?;
+
+
+        // Build a map of (trip, stop) -> Arrival for faster lookup
+        let mut lookup: HashMap<(&String, &String), mut Arrival<'_>> = HashMap::new();
+        for arrival in arrivals {
+            lookup.insert((&(arrival.trip.id), &(arrival.stop.id)), arrival);
+        }
 
         // Match deltas to existing arrivals
+        for entity  in realtime_info.entity {
+            if entity.trip_update.is_some() && entity.stop.is_some() {
+                let trip_update = entity.trip_update.unwrap();
+                
+                // Look up the entry in arrivals for the current
+                for stop_update in trip_update.stop_time_update {
+                    if stop_update.stop_id.is_none() {
+                        continue;
+                    } 
+                    let trip_id = trip_update.trip.trip_id.as_ref().unwrap();
+                    let key = (trip_id, &(stop_update.stop_id.unwrap())); 
+                    if lookup.contains_key(&key) {
+                       
+                        // Update the arrival time
+                        let arrival_value = lookup.get(&key);
+                        let arrival: &&mut Arrival<'_> = arrival_value.unwrap();
+                        if let Some(update) = stop_update.departure.or(stop_update.arrival) {
+                            let new_time = (((arrival.departure_time as i64) 
+                                                        + (update.delay.unwrap_or(0) as i64))) as u32;
+                            arrival.departure_time = new_time;
+                        }
+                    }
+                }
+            }
+        }
 
-        // Update the arrival times       
-
+        return Ok(());
+        }
     }
 
 
 
-}