feat: impl subscribe feed usecase

Author: ymgyt

Cargo.lock

@@ -3,8 +3,11 @@ resolver = "2"
 members = ["synd", "syndapi", "syndterm"]
 
 [workspace.dependencies]
+async-trait = { version = "0.1.77" }
 anyhow = { version = "1" }
 clap = { version = "4.4" }
+feed-rs = { version = "1.4" }
+futures = { version = "0.3" }
 graphql_client = { version = "0.13.0", default-features = false }
 reqwest = { version = "0.11.23", default-features = false, features = [
   "rustls-tls",

Cargo.toml

@@ -13,7 +13,7 @@ fmt-toml:
   taplo fmt **.toml
 
 update-gql-schema:
-  graphql-client introspect-schema http://localhost:5959/gql \
+  graphql-client introspect-schema http://localhost:5959/graphql \
     --header 'authorization: me' out> syndterm/gql/schema.json
 
 

justfile

@@ -6,3 +6,9 @@ edition = "2021"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
+anyhow = { workspace = true }
+async-trait = { workspace = true }
+feed-rs = { workspace = true }
+futures = { workspace = true }
+reqwest = { workspace = true, features = ["stream"] }
+thiserror = "1.0.56"

synd/Cargo.toml

@@ -0,0 +1 @@
+pub mod parser;

synd/src/feed/mod.rs

@@ -0,0 +1,91 @@
+use std::time::Duration;
+
+use async_trait::async_trait;
+use feed_rs::parser::Parser;
+
+use crate::types::Feed;
+
+pub type ParseResult<T> = std::result::Result<T, ParserError>;
+
+#[derive(Debug, thiserror::Error)]
+pub enum ParserError {
+    #[error("fetch failed")]
+    Fetch(#[from] reqwest::Error),
+    #[error("response size limit exceeded")]
+    ResponseLimitExceed,
+
+    #[error(transparent)]
+    Other(#[from] anyhow::Error),
+}
+
+#[async_trait]
+pub trait FetchFeed: Send + Sync {
+    async fn fetch(&self, url: String) -> ParseResult<Feed>;
+}
+
+/// Feed Process entry point
+pub struct FeedService {
+    http: reqwest::Client,
+    buff_limit: usize,
+}
+
+#[async_trait]
+impl FetchFeed for FeedService {
+    async fn fetch(&self, url: String) -> ParseResult<Feed> {
+        use futures::StreamExt;
+        let mut stream = self
+            .http
+            .get(&url)
+            .send()
+            .await
+            .map_err(ParserError::Fetch)?
+            .error_for_status()
+            .map_err(ParserError::Fetch)?
+            .bytes_stream();
+
+        let mut buff = Vec::new();
+
+        while let Some(chunk) = stream.next().await {
+            let chunk = chunk.map_err(ParserError::Fetch)?;
+            if buff.len() + chunk.len() > self.buff_limit {
+                return Err(ParserError::ResponseLimitExceed);
+            }
+            buff.extend(chunk);
+        }
+
+        self.parse(url, buff.as_slice())
+    }
+}
+
+impl FeedService {
+    pub fn new(user_agent: &str, buff_limit: usize) -> Self {
+        let http = reqwest::ClientBuilder::new()
+            .user_agent(user_agent)
+            .timeout(Duration::from_secs(10))
+            .connect_timeout(Duration::from_secs(10))
+            .build()
+            .unwrap();
+
+        Self { http, buff_limit }
+    }
+
+    pub fn parse<S>(&self, url: impl Into<String>, source: S) -> ParseResult<Feed>
+    where
+        S: std::io::Read,
+    {
+        let url = url.into();
+        let parser = self.build_parser(&url);
+
+        match parser.parse(source) {
+            Ok(feed) => Ok(Feed::from((url, feed))),
+            // TODO: handle error
+            Err(err) => Err(ParserError::Other(err.into())),
+        }
+    }
+
+    fn build_parser(&self, base_uri: impl AsRef<str>) -> Parser {
+        feed_rs::parser::Builder::new()
+            .base_uri(Some(base_uri))
+            .build()
+    }
+}

synd/src/feed/parser.rs

@@ -1,10 +1,2 @@
-#[derive(Debug, Clone)]
-pub struct Feed {
-    pub url: String,
-}
-
-impl Feed {
-    pub fn new(url: String) -> Self {
-        Self { url }
-    }
-}
+pub mod feed;
+pub mod types;

synd/src/lib.rs

@@ -0,0 +1,41 @@
+#[derive(Debug, Clone)]
+pub struct Feed {
+    url: String,
+    #[allow(dead_code)]
+    feed: feed_rs::model::Feed,
+}
+
+impl Feed {
+    pub fn meta(&self) -> FeedMeta {
+        FeedMeta::new(self.title().into(), self.url.clone())
+    }
+
+    pub fn title(&self) -> &str {
+        self.feed
+            .title
+            .as_ref()
+            .map(|text| text.content.as_str())
+            .unwrap_or("???")
+    }
+}
+
+impl From<(String, feed_rs::model::Feed)> for Feed {
+    fn from(feed: (String, feed_rs::model::Feed)) -> Self {
+        Feed {
+            url: feed.0,
+            feed: feed.1,
+        }
+    }
+}
+
+#[derive(Debug, Clone)]
+pub struct FeedMeta {
+    pub title: String,
+    pub url: String,
+}
+
+impl FeedMeta {
+    pub fn new(title: String, url: String) -> Self {
+        Self { title, url }
+    }
+}

synd/src/types.rs

@@ -10,9 +10,11 @@ async-graphql = { git = "https://github.com/async-graphql/async-graphql.git", br
   "tracing",
 ] }
 async-graphql-axum = { git = "https://github.com/async-graphql/async-graphql.git", branch = "master" }
+async-trait = { workspace = true }
 # TODO configure features
 axum = { version = "0.7.3" }
 clap = { workspace = true, features = ["derive"] }
+either = "1.9.0"
 graphql_client = { workspace = true }
 kvsd = { version = "0.1.2" }
 reqwest = { workspace = true }

syndapi/Cargo.toml

@@ -1,2 +1,4 @@
 pub const USER_AGENT: &'static str =
     concat!(env!("CARGO_PKG_NAME"), "/", env!("CARGO_PKG_VERSION"));
+
+pub const PORT: u16 = 5959;