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

Initial switch to using r2d2.

Browse Source
This commit is contained in:
Jordan Petridis
2017-11-19 22:06:10 +02:00
parent 052988a2b1
commit 55442529a8
9 changed files with 175 additions and 149 deletions
Download Patch File Download Diff File Expand all files Collapse all files
hammond-data/src/dbqueries.rs
+16 -8
View File
@@ -8,22 +8,29 @@ use chrono::prelude::*;
/// Random db querries helper functions.
/// Probably needs cleanup.
pub fn get_sources(con: &SqliteConnection) -> QueryResult<Vec<Source>> {
use POOL;
pub fn get_sources() -> QueryResult<Vec<Source>> {
use schema::source::dsl::*;
source.load::<Source>(con)
let con = POOL.get().unwrap();
let s = source.load::<Source>(&*con);
// s.iter().for_each(|x| println!("{:#?}", x));
s
}
pub fn get_podcasts(con: &SqliteConnection) -> QueryResult<Vec<Podcast>> {
pub fn get_podcasts() -> QueryResult<Vec<Podcast>> {
use schema::podcast::dsl::*;
podcast.load::<Podcast>(con)
let con = POOL.get().unwrap();
podcast.load::<Podcast>(&*con)
}
pub fn get_episodes(con: &SqliteConnection) -> QueryResult<Vec<Episode>> {
pub fn get_episodes() -> QueryResult<Vec<Episode>> {
use schema::episode::dsl::*;
episode.order(epoch.desc()).load::<Episode>(con)
let con = POOL.get().unwrap();
episode.order(epoch.desc()).load::<Episode>(&*con)
}
pub fn get_downloaded_episodes(con: &SqliteConnection) -> QueryResult<Vec<Episode>> {
@@ -104,10 +111,11 @@ pub fn get_pd_episodes_limit(
.load::<Episode>(con)
}
pub fn get_source_from_uri(con: &SqliteConnection, uri_: &str) -> QueryResult<Source> {
pub fn get_source_from_uri(uri_: &str) -> QueryResult<Source> {
use schema::source::dsl::*;
source.filter(uri.eq(uri_)).get_result::<Source>(con)
let con = POOL.get().unwrap();
source.filter(uri.eq(uri_)).get_result::<Source>(&*con)
}
pub fn get_podcast_from_title(con: &SqliteConnection, title_: &str) -> QueryResult<Podcast> {
hammond-data/src/feed.rs
+33 -76
View File
@@ -6,13 +6,11 @@ use rss;
use dbqueries;
use parser;
use Database;
use POOL;
use models::{Podcast, Source};
use errors::*;
use std::sync::Arc;
#[derive(Debug)]
pub struct Feed {
@@ -21,42 +19,42 @@ pub struct Feed {
}
impl Feed {
pub fn new_from_source(db: &Database, s: Source) -> Result<Feed> {
s.refresh(db)
pub fn from_source(s: Source) -> Result<Feed> {
s.refresh()
}
pub fn new_from_channel_source(chan: rss::Channel, s: Source) -> Feed {
pub fn from_channel_source(chan: rss::Channel, s: Source) -> Feed {
Feed {
channel: chan,
source: s,
}
}
fn index(&self, db: &Database) -> Result<()> {
let pd = self.index_channel(db)?;
fn index(&self) -> Result<()> {
let pd = self.index_channel()?;
self.index_channel_items(db, &pd)?;
self.index_channel_items(&pd)?;
Ok(())
}
fn index_channel(&self, db: &Database) -> Result<Podcast> {
fn index_channel(&self) -> Result<Podcast> {
let pd = parser::new_podcast(&self.channel, *self.source.id());
// Convert NewPodcast to Podcast
pd.into_podcast(db)
pd.into_podcast()
}
// TODO: Refactor transcactions and find a way to do it in parallel.
fn index_channel_items(&self, db: &Database, pd: &Podcast) -> Result<()> {
fn index_channel_items(&self, pd: &Podcast) -> Result<()> {
let items = self.channel.items();
let episodes: Vec<_> = items
.into_par_iter()
.map(|item| parser::new_episode(item, *pd.id()))
.collect();
let tempdb = db.lock().unwrap();
let tempdb = POOL.clone().get().unwrap();
let _ = tempdb.transaction::<(), Error, _>(|| {
episodes.into_iter().for_each(|x| {
let e = x.index(&tempdb);
let e = x.index(&*tempdb);
if let Err(err) = e {
error!("Failed to index episode: {:?}.", x);
error!("Error msg: {}", err);
@@ -68,17 +66,17 @@ impl Feed {
}
}
pub fn index_all(db: &Database) -> Result<()> {
let mut f = fetch_all(db)?;
pub fn index_all() -> Result<()> {
let mut f = fetch_all()?;
index(db, &mut f);
index(&mut f);
info!("Indexing done.");
Ok(())
}
pub fn index(db: &Database, feeds: &mut [Feed]) {
pub fn index(feeds: &mut [Feed]) {
feeds.into_par_iter().for_each(|f| {
let e = f.index(&Arc::clone(db));
let e = f.index();
if e.is_err() {
error!("Error While trying to update the database.");
error!("Error msg: {}", e.unwrap_err());
@@ -86,22 +84,19 @@ pub fn index(db: &Database, feeds: &mut [Feed]) {
});
}
pub fn fetch_all(db: &Database) -> Result<Vec<Feed>> {
let feeds = {
let conn = db.lock().unwrap();
dbqueries::get_sources(&conn)?
};
pub fn fetch_all() -> Result<Vec<Feed>> {
let feeds = dbqueries::get_sources()?;
let results = fetch(db, feeds);
let results = fetch(feeds);
Ok(results)
}
pub fn fetch(db: &Database, feeds: Vec<Source>) -> Vec<Feed> {
pub fn fetch(feeds: Vec<Source>) -> Vec<Feed> {
let results: Vec<_> = feeds
.into_par_iter()
.filter_map(|x| {
let uri = x.uri().to_owned();
let l = Feed::new_from_source(&Arc::clone(db), x);
let l = Feed::from_source(x);
if l.is_ok() {
l.ok()
} else {
@@ -118,46 +113,17 @@ pub fn fetch(db: &Database, feeds: Vec<Source>) -> Vec<Feed> {
#[cfg(test)]
mod tests {
extern crate rand;
extern crate tempdir;
use diesel::prelude::*;
use rss;
use self::rand::Rng;
use models::NewSource;
use utils::run_migration_on;
use std::io::BufReader;
use std::path::PathBuf;
use std::fs;
use std::sync::Mutex;
use std::io::BufReader;
use super::*;
struct TempDB(tempdir::TempDir, PathBuf, SqliteConnection);
/// Create and return a Temporary DB.
/// Will be destroed once the returned variable(s) is dropped.
fn get_temp_db() -> TempDB {
let mut rng = rand::thread_rng();
let tmp_dir = tempdir::TempDir::new("hammond_unit_test").unwrap();
let db_path = tmp_dir
.path()
.join(format!("hammonddb_{}.db", rng.gen::<usize>()));
let db = SqliteConnection::establish(db_path.to_str().unwrap()).unwrap();
run_migration_on(&db).unwrap();
TempDB(tmp_dir, db_path, db)
}
#[test]
/// Insert feeds and update/index them.
fn test_index_loop() {
let TempDB(_tmp_dir, _db_path, db) = get_temp_db();
let db = Arc::new(Mutex::new(db));
let inpt = vec![
"https://request-for-explanation.github.io/podcast/rss.xml",
"https://feeds.feedburner.com/InterceptedWithJeremyScahill",
@@ -166,23 +132,17 @@ mod tests {
];
inpt.iter().for_each(|feed| {
NewSource::new_with_uri(feed)
.into_source(&db.clone())
.unwrap();
NewSource::new_with_uri(feed).into_source().unwrap();
});
index_all(&db).unwrap();
index_all().unwrap();
// Run again to cover Unique constrains erros.
index_all(&db).unwrap();
index_all().unwrap();
}
#[test]
fn test_complete_index() {
let TempDB(_tmp_dir, _db_path, db) = get_temp_db();
// complete_index runs in parallel so it requires a mutex as argument.
let m = Arc::new(Mutex::new(db));
// vec of (path, url) tuples.
let urls = vec![
(
@@ -195,7 +155,7 @@ mod tests {
),
(
"tests/feeds/TheBreakthrough.xml",
"http://feeds.feedburner.com/propublica/podcast",
"http://feeds.propublica.org/propublica/podcast",
),
(
"tests/feeds/R4Explanation.xml",
@@ -206,25 +166,22 @@ mod tests {
let mut feeds: Vec<_> = urls.iter()
.map(|&(path, url)| {
// Create and insert a Source into db
let s = NewSource::new_with_uri(url)
.into_source(&m.clone())
.unwrap();
let s = NewSource::new_with_uri(url).into_source().unwrap();
// open the xml file
let feed = fs::File::open(path).unwrap();
// parse it into a channel
let chan = rss::Channel::read_from(BufReader::new(feed)).unwrap();
Feed::new_from_channel_source(chan, s)
Feed::from_channel_source(chan, s)
})
.collect();
// Index the channels
index(&m, &mut feeds);
index(&mut feeds);
// Assert the index rows equal the controlled results
let tempdb = m.lock().unwrap();
assert_eq!(dbqueries::get_sources(&tempdb).unwrap().len(), 4);
assert_eq!(dbqueries::get_podcasts(&tempdb).unwrap().len(), 4);
assert_eq!(dbqueries::get_episodes(&tempdb).unwrap().len(), 274);
assert_eq!(dbqueries::get_sources().unwrap().len(), 4);
assert_eq!(dbqueries::get_podcasts().unwrap().len(), 4);
assert_eq!(dbqueries::get_episodes().unwrap().len(), 274);
}
}
hammond-data/src/lib.rs
+44 -4
View File
@@ -16,6 +16,8 @@ extern crate diesel;
extern crate diesel_codegen;
extern crate chrono;
extern crate r2d2;
extern crate r2d2_diesel;
extern crate rayon;
extern crate reqwest;
extern crate rfc822_sanitizer;
@@ -55,11 +57,49 @@ lazy_static!{
HAMMOND_XDG.create_cache_directory(HAMMOND_XDG.get_cache_home()).unwrap()
};
static ref DB_PATH: PathBuf = {
HAMMOND_XDG.place_data_file("hammond.db").unwrap()
};
pub static ref DL_DIR: PathBuf = {
HAMMOND_XDG.create_data_directory("Downloads").unwrap()
};
pub static ref DB_PATH: PathBuf = HAMMOND_XDG.place_data_file("hammond.db").unwrap();
}
#[cfg(not(test))]
lazy_static! {
pub static ref POOL: utils::Pool = utils::init_pool(DB_PATH.to_str().unwrap());
}
#[cfg(test)]
lazy_static! {
static ref TEMPDB: TempDB = get_temp_db();
pub static ref POOL: &'static utils::Pool = &TEMPDB.2;
}
#[cfg(test)]
struct TempDB(tempdir::TempDir, PathBuf, utils::Pool);
#[cfg(test)]
extern crate rand;
#[cfg(test)]
extern crate tempdir;
#[cfg(test)]
use rand::Rng;
#[cfg(test)]
/// Create and return a Temporary DB.
/// Will be destroed once the returned variable(s) is dropped.
fn get_temp_db() -> TempDB {
let mut rng = rand::thread_rng();
let tmp_dir = tempdir::TempDir::new("hammond_unit_test").unwrap();
let db_path = tmp_dir
.path()
.join(format!("hammonddb_{}.db", rng.gen::<usize>()));
let pool = utils::init_pool(db_path.to_str().unwrap());
let db = pool.get().unwrap();
utils::run_migration_on(&db).unwrap();
TempDB(tmp_dir, db_path, pool)
}
hammond-data/src/models/insertables.rs
+15 -16
View File
@@ -3,7 +3,7 @@ use diesel;
use schema::{episode, podcast, source};
use models::{Podcast, Source};
use Database;
use POOL;
use errors::*;
use dbqueries;
@@ -26,21 +26,20 @@ impl<'a> NewSource<'a> {
}
}
fn index(&self, db: &Database) {
fn index(&self) {
use schema::source::dsl::*;
let tempdb = db.lock().unwrap();
let tempdb = POOL.clone().get().unwrap();
// Throw away the result like `insert or ignore`
// Diesel deos not support `insert or ignore` yet.
let _ = diesel::insert_into(source).values(self).execute(&*tempdb);
}
// Look out for when tryinto lands into stable.
pub fn into_source(self, db: &Database) -> QueryResult<Source> {
self.index(db);
pub fn into_source(self) -> QueryResult<Source> {
self.index();
let tempdb = db.lock().unwrap();
dbqueries::get_source_from_uri(&tempdb, self.uri)
dbqueries::get_source_from_uri(self.uri)
}
}
@@ -104,28 +103,28 @@ pub struct NewPodcast {
impl NewPodcast {
// Look out for when tryinto lands into stable.
pub fn into_podcast(self, db: &Database) -> Result<Podcast> {
self.index(db)?;
let tempdb = db.lock().unwrap();
Ok(dbqueries::get_podcast_from_title(&tempdb, &self.title)?)
pub fn into_podcast(self) -> Result<Podcast> {
self.index()?;
let tempdb = POOL.clone().get().unwrap();
Ok(dbqueries::get_podcast_from_title(&*tempdb, &self.title)?)
}
fn index(&self, db: &Database) -> QueryResult<()> {
fn index(&self) -> QueryResult<()> {
use schema::podcast::dsl::*;
let pd = {
let tempdb = db.lock().unwrap();
dbqueries::get_podcast_from_title(&tempdb, &self.title)
let tempdb = POOL.clone().get().unwrap();
dbqueries::get_podcast_from_title(&*tempdb, &self.title)
};
match pd {
Ok(foo) => if foo.link() != self.link {
let tempdb = db.lock().unwrap();
let tempdb = POOL.clone().get().unwrap();
diesel::replace_into(podcast)
.values(self)
.execute(&*tempdb)?;
},
Err(_) => {
let tempdb = db.lock().unwrap();
let tempdb = POOL.clone().get().unwrap();
diesel::insert_into(podcast).values(self).execute(&*tempdb)?;
}
}
hammond-data/src/models/queryables.rs
+8 -7
View File
@@ -11,6 +11,7 @@ use errors::*;
use models::insertables::NewPodcast;
use Database;
use POOL;
use std::io::Read;
use std::str::FromStr;
@@ -265,7 +266,7 @@ impl<'a> Source {
/// Extract Etag and LastModifier from req, and update self and the
/// corresponding db row.
fn update_etag(&mut self, db: &Database, req: &reqwest::Response) -> Result<()> {
fn update_etag(&mut self, req: &reqwest::Response) -> Result<()> {
let headers = req.headers();
// let etag = headers.get_raw("ETag").unwrap();
@@ -278,18 +279,18 @@ impl<'a> Source {
{
self.http_etag = etag.map(|x| x.tag().to_string().to_owned());
self.last_modified = lmod.map(|x| format!("{}", x));
self.save(db)?;
self.save()?;
}
Ok(())
}
pub fn save(&self, db: &Database) -> QueryResult<Source> {
let tempdb = db.lock().unwrap();
pub fn save(&self) -> QueryResult<Source> {
let tempdb = POOL.clone().get().unwrap();
self.save_changes::<Source>(&*tempdb)
}
pub fn refresh(mut self, db: &Database) -> Result<Feed> {
pub fn refresh(mut self) -> Result<Feed> {
use reqwest::header::{ETag, EntityTag, Headers, HttpDate, LastModified};
let mut headers = Headers::new();
@@ -322,12 +323,12 @@ impl<'a> Source {
// _ => (),
// };
self.update_etag(db, &req)?;
self.update_etag(&req)?;
let mut buf = String::new();
req.read_to_string(&mut buf)?;
let chan = Channel::from_str(&buf)?;
Ok(Feed::new_from_channel_source(chan, self))
Ok(Feed::from_channel_source(chan, self))
}
}
hammond-data/src/utils.rs
+12
View File
@@ -2,6 +2,10 @@ use rayon::prelude::*;
use diesel::prelude::*;
use chrono::prelude::*;
use r2d2;
use diesel::sqlite::SqliteConnection;
use r2d2_diesel::ConnectionManager;
use errors::*;
use dbqueries;
use Database;
@@ -15,11 +19,19 @@ use DB_PATH;
embed_migrations!("migrations/");
pub type Pool = r2d2::Pool<ConnectionManager<SqliteConnection>>;
pub fn init() -> Result<()> {
let conn = establish_connection();
run_migration_on(&conn)
}
pub fn init_pool(db_path: &str) -> Pool {
let config = r2d2::Config::default();
let manager = ConnectionManager::<SqliteConnection>::new(db_path);
r2d2::Pool::new(config, manager).expect("Failed to create pool.")
}
pub fn run_migration_on(connection: &SqliteConnection) -> Result<()> {
info!("Running DB Migrations...");
embedded_migrations::run(connection)?;