toxblh/podcasts
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Pipeline: remove submodule hack.

Browse Source
This commit is contained in:
Jordan Petridis 2018-01-16 14:37:51 +02:00
parent b3460b15a2
commit bf4f655ed2
No known key found for this signature in database
GPG Key ID: CEABAD9F5683B9A6
1 changed files with 33 additions and 39 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

72
hammond-data/src/pipeline.rs
View File

@ -7,15 +7,46 @@ use hyper_tls::HttpsConnector;
use futures::prelude::*; use futures::prelude::*;
use futures::future::*; use futures::future::*;
use errors::Error; use errors::*;
use Source; use Source;
// use Feed; // use Feed;
use std;
/// The pipline to be run for indexing and updating a Podcast feed that originates from
/// `Source.uri`.
///
/// Messy temp diagram:
/// Source -> GET Request -> Update Etags -> Check Status -> Parse xml/Rss ->
/// Convert rss::Channel into Feed -> Index Podcast -> Index Episodes.
pub fn pipeline<S: IntoIterator<Item = Source>>(sources: S, ignore_etags: bool) -> Result<()> {
let mut core = Core::new()?;
let handle = core.handle();
let client = Client::configure()
// FIXME: numcpus instead of 4
.connector(HttpsConnector::new(4, &handle)?)
.build(&handle);
let list = sources
.into_iter()
// FIXME: Make proper indexing futures instead of wrapping up existing
// blocking functions
.map(|s| s.into_fututre_feed(&client, ignore_etags).map(|feed| feed.index_future()))
.collect();
let f = core.run(collect_futures(list))?;
f.into_iter()
.filter_map(|x| x.err())
.for_each(|err| error!("Error: {}", err));
Ok(())
}
// Weird magic from #rust irc channel // Weird magic from #rust irc channel
// kudos to remexre // kudos to remexre
fn collect_futures<F>( fn collect_futures<F>(
futures: Vec<F>, futures: Vec<F>,
) -> Box<Future<Item = Vec<Result<F::Item, F::Error>>, Error = Error>> ) -> Box<Future<Item = Vec<std::result::Result<F::Item, F::Error>>, Error = Error>>
where where
F: 'static + Future, F: 'static + Future,
<F as Future>::Item: 'static, <F as Future>::Item: 'static,
@ -36,40 +67,3 @@ where
}) })
})) }))
} }
pub use self::dirtyhack::pipeline;
// Use a submodule ot not polute the collect_futures definition with the errorchain Result<T>.
mod dirtyhack {
use super::*;
use errors::*;
/// The pipline to be run for indexing and updating a Podcast feed that originates from
/// `Source.uri`.
///
/// Messy temp diagram:
/// Source -> GET Request -> Update Etags -> Check Status -> Parse xml/Rss ->
/// Convert rss::Channel into Feed -> Index Podcast -> Index Episodes.
pub fn pipeline<S: IntoIterator<Item = Source>>(sources: S, ignore_etags: bool) -> Result<()> {
let mut core = Core::new()?;
let handle = core.handle();
let client = Client::configure()
// FIXME: numcpus instead of 4
.connector(HttpsConnector::new(4, &handle)?)
.build(&handle);
let list = sources
.into_iter()
// FIXME: Make proper indexing futures instead of wrapping up existing
// blocking functions
.map(|s| s.into_fututre_feed(&client, ignore_etags).map(|feed| feed.index_future()))
.collect();
let f = core.run(collect_futures(list))?;
f.into_iter()
.filter_map(|x| x.err())
.for_each(|err| error!("Error: {}", err));
Ok(())
}
}