From e66a33746826b309dbae7e4da31ad2161e70b839 Mon Sep 17 00:00:00 2001
From: Jordan Petridis <jordanpetridis@protonmail.com>
Date: Fri, 17 Nov 2017 20:01:59 +0200
Subject: [PATCH] Moved indexing episode into a NewEpisode method.

There is a performance reggresion caused by the introduction of
Arc<Mutex<Connection>> instead of Connection that should be refactored.

Also removed the db transcaction as it was incomplete.
---
 hammond-data/src/index_feed.rs         | 43 ++++++--------------------
 hammond-data/src/models/insertables.rs | 32 ++++++++++++++++++-
 2 files changed, 41 insertions(+), 34 deletions(-)

diff --git a/hammond-data/src/index_feed.rs b/hammond-data/src/index_feed.rs
index e8d2b61..949ff5b 100644
--- a/hammond-data/src/index_feed.rs
+++ b/hammond-data/src/index_feed.rs
@@ -1,7 +1,6 @@
 use diesel::prelude::*;
 use rayon::prelude::*;
 
-use diesel;
 use rss;
 
 use dbqueries;
@@ -45,48 +44,26 @@ impl Feed {
         pd.into_podcast(db)
     }
 
+    // TODO: Figure out transcactions.
+    // The synchronous version where there was a db.lock() before the episodes.iter()
+    // is actually faster.
     fn index_channel_items(&self, db: &Database, pd: &Podcast) -> Result<()> {
         let it = self.channel.items();
         let episodes: Vec<_> = it.par_iter()
             .map(|x| feedparser::parse_episode(x, pd.id()))
             .collect();
 
-        let conn = db.lock().unwrap();
-        let e = conn.transaction::<(), Error, _>(|| {
-            // TODO:
-            episodes.iter().for_each(|x| {
-                let e = index_episode(&conn, x);
-                if let Err(err) = e {
-                    error!("Failed to index episode: {:?}.", x);
-                    error!("Error msg: {}", err);
-                };
-            });
-            Ok(())
+        episodes.into_par_iter().for_each(|x| {
+            let e = x.index(&Arc::clone(db));
+            if let Err(err) = e {
+                error!("Failed to index episode: {:?}.", x);
+                error!("Error msg: {}", err);
+            };
         });
-        drop(conn);
-
-        e
+        Ok(())
     }
 }
 
-// TODO: Currently using diesel from master git.
-// Watch out for v0.99.0 beta and change the toml.
-fn index_episode(con: &SqliteConnection, ep: &NewEpisode) -> QueryResult<()> {
-    use schema::episode::dsl::*;
-
-    match dbqueries::get_episode_from_uri(con, ep.uri.unwrap()) {
-        Ok(foo) => if foo.title() != ep.title
-            || foo.published_date() != ep.published_date.as_ref().map(|x| x.as_str())
-        {
-            diesel::replace_into(episode).values(ep).execute(con)?;
-        },
-        Err(_) => {
-            diesel::insert_into(episode).values(ep).execute(con)?;
-        }
-    }
-    Ok(())
-}
-
 pub fn full_index_loop(db: &Database) -> Result<()> {
     let mut f = fetch_all_feeds(db)?;
 
diff --git a/hammond-data/src/models/insertables.rs b/hammond-data/src/models/insertables.rs
index 03d5b25..e292930 100644
--- a/hammond-data/src/models/insertables.rs
+++ b/hammond-data/src/models/insertables.rs
@@ -59,6 +59,36 @@ pub struct NewEpisode<'a> {
     pub podcast_id: i32,
 }
 
+impl<'a> NewEpisode<'a> {
+    // TODO: Currently using diesel from master git.
+    // Watch out for v0.99.0 beta and change the toml.
+    // TODO: Refactor into batch indexes instead.
+    pub fn index(&self, db: &Database) -> QueryResult<()> {
+        use schema::episode::dsl::*;
+
+        let ep = {
+            let tempdb = db.lock().unwrap();
+            dbqueries::get_episode_from_uri(&tempdb, self.uri.unwrap())
+        };
+
+        match ep {
+            Ok(foo) => if foo.title() != self.title
+                || foo.published_date() != self.published_date.as_ref().map(|x| x.as_str())
+            {
+                let tempdb = db.lock().unwrap();
+                diesel::replace_into(episode)
+                    .values(self)
+                    .execute(&*tempdb)?;
+            },
+            Err(_) => {
+                let tempdb = db.lock().unwrap();
+                diesel::insert_into(episode).values(self).execute(&*tempdb)?;
+            }
+        }
+        Ok(())
+    }
+}
+
 #[derive(Insertable)]
 #[table_name = "podcast"]
 #[derive(Debug, Clone)]
@@ -78,7 +108,7 @@ impl NewPodcast {
         Ok(dbqueries::get_podcast_from_title(&tempdb, &self.title)?)
     }
 
-    fn index(&self, db: &Database) -> Result<()> {
+    fn index(&self, db: &Database) -> QueryResult<()> {
         use schema::podcast::dsl::*;
         let pd = {
             let tempdb = db.lock().unwrap();