Getting somewhere?

This commit is contained in:
Jordan Petridis 2018-01-14 06:37:47 +02:00
parent a5fd79e220
commit 1031315cdd
No known key found for this signature in database
GPG Key ID: CEABAD9F5683B9A6
8 changed files with 88 additions and 7 deletions

1
Cargo.lock generated
View File

@ -542,6 +542,7 @@ dependencies = [
"itertools 0.7.6 (registry+https://github.com/rust-lang/crates.io-index)",
"lazy_static 1.0.0 (registry+https://github.com/rust-lang/crates.io-index)",
"log 0.3.9 (registry+https://github.com/rust-lang/crates.io-index)",
"native-tls 0.1.4 (registry+https://github.com/rust-lang/crates.io-index)",
"r2d2 0.8.2 (registry+https://github.com/rust-lang/crates.io-index)",
"r2d2-diesel 1.0.0 (registry+https://github.com/rust-lang/crates.io-index)",
"rand 0.4.2 (registry+https://github.com/rust-lang/crates.io-index)",

View File

@ -25,6 +25,7 @@ futures = "0.1.17"
hyper = "0.11.12"
tokio-core = "0.1.12"
hyper-tls = "0.1.2"
native-tls = "0.1.4"
[dependencies.diesel]
features = ["sqlite"]

View File

@ -4,6 +4,7 @@ use rss;
use reqwest;
use r2d2;
use hyper;
use native_tls;
use std::io;
@ -15,6 +16,7 @@ error_chain! {
RSSError(rss::Error);
ReqError(reqwest::Error);
HyperError(hyper::Error);
TLSError(native_tls::Error);
IoError(io::Error);
}
}

View File

@ -36,7 +36,8 @@ impl Feed {
}
}
fn index(&self) -> Result<()> {
/// docs
pub fn index(&self) -> Result<()> {
let pd = self.get_podcast()?;
self.index_channel_items(&pd)
}

View File

@ -46,6 +46,7 @@ extern crate futures;
extern crate hyper;
extern crate hyper_tls;
extern crate itertools;
extern crate native_tls;
extern crate r2d2;
extern crate r2d2_diesel;
extern crate rayon;
@ -66,7 +67,7 @@ pub mod database;
pub(crate) mod models;
mod parser;
mod schema;
mod pipeline;
pub mod pipeline;
pub use models::queryables::{Episode, EpisodeWidgetQuery, Podcast, PodcastCoverQuery, Source};

View File

@ -16,6 +16,7 @@ use hyper_tls::HttpsConnector;
// use hyper::header::{ETag, LastModified};
use futures::prelude::*;
// use futures::future::ok;
use schema::{episode, podcast, source};
use feed::Feed;

View File

@ -1 +1,66 @@
//! Docs.
use futures::future::*;
use errors::Error;
use Source;
use tokio_core::reactor::Core;
use hyper::Client;
use hyper_tls::HttpsConnector;
// use futures::future::*;
// Weird magic from #rust irc channel
// kudos to remexre
fn collect_futures<F>(
futures: Vec<F>,
) -> Box<Future<Item = Vec<Result<F::Item, F::Error>>, Error = Error>>
where
F: 'static + Future,
<F as Future>::Item: 'static,
<F as Future>::Error: 'static,
{
Box::new(loop_fn((futures, vec![]), |(futures, mut done)| {
select_all(futures).then(|r| {
let (r, rest) = match r {
Ok((r, _, rest)) => (Ok(r), rest),
Err((r, _, rest)) => (Err(r), rest),
};
done.push(r);
if rest.len() == 0 {
Ok(Loop::Break(done))
} else {
Ok(Loop::Continue((rest, done)))
}
})
}))
}
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::*;
/// Docs
pub fn pipeline<S: IntoIterator<Item = Source>>(sources: S) -> 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()
.map(|s| s.into_fututre_feed(&client, false).map(|feed| feed.index()))
.collect();
let f = core.run(collect_futures(list))?;
f.into_iter()
.filter_map(|x| x.err())
.for_each(|err| error!("Error: {}", err));
Ok(())
}
}

View File

@ -1,7 +1,9 @@
use send_cell::SendCell;
use gdk_pixbuf::Pixbuf;
use hammond_data::feed;
// use hammond_data::feed;
use hammond_data::pipeline;
use hammond_data::dbqueries;
use hammond_data::{PodcastCoverQuery, Source};
use hammond_downloader::downloader;
@ -21,10 +23,17 @@ pub fn refresh_feed(headerbar: Arc<Header>, source: Option<Vec<Source>>, sender:
thread::spawn(move || {
if let Some(s) = source {
feed::index_loop(s);
} else if let Err(err) = feed::index_all() {
// feed::index_loop(s);
if let Err(err) = pipeline::pipeline(s) {
error!("Error While trying to update the database.");
error!("Error msg: {}", err);
}
} else {
let sources = dbqueries::get_sources().unwrap();
if let Err(err) = pipeline::pipeline(sources) {
error!("Error While trying to update the database.");
error!("Error msg: {}", err);
}
};
sender.send(Action::HeaderBarHideUpdateIndicator).unwrap();