Browse Source

Allow SQL backend to support alternative drivers (#95)

* backends: Replace MySQL backend with generic SQL backend

This replaces the `mysql` backend with an `sql` backend to avoid a
direct dependency for clients that don't need this plugin and to allow
alternative SQL drivers to be used instead.

* backends: Extend MySQL/Redis backend to support any SQL database

This decouples the `GuerrillaRedisDB` backend from MySQL to avoid a
direct dependency for clients that don't need this plugin and to allow
alternative SQL drivers to be used instead.

* backends: Add a test suite for the SQL backend

This provides some more confidence in the behaviour of the SQL
test. By default, it is skipped, but can be enabled by providing
appropriate configuration.

For example:

	go test -run SQL -sql-dsn="test:secret@(127.0.0.1:3306)/guerrilla_test"

The test itself does no setup, so a database, user, and table must be
created in advance:

	CREATE DATABASE IF NOT EXISTS `guerrilla_test`;
        USE `guerrilla_test`;
	CREATE TABLE IF NOT EXISTS `test` (
	  `mail_id` BIGINT(20) unsigned NOT NULL AUTO_INCREMENT,
	  `message_id` varchar(256) character set latin1 NOT NULL COMMENT 'value of [Message-ID] from headers',
	  `date` datetime NOT NULL,
	  `from` varchar(256) character set latin1 NOT NULL COMMENT 'value of [From] from headers or return_path (MAIL FROM) if no header present',
	  `to` varchar(256) character set latin1 NOT NULL COMMENT 'value of [To] from headers or recipient (RCPT TO) if no header present',
	  `reply_to` varchar(256) NULL COMMENT 'value of [Reply-To] from headers if present',
	  `sender` varchar(256) NULL COMMENT 'value of [Sender] from headers of present',
	  `subject` varchar(255) NOT NULL,
	  `body` varchar(16) NOT NULL,
	  `mail` longblob NOT NULL,
	  `spam_score` float NOT NULL,
	  `hash` char(32) character set latin1 NOT NULL,
	  `content_type` varchar(64) character set latin1 NOT NULL,
	  `recipient` varchar(255) character set latin1 NOT NULL COMMENT 'set by the RCPT TO command.',
	  `has_attach` int(11) NOT NULL,
	  `ip_addr` varbinary(16) NOT NULL,
	  `return_path` VARCHAR(255) NOT NULL COMMENT 'set by the MAIL FROM command. Can be empty to indicate a bounce, i.e <>',
	  `is_tls` BIT(1) DEFAULT b'0' NOT NULL,
	  PRIMARY KEY  (`mail_id`),
            KEY `to` (`to`),
	  KEY `hash` (`hash`),
	  KEY `date` (`date`)
        ) ENGINE=InnoDB  DEFAULT CHARSET=utf8;

	GRANT SELECT ON guerrilla_test.* TO 'test'@'localhost' IDENTIFIED BY 'secret';
	GRANT INSERT ON guerrilla_test.* TO 'test'@'localhost' IDENTIFIED BY 'secret';
Daniel White 7 years ago
parent
commit
d2bf4f4de2
5 changed files with 169 additions and 109 deletions
  1. 16 28
      backends/p_guerrilla_db_redis.go
  2. 42 63
      backends/p_sql.go
  3. 93 0
      backends/p_sql_test.go
  4. 6 3
      cmd/guerrillad/serve.go
  5. 12 15
      cmd/guerrillad/serve_test.go

+ 16 - 28
backends/p_guerrilla_db_redis.go

@@ -5,21 +5,21 @@ import (
 	"compress/zlib"
 	"compress/zlib"
 	"database/sql"
 	"database/sql"
 	"fmt"
 	"fmt"
-	"github.com/flashmob/go-guerrilla/mail"
-	"github.com/garyburd/redigo/redis"
-	"github.com/go-sql-driver/mysql"
 	"io"
 	"io"
 	"math/rand"
 	"math/rand"
 	"runtime/debug"
 	"runtime/debug"
 	"strings"
 	"strings"
 	"sync"
 	"sync"
 	"time"
 	"time"
+
+	"github.com/flashmob/go-guerrilla/mail"
+	"github.com/garyburd/redigo/redis"
 )
 )
 
 
 // ----------------------------------------------------------------------------------
 // ----------------------------------------------------------------------------------
-// Processor Name: GuerrillaRedsDB
+// Processor Name: GuerrillaRedisDB
 // ----------------------------------------------------------------------------------
 // ----------------------------------------------------------------------------------
-// Description   : Saves the body to redis, meta data to mysql. Example only.
+// Description   : Saves the body to redis, meta data to SQL. Example only.
 //               : Limitation: it doesn't save multiple recipients or validate them
 //               : Limitation: it doesn't save multiple recipients or validate them
 // ----------------------------------------------------------------------------------
 // ----------------------------------------------------------------------------------
 // Config Options: ...
 // Config Options: ...
@@ -56,15 +56,13 @@ type stmtCache [GuerrillaDBAndRedisBatchMax]*sql.Stmt
 
 
 type guerrillaDBAndRedisConfig struct {
 type guerrillaDBAndRedisConfig struct {
 	NumberOfWorkers    int    `json:"save_workers_size"`
 	NumberOfWorkers    int    `json:"save_workers_size"`
-	MysqlTable         string `json:"mail_table"`
-	MysqlDB            string `json:"mysql_db"`
-	MysqlHost          string `json:"mysql_host"`
-	MysqlPass          string `json:"mysql_pass"`
-	MysqlUser          string `json:"mysql_user"`
+	Table              string `json:"mail_table"`
+	Driver             string `json:"sql_driver"`
+	DSN                string `json:"sql_dsn"`
 	RedisExpireSeconds int    `json:"redis_expire_seconds"`
 	RedisExpireSeconds int    `json:"redis_expire_seconds"`
 	RedisInterface     string `json:"redis_interface"`
 	RedisInterface     string `json:"redis_interface"`
 	PrimaryHost        string `json:"primary_mail_host"`
 	PrimaryHost        string `json:"primary_mail_host"`
-	BatchTimeout       int    `json:"redis_mysql_batch_timeout,omitempty"`
+	BatchTimeout       int    `json:"redis_sql_batch_timeout,omitempty"`
 }
 }
 
 
 // Load the backend config for the backend. It has already been unmarshalled
 // Load the backend config for the backend. It has already been unmarshalled
@@ -153,7 +151,7 @@ func (g *GuerrillaDBAndRedisBackend) prepareInsertQuery(rows int, db *sql.DB) *s
 	if g.cache[rows-1] != nil {
 	if g.cache[rows-1] != nil {
 		return g.cache[rows-1]
 		return g.cache[rows-1]
 	}
 	}
-	sqlstr := "INSERT INTO " + g.config.MysqlTable + " "
+	sqlstr := "INSERT INTO " + g.config.Table + " "
 	sqlstr += "(`date`, `to`, `from`, `subject`, `body`, `charset`, `mail`, `spam_score`, `hash`, `content_type`, `recipient`, `has_attach`, `ip_addr`, `return_path`, `is_tls`)"
 	sqlstr += "(`date`, `to`, `from`, `subject`, `body`, `charset`, `mail`, `spam_score`, `hash`, `content_type`, `recipient`, `has_attach`, `ip_addr`, `return_path`, `is_tls`)"
 	sqlstr += " values "
 	sqlstr += " values "
 	values := "(NOW(), ?, ?, ?, ? , 'UTF-8' , ?, 0, ?, '', ?, 0, ?, ?, ?)"
 	values := "(NOW(), ?, ?, ?, ? , 'UTF-8' , ?, 0, ?, '', ?, 0, ?, ?, ?)"
@@ -304,7 +302,7 @@ func trimToLimit(str string, limit int) string {
 	return ret
 	return ret
 }
 }
 
 
-func (g *GuerrillaDBAndRedisBackend) mysqlConnect() (*sql.DB, error) {
+func (g *GuerrillaDBAndRedisBackend) sqlConnect() (*sql.DB, error) {
 	tOut := GuerrillaDBAndRedisBatchTimeout
 	tOut := GuerrillaDBAndRedisBatchTimeout
 	if g.config.BatchTimeout > 0 {
 	if g.config.BatchTimeout > 0 {
 		tOut = time.Duration(g.config.BatchTimeout)
 		tOut = time.Duration(g.config.BatchTimeout)
@@ -314,22 +312,12 @@ func (g *GuerrillaDBAndRedisBackend) mysqlConnect() (*sql.DB, error) {
 	if tOut >= 30 {
 	if tOut >= 30 {
 		tOut = 29
 		tOut = 29
 	}
 	}
-	conf := mysql.Config{
-		User:         g.config.MysqlUser,
-		Passwd:       g.config.MysqlPass,
-		DBName:       g.config.MysqlDB,
-		Net:          "tcp",
-		Addr:         g.config.MysqlHost,
-		ReadTimeout:  tOut * time.Second,
-		WriteTimeout: tOut * time.Second,
-		Params:       map[string]string{"collation": "utf8_general_ci"},
-	}
-	if db, err := sql.Open("mysql", conf.FormatDSN()); err != nil {
-		Log().Error("cannot open mysql", err, "]")
+	if db, err := sql.Open(g.config.Driver, g.config.DSN); err != nil {
+		Log().Error("cannot open database", err, "]")
 		return nil, err
 		return nil, err
 	} else {
 	} else {
 		// do we have access?
 		// do we have access?
-		_, err = db.Query("SELECT mail_id FROM " + g.config.MysqlTable + " LIMIT 1")
+		_, err = db.Query("SELECT mail_id FROM " + g.config.Table + " LIMIT 1")
 		if err != nil {
 		if err != nil {
 			Log().Error("cannot select table", err)
 			Log().Error("cannot select table", err)
 			return nil, err
 			return nil, err
@@ -376,7 +364,7 @@ func GuerrillaDbRedis() Decorator {
 			return err
 			return err
 		}
 		}
 		g.config = bcfg.(*guerrillaDBAndRedisConfig)
 		g.config = bcfg.(*guerrillaDBAndRedisConfig)
-		db, err = g.mysqlConnect()
+		db, err = g.sqlConnect()
 		if err != nil {
 		if err != nil {
 			return err
 			return err
 		}
 		}
@@ -401,7 +389,7 @@ func GuerrillaDbRedis() Decorator {
 
 
 	Svc.AddShutdowner(ShutdownWith(func() error {
 	Svc.AddShutdowner(ShutdownWith(func() error {
 		db.Close()
 		db.Close()
-		Log().Infof("closed mysql")
+		Log().Infof("closed sql")
 		if redisClient.conn != nil {
 		if redisClient.conn != nil {
 			Log().Infof("closed redis")
 			Log().Infof("closed redis")
 			redisClient.conn.Close()
 			redisClient.conn.Close()

+ 42 - 63
backends/p_mysql.go → backends/p_sql.go

@@ -4,29 +4,26 @@ import (
 	"database/sql"
 	"database/sql"
 	"fmt"
 	"fmt"
 	"strings"
 	"strings"
-	"time"
 
 
 	"github.com/flashmob/go-guerrilla/mail"
 	"github.com/flashmob/go-guerrilla/mail"
-	"github.com/go-sql-driver/mysql"
 
 
-	"github.com/flashmob/go-guerrilla/response"
 	"math/big"
 	"math/big"
 	"net"
 	"net"
 	"runtime/debug"
 	"runtime/debug"
+
+	"github.com/flashmob/go-guerrilla/response"
 )
 )
 
 
 // ----------------------------------------------------------------------------------
 // ----------------------------------------------------------------------------------
-// Processor Name: mysql
+// Processor Name: sql
 // ----------------------------------------------------------------------------------
 // ----------------------------------------------------------------------------------
-// Description   : Saves the e.Data (email data) and e.DeliveryHeader together in mysql
+// Description   : Saves the e.Data (email data) and e.DeliveryHeader together in sql
 //               : using the hash generated by the "hash" processor and stored in
 //               : using the hash generated by the "hash" processor and stored in
 //               : e.Hashes
 //               : e.Hashes
 // ----------------------------------------------------------------------------------
 // ----------------------------------------------------------------------------------
-// Config Options: mail_table string - mysql table name
-//               : mysql_db string - mysql database name
-//               : mysql_host string - mysql host name, eg. 127.0.0.1
-//               : mysql_pass string - mysql password
-//               : mysql_user string - mysql username
+// Config Options: mail_table string - name of table for storing emails
+//               : sql_driver string - database driver name, eg. mysql
+//               : sql_dsn string - driver-specific data source name
 //               : primary_mail_host string - primary host name
 //               : primary_mail_host string - primary host name
 // --------------:-------------------------------------------------------------------
 // --------------:-------------------------------------------------------------------
 // Input         : e.Data
 // Input         : e.Data
@@ -37,64 +34,47 @@ import (
 // Output        : Sets e.QueuedId with the first item fromHashes[0]
 // Output        : Sets e.QueuedId with the first item fromHashes[0]
 // ----------------------------------------------------------------------------------
 // ----------------------------------------------------------------------------------
 func init() {
 func init() {
-	processors["mysql"] = func() Decorator {
-		return MySql()
+	processors["sql"] = func() Decorator {
+		return SQL()
 	}
 	}
 }
 }
 
 
-const procMySQLReadTimeout = time.Second * 10
-const procMySQLWriteTimeout = time.Second * 10
-
-type MysqlProcessorConfig struct {
-	MysqlTable  string `json:"mysql_mail_table"`
-	MysqlDB     string `json:"mysql_db"`
-	MysqlHost   string `json:"mysql_host"`
-	MysqlPass   string `json:"mysql_pass"`
-	MysqlUser   string `json:"mysql_user"`
+type SQLProcessorConfig struct {
+	Table       string `json:"mail_table"`
+	Driver      string `json:"sql_driver"`
+	DSN         string `json:"sql_dsn"`
 	PrimaryHost string `json:"primary_mail_host"`
 	PrimaryHost string `json:"primary_mail_host"`
 }
 }
 
 
-type MysqlProcessor struct {
+type SQLProcessor struct {
 	cache  stmtCache
 	cache  stmtCache
-	config *MysqlProcessorConfig
+	config *SQLProcessorConfig
 }
 }
 
 
-func (m *MysqlProcessor) connect(config *MysqlProcessorConfig) (*sql.DB, error) {
+func (s *SQLProcessor) connect() (*sql.DB, error) {
 	var db *sql.DB
 	var db *sql.DB
 	var err error
 	var err error
-	conf := mysql.Config{
-		User:         config.MysqlUser,
-		Passwd:       config.MysqlPass,
-		DBName:       config.MysqlDB,
-		Net:          "tcp",
-		Addr:         config.MysqlHost,
-		ReadTimeout:  procMySQLReadTimeout,
-		WriteTimeout: procMySQLWriteTimeout,
-		Params:       map[string]string{"collation": "utf8_general_ci"},
-	}
-	if db, err = sql.Open("mysql", conf.FormatDSN()); err != nil {
-		Log().Error("cannot open mysql", err)
+	if db, err = sql.Open(s.config.Driver, s.config.DSN); err != nil {
+		Log().Error("cannot open database: ", err)
 		return nil, err
 		return nil, err
 	}
 	}
 	// do we have permission to access the table?
 	// do we have permission to access the table?
-	_, err = db.Query("SELECT mail_id FROM " + m.config.MysqlTable + " LIMIT 1")
+	_, err = db.Query("SELECT mail_id FROM " + s.config.Table + " LIMIT 1")
 	if err != nil {
 	if err != nil {
-		//Log().Error("cannot select table", err)
 		return nil, err
 		return nil, err
 	}
 	}
-	Log().Info("connected to mysql on tcp ", config.MysqlHost)
 	return db, err
 	return db, err
 }
 }
 
 
 // prepares the sql query with the number of rows that can be batched with it
 // prepares the sql query with the number of rows that can be batched with it
-func (g *MysqlProcessor) prepareInsertQuery(rows int, db *sql.DB) *sql.Stmt {
+func (s *SQLProcessor) prepareInsertQuery(rows int, db *sql.DB) *sql.Stmt {
 	if rows == 0 {
 	if rows == 0 {
 		panic("rows argument cannot be 0")
 		panic("rows argument cannot be 0")
 	}
 	}
-	if g.cache[rows-1] != nil {
-		return g.cache[rows-1]
+	if s.cache[rows-1] != nil {
+		return s.cache[rows-1]
 	}
 	}
-	sqlstr := "INSERT INTO " + g.config.MysqlTable + " "
+	sqlstr := "INSERT INTO " + s.config.Table + " "
 	sqlstr += "(`date`, `to`, `from`, `subject`, `body`,  `mail`, `spam_score`, "
 	sqlstr += "(`date`, `to`, `from`, `subject`, `body`,  `mail`, `spam_score`, "
 	sqlstr += "`hash`, `content_type`, `recipient`, `has_attach`, `ip_addr`, "
 	sqlstr += "`hash`, `content_type`, `recipient`, `has_attach`, `ip_addr`, "
 	sqlstr += "`return_path`, `is_tls`, `message_id`, `reply_to`, `sender`)"
 	sqlstr += "`return_path`, `is_tls`, `message_id`, `reply_to`, `sender`)"
@@ -113,11 +93,11 @@ func (g *MysqlProcessor) prepareInsertQuery(rows int, db *sql.DB) *sql.Stmt {
 		Log().WithError(sqlErr).Panic("failed while db.Prepare(INSERT...)")
 		Log().WithError(sqlErr).Panic("failed while db.Prepare(INSERT...)")
 	}
 	}
 	// cache it
 	// cache it
-	g.cache[rows-1] = stmt
+	s.cache[rows-1] = stmt
 	return stmt
 	return stmt
 }
 }
 
 
-func (g *MysqlProcessor) doQuery(c int, db *sql.DB, insertStmt *sql.Stmt, vals *[]interface{}) (execErr error) {
+func (s *SQLProcessor) doQuery(c int, db *sql.DB, insertStmt *sql.Stmt, vals *[]interface{}) (execErr error) {
 	defer func() {
 	defer func() {
 		if r := recover(); r != nil {
 		if r := recover(); r != nil {
 			Log().Error("Recovered form panic:", r, string(debug.Stack()))
 			Log().Error("Recovered form panic:", r, string(debug.Stack()))
@@ -132,7 +112,7 @@ func (g *MysqlProcessor) doQuery(c int, db *sql.DB, insertStmt *sql.Stmt, vals *
 		}
 		}
 	}()
 	}()
 	// prepare the query used to insert when rows reaches batchMax
 	// prepare the query used to insert when rows reaches batchMax
-	insertStmt = g.prepareInsertQuery(c, db)
+	insertStmt = s.prepareInsertQuery(c, db)
 	_, execErr = insertStmt.Exec(*vals...)
 	_, execErr = insertStmt.Exec(*vals...)
 	if execErr != nil {
 	if execErr != nil {
 		Log().WithError(execErr).Error("There was a problem the insert")
 		Log().WithError(execErr).Error("There was a problem the insert")
@@ -141,7 +121,7 @@ func (g *MysqlProcessor) doQuery(c int, db *sql.DB, insertStmt *sql.Stmt, vals *
 }
 }
 
 
 // for storing ip addresses in the ip_addr column
 // for storing ip addresses in the ip_addr column
-func (g *MysqlProcessor) ip2bint(ip string) *big.Int {
+func (s *SQLProcessor) ip2bint(ip string) *big.Int {
 	bint := big.NewInt(0)
 	bint := big.NewInt(0)
 	addr := net.ParseIP(ip)
 	addr := net.ParseIP(ip)
 	if strings.Index(ip, "::") > 0 {
 	if strings.Index(ip, "::") > 0 {
@@ -152,7 +132,7 @@ func (g *MysqlProcessor) ip2bint(ip string) *big.Int {
 	return bint
 	return bint
 }
 }
 
 
-func (g *MysqlProcessor) fillAddressFromHeader(e *mail.Envelope, headerKey string) string {
+func (s *SQLProcessor) fillAddressFromHeader(e *mail.Envelope, headerKey string) string {
 	if v, ok := e.Header[headerKey]; ok {
 	if v, ok := e.Header[headerKey]; ok {
 		addr, err := mail.NewAddress(v[0])
 		addr, err := mail.NewAddress(v[0])
 		if err != nil {
 		if err != nil {
@@ -163,23 +143,22 @@ func (g *MysqlProcessor) fillAddressFromHeader(e *mail.Envelope, headerKey strin
 	return ""
 	return ""
 }
 }
 
 
-func MySql() Decorator {
-
-	var config *MysqlProcessorConfig
+func SQL() Decorator {
+	var config *SQLProcessorConfig
 	var vals []interface{}
 	var vals []interface{}
 	var db *sql.DB
 	var db *sql.DB
-	m := &MysqlProcessor{}
+	s := &SQLProcessor{}
 
 
 	// open the database connection (it will also check if we can select the table)
 	// open the database connection (it will also check if we can select the table)
 	Svc.AddInitializer(InitializeWith(func(backendConfig BackendConfig) error {
 	Svc.AddInitializer(InitializeWith(func(backendConfig BackendConfig) error {
-		configType := BaseConfig(&MysqlProcessorConfig{})
+		configType := BaseConfig(&SQLProcessorConfig{})
 		bcfg, err := Svc.ExtractConfig(backendConfig, configType)
 		bcfg, err := Svc.ExtractConfig(backendConfig, configType)
 		if err != nil {
 		if err != nil {
 			return err
 			return err
 		}
 		}
-		config = bcfg.(*MysqlProcessorConfig)
-		m.config = config
-		db, err = m.connect(config)
+		config = bcfg.(*SQLProcessorConfig)
+		s.config = config
+		db, err = s.connect()
 		if err != nil {
 		if err != nil {
 			return err
 			return err
 		}
 		}
@@ -221,19 +200,19 @@ func MySql() Decorator {
 				for i := range e.RcptTo {
 				for i := range e.RcptTo {
 
 
 					// use the To header, otherwise rcpt to
 					// use the To header, otherwise rcpt to
-					to = trimToLimit(m.fillAddressFromHeader(e, "To"), 255)
+					to = trimToLimit(s.fillAddressFromHeader(e, "To"), 255)
 					if to == "" {
 					if to == "" {
 						// trimToLimit(strings.TrimSpace(e.RcptTo[i].User)+"@"+config.PrimaryHost, 255)
 						// trimToLimit(strings.TrimSpace(e.RcptTo[i].User)+"@"+config.PrimaryHost, 255)
 						to = trimToLimit(strings.TrimSpace(e.RcptTo[i].String()), 255)
 						to = trimToLimit(strings.TrimSpace(e.RcptTo[i].String()), 255)
 					}
 					}
-					mid := trimToLimit(m.fillAddressFromHeader(e, "Message-Id"), 255)
+					mid := trimToLimit(s.fillAddressFromHeader(e, "Message-Id"), 255)
 					if mid == "" {
 					if mid == "" {
 						mid = fmt.Sprintf("%s.%s@%s", hash, e.RcptTo[i].User, config.PrimaryHost)
 						mid = fmt.Sprintf("%s.%s@%s", hash, e.RcptTo[i].User, config.PrimaryHost)
 					}
 					}
 					// replyTo is the 'Reply-to' header, it may be blank
 					// replyTo is the 'Reply-to' header, it may be blank
-					replyTo := trimToLimit(m.fillAddressFromHeader(e, "Reply-To"), 255)
+					replyTo := trimToLimit(s.fillAddressFromHeader(e, "Reply-To"), 255)
 					// sender is the 'Sender' header, it may be blank
 					// sender is the 'Sender' header, it may be blank
-					sender := trimToLimit(m.fillAddressFromHeader(e, "Sender"), 255)
+					sender := trimToLimit(s.fillAddressFromHeader(e, "Sender"), 255)
 
 
 					recipient := trimToLimit(strings.TrimSpace(e.RcptTo[i].String()), 255)
 					recipient := trimToLimit(strings.TrimSpace(e.RcptTo[i].String()), 255)
 					contentType := ""
 					contentType := ""
@@ -265,7 +244,7 @@ func MySql() Decorator {
 						hash, // hash (redis hash if saved in redis)
 						hash, // hash (redis hash if saved in redis)
 						contentType,
 						contentType,
 						recipient,
 						recipient,
-						m.ip2bint(e.RemoteIP).Bytes(),         // ip_addr store as varbinary(16)
+						s.ip2bint(e.RemoteIP).Bytes(),         // ip_addr store as varbinary(16)
 						trimToLimit(e.MailFrom.String(), 255), // return_path
 						trimToLimit(e.MailFrom.String(), 255), // return_path
 						e.TLS,   // is_tls
 						e.TLS,   // is_tls
 						mid,     // message_id
 						mid,     // message_id
@@ -273,8 +252,8 @@ func MySql() Decorator {
 						sender,
 						sender,
 					)
 					)
 
 
-					stmt := m.prepareInsertQuery(1, db)
-					err := m.doQuery(1, db, stmt, &vals)
+					stmt := s.prepareInsertQuery(1, db)
+					err := s.doQuery(1, db, stmt, &vals)
 					if err != nil {
 					if err != nil {
 						return NewResult(fmt.Sprint("554 Error: could not save email")), StorageError
 						return NewResult(fmt.Sprint("554 Error: could not save email")), StorageError
 					}
 					}

+ 93 - 0
backends/p_sql_test.go

@@ -0,0 +1,93 @@
+package backends
+
+import (
+	"database/sql"
+	"flag"
+	"fmt"
+	"strconv"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/flashmob/go-guerrilla/log"
+	"github.com/flashmob/go-guerrilla/mail"
+
+	_ "github.com/go-sql-driver/mysql"
+)
+
+var (
+	mailTableFlag = flag.String("mail-table", "test", "Table to use for testing the SQL backend")
+	sqlDSNFlag    = flag.String("sql-dsn", "", "DSN to use for testing the SQL backend")
+	sqlDriverFlag = flag.String("sql-driver", "mysql", "Driver to use for testing the SQL backend")
+)
+
+func TestSQL(t *testing.T) {
+	if *sqlDSNFlag == "" {
+		t.Skip("requires -sql-dsn to run")
+	}
+
+	logger, err := log.GetLogger(log.OutputOff.String(), log.DebugLevel.String())
+	if err != nil {
+		t.Fatal("get logger:", err)
+	}
+
+	cfg := BackendConfig{
+		"save_process":      "sql",
+		"mail_table":        *mailTableFlag,
+		"primary_mail_host": "example.com",
+		"sql_driver":        *sqlDriverFlag,
+		"sql_dsn":           *sqlDSNFlag,
+	}
+	backend, err := New(cfg, logger)
+	if err != nil {
+		t.Fatal("new backend:", err)
+	}
+	if err := backend.Start(); err != nil {
+		t.Fatal("start backend: ", err)
+	}
+
+	hash := strconv.FormatInt(time.Now().UnixNano(), 10)
+	envelope := &mail.Envelope{
+		RcptTo: []mail.Address{{User: "user", Host: "example.com"}},
+		Hashes: []string{hash},
+	}
+
+	// The SQL processor is expected to use the hash to queue the mail.
+	result := backend.Process(envelope)
+	if !strings.Contains(result.String(), hash) {
+		t.Errorf("expected message to be queued with hash, got %q", result)
+	}
+
+	// Ensure that a record actually exists.
+	results, err := findRows(hash)
+	if err != nil {
+		t.Fatal("find rows: ", err)
+	}
+	if len(results) != 1 {
+		t.Fatalf("expected one row, got %d", len(results))
+	}
+}
+
+func findRows(hash string) ([]string, error) {
+	db, err := sql.Open(*sqlDriverFlag, *sqlDSNFlag)
+	if err != nil {
+		return nil, err
+	}
+	defer db.Close()
+
+	stmt := fmt.Sprintf(`SELECT hash FROM %s WHERE hash = ?`, *mailTableFlag)
+	rows, err := db.Query(stmt, hash)
+	if err != nil {
+		return nil, err
+	}
+
+	var results []string
+	for rows.Next() {
+		var result string
+		if err := rows.Scan(&result); err != nil {
+			return nil, err
+		}
+		results = append(results, result)
+	}
+	return results, nil
+}

+ 6 - 3
cmd/guerrillad/serve.go

@@ -2,9 +2,6 @@ package main
 
 
 import (
 import (
 	"fmt"
 	"fmt"
-	"github.com/flashmob/go-guerrilla"
-	"github.com/flashmob/go-guerrilla/log"
-	"github.com/spf13/cobra"
 	"os"
 	"os"
 	"os/exec"
 	"os/exec"
 	"os/signal"
 	"os/signal"
@@ -12,6 +9,12 @@ import (
 	"strings"
 	"strings"
 	"syscall"
 	"syscall"
 	"time"
 	"time"
+
+	"github.com/flashmob/go-guerrilla"
+	"github.com/flashmob/go-guerrilla/log"
+	"github.com/spf13/cobra"
+
+	_ "github.com/go-sql-driver/mysql"
 )
 )
 
 
 const (
 const (

+ 12 - 15
cmd/guerrillad/serve_test.go

@@ -3,12 +3,6 @@ package main
 import (
 import (
 	"crypto/tls"
 	"crypto/tls"
 	"encoding/json"
 	"encoding/json"
-	"github.com/flashmob/go-guerrilla"
-	"github.com/flashmob/go-guerrilla/backends"
-	"github.com/flashmob/go-guerrilla/log"
-	test "github.com/flashmob/go-guerrilla/tests"
-	"github.com/flashmob/go-guerrilla/tests/testcert"
-	"github.com/spf13/cobra"
 	"io/ioutil"
 	"io/ioutil"
 	"os"
 	"os"
 	"os/exec"
 	"os/exec"
@@ -18,6 +12,13 @@ import (
 	"sync"
 	"sync"
 	"testing"
 	"testing"
 	"time"
 	"time"
+
+	"github.com/flashmob/go-guerrilla"
+	"github.com/flashmob/go-guerrilla/backends"
+	"github.com/flashmob/go-guerrilla/log"
+	test "github.com/flashmob/go-guerrilla/tests"
+	"github.com/flashmob/go-guerrilla/tests/testcert"
+	"github.com/spf13/cobra"
 )
 )
 
 
 var configJsonA = `
 var configJsonA = `
@@ -120,10 +121,8 @@ var configJsonC = `
     "backend_name": "guerrilla-redis-db",
     "backend_name": "guerrilla-redis-db",
     "backend_config" :
     "backend_config" :
         {
         {
-            "mysql_db":"gmail_mail",
-            "mysql_host":"127.0.0.1:3306",
-            "mysql_pass":"ok",
-            "mysql_user":"root",
+            "sql_driver": "mysql",
+            "sql_dsn": "root:ok@tcp(127.0.0.1:3306)/gmail_mail?readTimeout=10&writeTimeout=10",
             "mail_table":"new_mail",
             "mail_table":"new_mail",
             "redis_interface" : "127.0.0.1:6379",
             "redis_interface" : "127.0.0.1:6379",
             "redis_expire_seconds" : 7200,
             "redis_expire_seconds" : 7200,
@@ -231,13 +230,11 @@ var configJsonE = `
             "save_process_old": "HeadersParser|Debugger|Hasher|Header|Compressor|Redis|MySql",
             "save_process_old": "HeadersParser|Debugger|Hasher|Header|Compressor|Redis|MySql",
             "save_process": "GuerrillaRedisDB",
             "save_process": "GuerrillaRedisDB",
             "log_received_mails" : true,
             "log_received_mails" : true,
-            "mysql_db":"gmail_mail",
-            "mysql_host":"127.0.0.1:3306",
-            "mysql_pass":"secret",
-            "mysql_user":"root",
+            "sql_driver": "mysql",
+            "sql_dsn": "root:secret@tcp(127.0.0.1:3306)/gmail_mail?readTimeout=10&writeTimeout=10",
             "mail_table":"new_mail",
             "mail_table":"new_mail",
             "redis_interface" : "127.0.0.1:6379",
             "redis_interface" : "127.0.0.1:6379",
-             "redis_expire_seconds" : 7200,
+            "redis_expire_seconds" : 7200,
             "save_workers_size" : 3,
             "save_workers_size" : 3,
             "primary_mail_host":"sharklasers.com"
             "primary_mail_host":"sharklasers.com"
         },
         },