add channel support

driver/driver.go

@@ -3,7 +3,7 @@ package driver
 import (
 	"errors"
 	"fmt"
-	"github.com/mattes/migrate/driver/bash"
+	// "github.com/mattes/migrate/driver/bash"
 	"github.com/mattes/migrate/driver/postgres"
 	"github.com/mattes/migrate/file"
 	neturl "net/url" // alias to allow `url string` func signature in New
@@ -12,7 +12,7 @@ import (
 type Driver interface {
 	Initialize(url string) error
 	FilenameExtension() string
-	Migrate(files file.Files) error
+	Migrate(files file.Files, pipe chan interface{})
 	Version() (uint64, error)
 }
 
@@ -31,13 +31,13 @@ func New(url string) (Driver, error) {
 			return nil, err
 		}
 		return d, nil
-	case "bash":
-		d := &bash.Driver{}
-		verifyFilenameExtension("bash", d)
-		if err := d.Initialize(url); err != nil {
-			return nil, err
-		}
-		return d, nil
+	// case "bash":
+	// 	d := &bash.Driver{}
+	// 	verifyFilenameExtension("bash", d)
+	// 	if err := d.Initialize(url); err != nil {
+	// 		return nil, err
+	// 	}
+	// 	return d, nil
 	default:
 		return nil, errors.New(fmt.Sprintf("Driver '%s' not found.", u.Scheme))
 	}

driver/postgres/postgres.go

@@ -2,9 +2,11 @@ package postgres
 
 import (
 	"database/sql"
+	"fmt"
 	_ "github.com/lib/pq"
 	"github.com/mattes/migrate/file"
 	"github.com/mattes/migrate/migrate/direction"
+	"time"
 )
 
 type Driver struct {
@@ -42,44 +44,52 @@ func (driver *Driver) FilenameExtension() string {
 	return "sql"
 }
 
-func (driver *Driver) Migrate(files file.Files) error {
+func (driver *Driver) Migrate(files file.Files, pipe chan interface{}) {
+	defer close(pipe)
 	for _, f := range files {
 
 		tx, err := driver.db.Begin()
 		if err != nil {
-			return err
+			pipe <- err
+			return
 		}
 
 		if f.Direction == direction.Up {
 			if _, err := tx.Exec(`INSERT INTO `+tableName+` (version) VALUES ($1)`, f.Version); err != nil {
+				pipe <- err
 				if err := tx.Rollback(); err != nil {
-					// haha, what now?
+					pipe <- err
 				}
-				return err
+				return
 			}
 		} else if f.Direction == direction.Down {
 			if _, err := tx.Exec(`DELETE FROM `+tableName+` WHERE version=$1`, f.Version); err != nil {
+				pipe <- err
 				if err := tx.Rollback(); err != nil {
-					// haha, what now?
+					pipe <- err
 				}
-				return err
+				return
 			}
 		}
 
 		f.Read()
 		if _, err := tx.Exec(string(f.Content)); err != nil {
+			pipe <- err
 			if err := tx.Rollback(); err != nil {
-				// haha, what now?
+				pipe <- err
 			}
-			return err
+			return
 		}
+		pipe <- fmt.Sprintf("Applied %s", f.FileName)
+		time.Sleep(3 * time.Second)
 
 		if err := tx.Commit(); err != nil {
-			return err
+			pipe <- err
+			return
 		}
 	}
 
-	return nil
+	return
 }
 
 func (driver *Driver) Version() (uint64, error) {

driver/postgres/postgres_test.go

@@ -66,8 +66,15 @@ func TestMigrate(t *testing.T) {
 	})
 
 	applyFiles, _ := files.ToLastFrom(0)
-	if err := d.Migrate(applyFiles); err != nil {
-		t.Fatal(err)
+	pipe1 := make(chan interface{}, 0)
+	go d.Migrate(applyFiles, pipe1)
+	for {
+		select {
+		case _, ok := <-pipe1:
+			if !ok {
+				return
+			}
+		}
 	}
 
 	version, _ = d.Version()
@@ -80,8 +87,15 @@ func TestMigrate(t *testing.T) {
 	}
 
 	applyFiles2, _ := files.ToFirstFrom(1)
-	if err := d.Migrate(applyFiles2); err != nil {
-		t.Fatal(err)
+	pipe2 := make(chan interface{}, 0)
+	go d.Migrate(applyFiles2, pipe2)
+	for {
+		select {
+		case _, ok := <-pipe2:
+			if !ok {
+				return
+			}
+		}
 	}
 
 	version, _ = d.Version()

main.go

@@ -66,6 +66,29 @@ func verifyMigrationsPath(path string) {
 	}
 }
 
+func writePipe(pipe chan interface{}) {
+	if pipe != nil {
+		for {
+			select {
+			case item, ok := <-pipe:
+				if !ok {
+					return
+
+				} else {
+					switch item.(type) {
+					case string:
+						fmt.Println(item.(string))
+					case error:
+						fmt.Println(item.(error).Error())
+					default:
+						fmt.Println("%v", item)
+					}
+				}
+			}
+		}
+	}
+}
+
 func createCmd(url, migrationsPath, name string) {
 	migrationFile, err := migrate.Create(url, migrationsPath, name)
 	if err != nil {
@@ -80,35 +103,19 @@ func createCmd(url, migrationsPath, name string) {
 }
 
 func upCmd(url, migrationsPath string) {
-	if err := migrate.Up(url, migrationsPath); err != nil {
-		fmt.Println(err)
-		os.Exit(1)
-	}
-
+	writePipe(migrate.UpPipe(url, migrationsPath, nil))
 }
 
 func downCmd(url, migrationsPath string) {
-	if err := migrate.Down(url, migrationsPath); err != nil {
-		fmt.Println(err)
-		os.Exit(1)
-	}
-
+	writePipe(migrate.DownPipe(url, migrationsPath, nil))
 }
 
 func redoCmd(url, migrationsPath string) {
-	if err := migrate.Redo(url, migrationsPath); err != nil {
-		fmt.Println(err)
-		os.Exit(1)
-	}
-
+	writePipe(migrate.RedoPipe(url, migrationsPath, nil))
 }
 
 func resetCmd(url, migrationsPath string) {
-	if err := migrate.Reset(url, migrationsPath); err != nil {
-		fmt.Println(err)
-		os.Exit(1)
-	}
-
+	writePipe(migrate.ResetPipe(url, migrationsPath, nil))
 }
 
 func versionCmd(url, migrationsPath string) {
@@ -121,10 +128,7 @@ func versionCmd(url, migrationsPath string) {
 }
 
 func migrateCmd(url, migrationsPath string, relativeN int) {
-	if err := migrate.Migrate(url, migrationsPath, relativeN); err != nil {
-		fmt.Println(err)
-		os.Exit(1)
-	}
+	writePipe(migrate.MigratePipe(url, migrationsPath, relativeN, nil))
 }
 
 func helpCmd() {