rss-funnel 0.0.5

A composable feed processing pipeline
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133

use std::time::Duration;

use serde::{Deserialize, Serialize};
use url::Url;

use crate::client::{Client, ClientConfig};
use crate::feed::Feed;
use crate::util::Result;

use super::{FeedFilter, FeedFilterConfig, FilterConfig, Filters};

#[derive(Serialize, Deserialize, Clone, Debug)]
#[serde(untagged)]
pub enum MergeConfig {
  Simple(MergeSimpleConfig),
  Full(MergeFullConfig),
}

#[derive(Serialize, Deserialize, Clone, Debug)]
#[serde(transparent)]
pub struct MergeSimpleConfig {
  source: String,
}

#[derive(Serialize, Deserialize, Clone, Debug)]
pub struct MergeFullConfig {
  source: String,
  #[serde(default)]
  client: ClientConfig,
  #[serde(default)]
  filters: Vec<FilterConfig>,
}

impl From<MergeSimpleConfig> for MergeFullConfig {
  fn from(config: MergeSimpleConfig) -> Self {
    Self {
      source: config.source,
      client: ClientConfig::default(),
      filters: Default::default(),
    }
  }
}

impl From<MergeConfig> for MergeFullConfig {
  fn from(config: MergeConfig) -> Self {
    match config {
      MergeConfig::Simple(config) => config.into(),
      MergeConfig::Full(config) => config,
    }
  }
}

#[async_trait::async_trait]
impl FeedFilterConfig for MergeConfig {
  type Filter = Merge;

  async fn build(self) -> Result<Self::Filter> {
    let MergeFullConfig {
      client,
      filters,
      source,
    } = self.into();
    let client = client.build(Duration::from_secs(15 * 60))?;
    let filters = Filters::from_config(filters).await?;
    let source = Url::parse(&source)?;

    Ok(Merge {
      client,
      source,
      filters,
    })
  }
}

pub struct Merge {
  client: Client,
  source: Url,
  filters: Filters,
}

#[async_trait::async_trait]
impl FeedFilter for Merge {
  async fn run(&self, feed: &mut Feed) -> Result<()> {
    let mut new_feed = self.client.fetch_feed(&self.source).await?;
    self.filters.process(&mut new_feed).await?;
    feed.merge(new_feed)?;
    Ok(())
  }
}

#[cfg(test)]
mod test {
  use crate::test_utils::fetch_endpoint;
  use std::collections::HashMap;

  #[tokio::test]
  async fn test_merge_filter() {
    let config = r#"
      !endpoint
      path: /feed.xml
      source: fixture:///scishow.xml
      filters:
        - merge:
            source: fixture:///scishow.xml
            filters:
              - js: |
                  function modify_post(feed, post) {
                    post.title.value += " (modified)";
                    return post;
                  }
    "#;

    let mut feed = fetch_endpoint(config, "").await;
    let posts = feed.take_posts();

    // First group posts by url. Then assert, in each group, one title
    // is "modified" of another
    let mut groups: HashMap<String, Vec<String>> = HashMap::new();
    for post in posts {
      let link = post.link().unwrap().into();
      let title = post.title().unwrap().into();
      groups.entry(link).or_default().push(title);
    }

    for (_, titles) in groups {
      assert_eq!(titles.len(), 2);
      assert!(
        titles[0] == format!("{} (modified)", titles[1])
          || titles[1] == format!("{} (modified)", titles[0])
      );
    }
  }
}