Add Cassandra Enhancements

* Authentication functionality,
* Execution of multiple statements in one migration file

Author: syndbg

database/cassandra/cassandra.go

@@ -2,13 +2,15 @@ package cassandra
 
 import (
 	"fmt"
+	"github.com/gocql/gocql"
+	"github.com/mattes/migrate/database"
 	"io"
 	"io/ioutil"
 	nurl "net/url"
-	"github.com/gocql/gocql"
-	"time"
-	"github.com/mattes/migrate/database"
+	"regexp"
 	"strconv"
+	"strings"
+	"time"
 )
 
 func init() {
@@ -20,8 +22,8 @@ var DefaultMigrationsTable = "schema_migrations"
 var dbLocked = false
 
 var (
-	ErrNilConfig = fmt.Errorf("no config")
-	ErrNoKeyspace = fmt.Errorf("no keyspace provided")
+	ErrNilConfig     = fmt.Errorf("no config")
+	ErrNoKeyspace    = fmt.Errorf("no keyspace provided")
 	ErrDatabaseDirty = fmt.Errorf("database is dirty")
 )
 
@@ -35,7 +37,7 @@ type Cassandra struct {
 	isLocked bool
 
 	// Open and WithInstance need to guarantee that config is never nil
-	config   *Config
+	config *Config
 }
 
 func (p *Cassandra) Open(url string) (database.Driver, error) {
@@ -59,11 +61,24 @@ func (p *Cassandra) Open(url string) (database.Driver, error) {
 		MigrationsTable: migrationsTable,
 	}
 
+	var username, password string
+
+	if u.User != nil {
+		username = u.User.Username()
+		password, _ = u.User.Password()
+	} else {
+		username = "cassandra"
+		password = "cassandra"
+	}
+
 	cluster := gocql.NewCluster(u.Host)
 	cluster.Keyspace = u.Path[1:len(u.Path)]
 	cluster.Consistency = gocql.All
 	cluster.Timeout = 1 * time.Minute
-
+	cluster.Authenticator = gocql.PasswordAuthenticator{
+		Username: username,
+		Password: password,
+	}
 
 	// Retrieve query string configuration
 	if len(u.Query().Get("consistency")) > 0 {
@@ -111,7 +126,7 @@ func (p *Cassandra) Close() error {
 }
 
 func (p *Cassandra) Lock() error {
-	if (dbLocked) {
+	if dbLocked {
 		return database.ErrLocked
 	}
 	dbLocked = true
@@ -130,9 +145,17 @@ func (p *Cassandra) Run(migration io.Reader) error {
 	}
 	// run migration
 	query := string(migr[:])
-	if err := p.session.Query(query).Exec(); err != nil {
-		// TODO: cast to Cassandra error and get line number
-		return database.Error{OrigErr: err, Err: "migration failed", Query: migr}
+	matches := regexp.MustCompile(`(?m:;$)`).Split(query, -1)
+	for _, match := range matches {
+		trimmedMatch := strings.Trim(match, " \t\r\n")
+		if len(trimmedMatch) == 0 {
+			continue
+		}
+
+		if err := p.session.Query(trimmedMatch).Exec(); err != nil {
+			// TODO: cast to Cassandra error and get line number
+			return database.Error{OrigErr: err, Err: "migration failed", Query: migr}
+		}
 	}
 
 	return nil
@@ -153,7 +176,6 @@ func (p *Cassandra) SetVersion(version int, dirty bool) error {
 	return nil
 }
 
-
 // Return current keyspace version
 func (p *Cassandra) Version() (version int, dirty bool, err error) {
 	query := `SELECT version, dirty FROM "` + p.config.MigrationsTable + `" LIMIT 1`
@@ -191,7 +213,6 @@ func (p *Cassandra) Drop() error {
 	return nil
 }
 
-
 // Ensure version table exists
 func (p *Cassandra) ensureVersionTable() error {
 	err := p.session.Query(fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s (version bigint, dirty boolean, PRIMARY KEY(version))", p.config.MigrationsTable)).Exec()
@@ -204,7 +225,6 @@ func (p *Cassandra) ensureVersionTable() error {
 	return nil
 }
 
-
 // ParseConsistency wraps gocql.ParseConsistency
 // to return an error instead of a panicking.
 func parseConsistency(consistencyStr string) (consistency gocql.Consistency, err error) {