Iterable run method

examples/main.rs

@@ -9,3 +9,11 @@ fn main() {
     let mut conn = Connection::open_in_memory().unwrap();
     embedded::migrations::runner().run(&mut conn).unwrap();
 }
+
+fn iter_main() {
+    let mut conn = Connection::open_in_memory().unwrap();
+    let runner = embedded::migrations::runner();
+    for migration in runner.run_stepwise(&mut conn) {
+        info!("Got a migration: {}", migration.expect("migration failed!"));
+    }
+}

refinery/tests/rusqlite.rs

@@ -106,6 +106,34 @@ mod rusqlite {
         assert_eq!(migrations[3].checksum(), applied_migrations[3].checksum());
     }
 
+    #[test]
+    fn report_contains_applied_migrations_stepwise() {
+        let mut conn = Connection::open_in_memory().unwrap();
+        let applied_migrations = embedded::migrations::runner()
+            .run_stepwise(&mut conn)
+            .collect::<Result<Vec<_>, _>>()
+            .unwrap();
+
+        let migrations = get_migrations();
+
+        assert_eq!(4, applied_migrations.len());
+
+        assert_eq!(migrations[0].version(), applied_migrations[0].version());
+        assert_eq!(migrations[1].version(), applied_migrations[1].version());
+        assert_eq!(migrations[2].version(), applied_migrations[2].version());
+        assert_eq!(migrations[3].version(), applied_migrations[3].version());
+
+        assert_eq!(migrations[0].name(), migrations[0].name());
+        assert_eq!(migrations[1].name(), applied_migrations[1].name());
+        assert_eq!(migrations[2].name(), applied_migrations[2].name());
+        assert_eq!(migrations[3].name(), applied_migrations[3].name());
+
+        assert_eq!(migrations[0].checksum(), applied_migrations[0].checksum());
+        assert_eq!(migrations[1].checksum(), applied_migrations[1].checksum());
+        assert_eq!(migrations[2].checksum(), applied_migrations[2].checksum());
+        assert_eq!(migrations[3].checksum(), applied_migrations[3].checksum());
+    }
+
     #[test]
     fn creates_migration_table() {
         let mut conn = Connection::open_in_memory().unwrap();
@@ -123,6 +151,26 @@ mod rusqlite {
         assert_eq!(DEFAULT_TABLE_NAME, table_name);
     }
 
+    #[test]
+    fn creates_migration_table_stepwise() {
+        let mut conn = Connection::open_in_memory().unwrap();
+        embedded::migrations::runner()
+            .run_stepwise(&mut conn)
+            .collect::<Result<Vec<_>, _>>()
+            .unwrap();
+        let table_name: String = conn
+            .query_row(
+                &format!(
+                    "SELECT name FROM sqlite_master WHERE type='table' AND name='{}'",
+                    DEFAULT_TABLE_NAME
+                ),
+                [],
+                |row| row.get(0),
+            )
+            .unwrap();
+        assert_eq!(DEFAULT_TABLE_NAME, table_name);
+    }
+
     #[test]
     fn creates_migration_table_grouped_transaction() {
         let mut conn = Connection::open_in_memory().unwrap();
@@ -163,6 +211,29 @@ mod rusqlite {
         assert_eq!("New York", city);
     }
 
+    #[test]
+    fn applies_migration_stepwise() {
+        let mut conn = Connection::open_in_memory().unwrap();
+
+        embedded::migrations::runner()
+            .run_stepwise(&mut conn)
+            .collect::<Result<Vec<_>, _>>()
+            .unwrap();
+
+        conn.execute(
+            "INSERT INTO persons (name, city) VALUES (?, ?)",
+            &["John Legend", "New York"],
+        )
+        .unwrap();
+        let (name, city): (String, String) = conn
+            .query_row("SELECT name, city FROM persons", [], |row| {
+                Ok((row.get(0).unwrap(), row.get(1).unwrap()))
+            })
+            .unwrap();
+        assert_eq!("John Legend", name);
+        assert_eq!("New York", city);
+    }
+
     #[test]
     fn applies_migration_grouped_transaction() {
         let mut conn = Connection::open_in_memory().unwrap();
@@ -205,6 +276,28 @@ mod rusqlite {
         );
     }
 
+    #[test]
+    fn updates_schema_history_stepwise() {
+        let mut conn = Connection::open_in_memory().unwrap();
+
+        embedded::migrations::runner()
+            .run_stepwise(&mut conn)
+            .collect::<Result<Vec<_>, _>>()
+            .unwrap();
+
+        let current = conn
+            .get_last_applied_migration(DEFAULT_TABLE_NAME)
+            .unwrap()
+            .unwrap();
+
+        assert_eq!(4, current.version());
+
+        assert_eq!(
+            OffsetDateTime::now_utc().date(),
+            current.applied_on().unwrap().date()
+        );
+    }
+
     #[test]
     fn updates_schema_history_grouped_transaction() {
         let mut conn = Connection::open_in_memory().unwrap();
@@ -259,6 +352,44 @@ mod rusqlite {
         assert_eq!(2959965718684201605, applied_migrations[0].checksum());
         assert_eq!(8238603820526370208, applied_migrations[1].checksum());
     }
+    #[test]
+
+    fn updates_to_last_working_if_stepwise() {
+        let mut conn = Connection::open_in_memory().unwrap();
+
+        let result: Result<Vec<_>, _> = broken::migrations::runner()
+            .run_stepwise(&mut conn)
+            .collect();
+
+        assert!(result.is_err());
+        let current = conn
+            .get_last_applied_migration(DEFAULT_TABLE_NAME)
+            .unwrap()
+            .unwrap();
+
+        let err = result.unwrap_err();
+        let migrations = get_migrations();
+        let applied_migrations = broken::migrations::runner()
+            .get_applied_migrations(&mut conn)
+            .unwrap();
+
+        assert_eq!(
+            OffsetDateTime::now_utc().date(),
+            current.applied_on().unwrap().date()
+        );
+        assert_eq!(2, current.version());
+        assert!(err.report().unwrap().applied_migrations().is_empty());
+        assert_eq!(2, applied_migrations.len());
+
+        assert_eq!(1, applied_migrations[0].version());
+        assert_eq!(2, applied_migrations[1].version());
+
+        assert_eq!("initial", migrations[0].name());
+        assert_eq!("add_cars_table", applied_migrations[1].name());
+
+        assert_eq!(2959965718684201605, applied_migrations[0].checksum());
+        assert_eq!(8238603820526370208, applied_migrations[1].checksum());
+    }
 
     #[test]
     fn doesnt_update_to_last_working_if_grouped() {
@@ -366,6 +497,40 @@ mod rusqlite {
         assert_eq!(migrations[2].checksum(), applied_migrations[2].checksum());
     }
 
+    #[test]
+    fn migrates_to_target_migration_stepwise() {
+        let mut conn = Connection::open_in_memory().unwrap();
+
+        let applied_migrations = embedded::migrations::runner()
+            .set_target(Target::Version(3))
+            .run_stepwise(&mut conn)
+            .collect::<Result<Vec<_>, _>>()
+            .unwrap();
+
+        let current = conn
+            .get_last_applied_migration(DEFAULT_TABLE_NAME)
+            .unwrap()
+            .unwrap();
+
+        let migrations = get_migrations();
+
+        assert_eq!(3, current.version());
+
+        assert_eq!(3, applied_migrations.len());
+
+        assert_eq!(migrations[0].version(), applied_migrations[0].version());
+        assert_eq!(migrations[1].version(), applied_migrations[1].version());
+        assert_eq!(migrations[2].version(), applied_migrations[2].version());
+
+        assert_eq!(migrations[0].name(), migrations[0].name());
+        assert_eq!(migrations[1].name(), applied_migrations[1].name());
+        assert_eq!(migrations[2].name(), applied_migrations[2].name());
+
+        assert_eq!(migrations[0].checksum(), applied_migrations[0].checksum());
+        assert_eq!(migrations[1].checksum(), applied_migrations[1].checksum());
+        assert_eq!(migrations[2].checksum(), applied_migrations[2].checksum());
+    }
+
     #[test]
     fn migrates_to_target_migration_grouped() {
         let mut conn = Connection::open_in_memory().unwrap();

refinery_core/src/runner.rs

@@ -2,12 +2,14 @@ use regex::Regex;
 use siphasher::sip::SipHasher13;
 use time::OffsetDateTime;
 
+use log::error;
 use std::cmp::Ordering;
+use std::collections::VecDeque;
 use std::fmt;
 use std::hash::{Hash, Hasher};
 
 use crate::error::Kind;
-use crate::traits::DEFAULT_MIGRATION_TABLE_NAME;
+use crate::traits::{sync::migrate as sync_migrate, DEFAULT_MIGRATION_TABLE_NAME};
 use crate::{AsyncMigrate, Error, Migrate};
 use std::fmt::Formatter;
 
@@ -366,13 +368,26 @@ impl Runner {
         self
     }
 
+    /// Creates an iterator over pending migrations, applying each before returning
+    /// the result from `next()`. If a migration fails, the iterator will return that
+    /// result and further calls to `next()` will return `None`.
+    pub fn run_stepwise<C>(
+        self,
+        connection: &mut C,
+    ) -> impl Iterator<Item = Result<Migration, Error>> + '_
+    where
+        C: Migrate,
+    {
+        RunIterator::new(self, connection)
+    }
+
     /// Runs the Migrations in the supplied database connection
-    pub fn run<C>(&self, conn: &'_ mut C) -> Result<Report, Error>
+    pub fn run<C>(&self, connection: &mut C) -> Result<Report, Error>
     where
         C: Migrate,
     {
         Migrate::migrate(
-            conn,
+            connection,
             &self.migrations,
             self.abort_divergent,
             self.abort_missing,
@@ -383,12 +398,12 @@ impl Runner {
     }
 
     /// Runs the Migrations asynchronously in the supplied database connection
-    pub async fn run_async<C>(&self, conn: &mut C) -> Result<Report, Error>
+    pub async fn run_async<C>(&self, connection: &mut C) -> Result<Report, Error>
     where
         C: AsyncMigrate + Send,
     {
         AsyncMigrate::migrate(
-            conn,
+            connection,
             &self.migrations,
             self.abort_divergent,
             self.abort_missing,
@@ -399,3 +414,80 @@ impl Runner {
         .await
     }
 }
+
+/*impl<C> IntoIterator for Runner
+where
+    C: Migrate,
+{
+    type Item = Result<Migration, Error>;
+    type IntoIter = RunIterator;
+
+    fn into_iter(self) -> Self::IntoIter {
+        Runner::into_iter(self)
+    }
+}*/
+
+pub struct RunIterator<'a, C> {
+    connection: &'a mut C,
+    target: Target,
+    migration_table_name: String,
+    items: VecDeque<Migration>,
+    failed: bool,
+}
+impl<'a, C> RunIterator<'a, C>
+where
+    C: Migrate,
+{
+    pub(crate) fn new(runner: Runner, connection: &'a mut C) -> RunIterator<'a, C> {
+        RunIterator {
+            items: VecDeque::from(
+                Migrate::get_unapplied_migrations(
+                    connection,
+                    &runner.migrations,
+                    runner.abort_divergent,
+                    runner.abort_missing,
+                    &runner.migration_table_name,
+                )
+                .unwrap(),
+            ),
+            connection,
+            target: runner.target,
+            migration_table_name: runner.migration_table_name.clone(),
+            failed: false,
+        }
+    }
+}
+impl<C> Iterator for RunIterator<'_, C>
+where
+    C: Migrate,
+{
+    type Item = Result<Migration, Error>;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        match self.failed {
+            true => None,
+            false => self.items.pop_front().and_then(|migration| {
+                sync_migrate(
+                    self.connection,
+                    vec![migration],
+                    self.target,
+                    &self.migration_table_name,
+                )
+                .map(|r| r.applied_migrations.first().cloned())
+                .map_err(|e| {
+                    error!("migration failed: {e:?}");
+                    self.failed = true;
+                    e
+                })
+                .transpose()
+            }),
+        }
+    }
+}
+
+fn flatten<T>(o: Option<Option<T>>) -> Option<T> {
+    match o {
+        Some(inner) => inner,
+        None => None,
+    }
+}