Getting a stream of events from github.

Cargo.lock

@@ -311,6 +311,7 @@ name = "github_watcher"
 version = "0.1.0"
 dependencies = [
  "reqwest",
+ "serde_json",
  "sqlx",
  "tokio",
 ]
@@ -958,9 +959,9 @@ checksum = "ce31e24b01e1e524df96f1c2fdd054405f8d7376249a5110886fb4b658484789"
 
 [[package]]
 name = "serde_json"
-version = "1.0.79"
+version = "1.0.81"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8e8d9fa5c3b304765ce1fd9c4c8a3de2c8db365a5b91be52f186efc675681d95"
+checksum = "9b7ce2b32a1aed03c558dc61a5cd328f15aff2dbc17daad8fb8af04d2100e15c"
 dependencies = [
  "itoa",
  "ryu",

Cargo.toml

@@ -5,5 +5,6 @@ edition = "2021"
 
 [dependencies]
 reqwest = { version = "0.11.10", features = ["json"] }
+serde_json = "1.0.81"
 sqlx = { version = "0.5", features = [ "runtime-tokio-rustls", "sqlite", "migrate" ] }
 tokio = { version = "1.16.1", features = ["full"] }

src/githubctl/github_endpoint_watcher.rs

@@ -0,0 +1,79 @@
+use reqwest::header::HeaderValue;
+use std::time::{SystemTime, UNIX_EPOCH};
+use tokio::time::{sleep, Duration};
+
+pub struct GithubEndpointWatcher {
+    username: String,
+    token: String,
+    url: String,
+    http_client: reqwest::Client,
+    etag: Option<HeaderValue>,
+    next_poll_allowed: u64,
+}
+
+impl GithubEndpointWatcher {
+    pub fn new(
+        username: String,
+        token: String,
+        url: String,
+    ) -> Result<Self, Box<dyn std::error::Error>> {
+        let http_client = reqwest::Client::new();
+        Ok(GithubEndpointWatcher {
+            username,
+            token,
+            url,
+            http_client,
+            etag: None,
+            next_poll_allowed: 0,
+        })
+    }
+
+    pub async fn get_results(&mut self) -> Result<serde_json::Value, Box<dyn std::error::Error>> {
+        let mut request = self
+            .http_client
+            .get(&self.url)
+            .basic_auth(self.username.trim(), Some(self.token.trim()))
+            .header("User-Agent", "github_watcher")
+            .header("Accept", "application/vnd.github.v3+json");
+        if let Some(etag) = &self.etag {
+            request = request.header(reqwest::header::ETAG, etag);
+        }
+        self.sleep_until_next_poll().await?;
+        let request_started_at = SystemTime::now().duration_since(UNIX_EPOCH)?.as_secs();
+        println!("Hitting url {}", self.url);
+        let response = request.send().await?;
+
+        let headers = response.headers();
+        let etag = headers.get(reqwest::header::ETAG).map(|x| x.to_owned());
+        let poll_interval = headers.get("x-poll-interval").map(|x| x.to_owned());
+        // let ratelimit_limit = headers.get("x-ratelimit-limit").map(|x| x.to_owned());
+        // let ratelimit_remaining = headers.get("x-ratelimit-remaining").map(|x| x.to_owned());
+        // let ratelimit_reset = headers.get("x-ratelimit-reset").map(|x| x.to_owned());
+        // let ratelimit_used = headers.get("x-ratelimit-used").map(|x| x.to_owned());
+        // let ratelimit_resource = headers.get("x-ratelimit-resource").map(|x| x.to_owned());
+
+        let body = response.json::<serde_json::Value>().await?;
+        println!("{}", serde_json::to_string(&body)?);
+        self.etag = etag;
+        let poll_interval_parsed: u64 = match poll_interval {
+            Some(header) => header.to_str()?.parse()?,
+            None => {
+                println!("No poll interval returned, defaulting to 5 minute polling.");
+                300
+            }
+        };
+        self.next_poll_allowed = request_started_at + poll_interval_parsed;
+        Ok(body)
+    }
+
+    async fn sleep_until_next_poll(&self) -> Result<(), Box<dyn std::error::Error>> {
+        loop {
+            let now = SystemTime::now().duration_since(UNIX_EPOCH)?.as_secs();
+            if now >= self.next_poll_allowed {
+                break;
+            }
+            sleep(Duration::from_secs(self.next_poll_allowed - now)).await;
+        }
+        Ok(())
+    }
+}

src/githubctl/githubctl.rs

@@ -1,18 +1,37 @@
+use reqwest::header::HeaderValue;
+use reqwest::header::ETAG;
 use std::collections::HashMap;
+use tokio::sync::mpsc;
+
+use super::github_endpoint_watcher::GithubEndpointWatcher;
 
 pub struct GithubCtl<'a> {
     username: &'a str,
     token: &'a str,
-    http_client: reqwest::Client,
+    event_writer: mpsc::Sender<serde_json::Value>,
+    event_reader: mpsc::Receiver<serde_json::Value>,
 }
 
 impl<'a> GithubCtl<'a> {
     pub fn new(username: &'a str, token: &'a str) -> Result<Self, Box<dyn std::error::Error>> {
-        let http_client = reqwest::Client::new();
-        Ok(GithubCtl { username, token, http_client })
+        let (tx, rx) = mpsc::channel::<serde_json::Value>(100);
+        Ok(GithubCtl {
+            username,
+            token,
+            event_writer: tx,
+            event_reader: rx,
+        })
+    }
+
+    pub async fn get_event(&mut self) -> Result<serde_json::Value, Box<dyn std::error::Error>> {
+        if let Some(val) = self.event_reader.recv().await {
+            return Ok(val);
+        } else {
+            return Err("No more events, perhaps all writers hung up?".into());
+        }
     }
 
-    pub async fn get_events<O, R>(
+    pub async fn watch_repo<O, R>(
         &mut self,
         owner: O,
         repo: R,
@@ -21,19 +40,38 @@ impl<'a> GithubCtl<'a> {
         O: AsRef<str>,
         R: AsRef<str>,
     {
-        // /repos/{owner}/{repo}/events
-        // -H "Accept: application/vnd.github.v3+json"
-        //   https://api.github.com/orgs/ORG/events
-        // -u username:token
-        // https://docs.github.com/en/authentication/keeping-your-account-and-data-secure/creating-a-personal-access-token
-        let url = format!("https://api.github.com/repos/{owner}/{repo}/events", owner=owner.as_ref(), repo=repo.as_ref());
-        let resp = self.http_client.get(url).basic_auth(self.username, Some(self.token)).send().await?;
-        // let body = resp.json::<HashMap<String, String>>().await?;
-        // let resp = reqwest::get(url)
-        //     .await?
-        //     .json::<HashMap<String, String>>()
-        //     .await?;
-        println!("{:#?}", body);
+        let url = format!(
+            "https://api.github.com/repos/{owner}/{repo}/events",
+            owner = owner.as_ref().trim(),
+            repo = repo.as_ref().trim()
+        );
+        let username = self.username.to_string();
+        let token = self.token.to_string();
+        let event_stream = self.event_writer.clone();
+
+        tokio::spawn(async move {
+            let mut endpoint_watcher = GithubEndpointWatcher::new(username, token, url)
+                .expect("Failed to create endpoint watcher.");
+            loop {
+                let api_result = match endpoint_watcher.get_results().await {
+                    Ok(result) => result,
+                    Err(_) => {
+                        println!("Failed to get results.");
+                        return;
+                    }
+                };
+                if let serde_json::Value::Array(events) = api_result {
+                    for event in events {
+                        if let Err(_) = event_stream.send(event).await {
+                            println!("Receiver dropped.");
+                            return;
+                        }
+                    }
+                } else {
+                    println!("Unsupported JSON type.");
+                }
+            }
+        });
         Ok(())
     }
 }

src/githubctl/mod.rs

@@ -1,3 +1,4 @@
+mod github_endpoint_watcher;
 mod githubctl;
 
 pub use githubctl::GithubCtl;

src/main.rs

@@ -1,5 +1,7 @@
 mod githubctl;
 
+use serde_json;
+
 const USERNAME: &'static str = include_str!("../.username");
 const TOKEN: &'static str = include_str!("../.pat");
 const ORG: &'static str = include_str!("../.org");
@@ -8,7 +10,9 @@ const REPO: &'static str = include_str!("../.repo");
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
     let mut github = githubctl::GithubCtl::new(USERNAME, TOKEN)?;
-    github.get_events(ORG, REPO).await?;
-    println!("done.");
-    Ok(())
+    github.watch_repo(ORG, REPO).await?;
+    loop {
+        let event = github.get_event().await?;
+        println!("{}", serde_json::to_string(&event)?);
+    }
 }