add decorator pattern

backends/abstract.go

@@ -5,12 +5,14 @@ import (
 	"fmt"
 	"github.com/flashmob/go-guerrilla/envelope"
 	"reflect"
+	"runtime/debug"
 	"strings"
 )
 
 type AbstractBackend struct {
 	config abstractConfig
-	extend Backend
+	Extend Worker
+	p      Processor
 }
 
 type abstractConfig struct {
@@ -30,12 +32,22 @@ func (b *AbstractBackend) loadConfig(backendConfig BackendConfig) (err error) {
 	}
 	m := bcfg.(*abstractConfig)
 	b.config = *m
+
 	return nil
 }
 
+func (b *AbstractBackend) SetProcessors(p ...Decorator) {
+	// This backend will parse headers and then debugger
+	if b.Extend != nil {
+		b.Extend.SetProcessors(p...)
+		return
+	}
+	b.p = Decorate(DefaultProcessor{}, p...)
+}
+
 func (b *AbstractBackend) Initialize(config BackendConfig) error {
-	if b.extend != nil {
-		return b.extend.loadConfig(config)
+	if b.Extend != nil {
+		return b.Extend.loadConfig(config)
 	}
 	err := b.loadConfig(config)
 	if err != nil {
@@ -45,35 +57,31 @@ func (b *AbstractBackend) Initialize(config BackendConfig) error {
 }
 
 func (b *AbstractBackend) Shutdown() error {
-	if b.extend != nil {
-		return b.extend.Shutdown()
+	if b.Extend != nil {
+		return b.Extend.Shutdown()
 	}
 	return nil
 }
 
 func (b *AbstractBackend) Process(mail *envelope.Envelope) BackendResult {
-	if b.extend != nil {
-		return b.extend.Process(mail)
-	}
-	mail.ParseHeaders()
-
-	if b.config.LogReceivedMails {
-		mainlog.Infof("Mail from: %s / to: %v", mail.MailFrom.String(), mail.RcptTo)
-		mainlog.Info("Headers are: %s", mail.Header)
-
+	if b.Extend != nil {
+		return b.Extend.Process(mail)
 	}
+	// call the decorated process function
+	b.p.Process(mail)
 	return NewBackendResult("250 OK")
 }
 
 func (b *AbstractBackend) saveMailWorker(saveMailChan chan *savePayload) {
-	if b.extend != nil {
-		b.extend.saveMailWorker(saveMailChan)
+	if b.Extend != nil {
+		b.Extend.saveMailWorker(saveMailChan)
 		return
 	}
 	defer func() {
 		if r := recover(); r != nil {
 			// recover form closed channel
-			fmt.Println("Recovered in f", r)
+			fmt.Println("Recovered in f", r, string(debug.Stack()))
+			mainlog.Error("Recovered form panic:", r, string(debug.Stack()))
 		}
 		// close any connections / files
 		// ...
@@ -98,15 +106,15 @@ func (b *AbstractBackend) saveMailWorker(saveMailChan chan *savePayload) {
 }
 
 func (b *AbstractBackend) getNumberOfWorkers() int {
-	if b.extend != nil {
-		return b.extend.getNumberOfWorkers()
+	if b.Extend != nil {
+		return b.Extend.getNumberOfWorkers()
 	}
 	return 1
 }
 
 func (b *AbstractBackend) testSettings() error {
-	if b.extend != nil {
-		return b.extend.testSettings()
+	if b.Extend != nil {
+		return b.Extend.testSettings()
 	}
 	return nil
 }

backends/backend.go

@@ -1,16 +1,11 @@
 package backends
 
 import (
-	"errors"
 	"fmt"
-	"strconv"
-	"strings"
-	"sync"
-	"time"
-
 	"github.com/flashmob/go-guerrilla/envelope"
 	"github.com/flashmob/go-guerrilla/log"
-	"github.com/flashmob/go-guerrilla/response"
+	"strconv"
+	"strings"
 )
 
 var mainlog log.Logger
@@ -24,7 +19,9 @@ type Backend interface {
 	Process(*envelope.Envelope) BackendResult
 	Initialize(BackendConfig) error
 	Shutdown() error
+}
 
+type Worker interface {
 	// start save mail worker(s)
 	saveMailWorker(chan *savePayload)
 	// get the number of workers that will be stared
@@ -33,11 +30,17 @@ type Backend interface {
 	testSettings() error
 	// parse the configuration files
 	loadConfig(BackendConfig) error
+
+	Shutdown() error
+	Process(*envelope.Envelope) BackendResult
+	Initialize(BackendConfig) error
+
+	SetProcessors(p ...Decorator)
 }
 
 type BackendConfig map[string]interface{}
 
-var backends = map[string]Backend{}
+var backends = map[string]Worker{}
 
 type baseConfig interface{}
 
@@ -47,9 +50,9 @@ type saveStatus struct {
 }
 
 type savePayload struct {
-	mail        *envelope.Envelope
-	from        *envelope.EmailAddress
-	recipient   *envelope.EmailAddress
+	mail *envelope.Envelope
+	//from        *envelope.EmailAddress
+	//recipient   *envelope.EmailAddress
 	savedNotify chan *saveStatus
 }
 
@@ -86,131 +89,3 @@ func (br backendResult) Code() int {
 func NewBackendResult(message string) BackendResult {
 	return backendResult(message)
 }
-
-// A backend gateway is a proxy that implements the Backend interface.
-// It is used to start multiple goroutine workers for saving mail, and then distribute email saving to the workers
-// via a channel. Shutting down via Shutdown() will stop all workers.
-// The rest of this program always talks to the backend via this gateway.
-type BackendGateway struct {
-	AbstractBackend
-	saveMailChan chan *savePayload
-	// waits for backend workers to start/stop
-	wg sync.WaitGroup
-	b  Backend
-	// controls access to state
-	stateGuard sync.Mutex
-	State      backendState
-	config     BackendConfig
-}
-
-// possible values for state
-const (
-	BackendStateRunning = iota
-	BackendStateShuttered
-	BackendStateError
-)
-
-type backendState int
-
-func (s backendState) String() string {
-	return strconv.Itoa(int(s))
-}
-
-// New retrieve a backend specified by the backendName, and initialize it using
-// backendConfig
-func New(backendName string, backendConfig BackendConfig, l log.Logger) (Backend, error) {
-	backend, found := backends[backendName]
-	mainlog = l
-	if !found {
-		return nil, fmt.Errorf("backend %q not found", backendName)
-	}
-	gateway := &BackendGateway{b: backend, config: backendConfig}
-	err := gateway.Initialize(backendConfig)
-	if err != nil {
-		return nil, fmt.Errorf("error while initializing the backend: %s", err)
-	}
-	gateway.State = BackendStateRunning
-	return gateway, nil
-}
-
-// Process distributes an envelope to one of the backend workers
-func (gw *BackendGateway) Process(e *envelope.Envelope) BackendResult {
-	if gw.State != BackendStateRunning {
-		return NewBackendResult(response.Canned.FailBackendNotRunning + gw.State.String())
-	}
-
-	to := e.RcptTo
-	from := e.MailFrom
-
-	// place on the channel so that one of the save mail workers can pick it up
-	// TODO: support multiple recipients
-	savedNotify := make(chan *saveStatus)
-	gw.saveMailChan <- &savePayload{e, &from, &to[0], savedNotify}
-	// wait for the save to complete
-	// or timeout
-	select {
-	case status := <-savedNotify:
-		if status.err != nil {
-			return NewBackendResult(response.Canned.FailBackendTransaction + status.err.Error())
-		}
-		return NewBackendResult(response.Canned.SuccessMessageQueued + status.hash)
-
-	case <-time.After(time.Second * 30):
-		mainlog.Infof("Backend has timed out")
-		return NewBackendResult(response.Canned.FailBackendTimeout)
-	}
-}
-func (gw *BackendGateway) Shutdown() error {
-	gw.stateGuard.Lock()
-	defer gw.stateGuard.Unlock()
-	if gw.State != BackendStateShuttered {
-		err := gw.b.Shutdown()
-		if err == nil {
-			close(gw.saveMailChan) // workers will stop
-			gw.wg.Wait()
-			gw.State = BackendStateShuttered
-		}
-		return err
-	}
-	return nil
-}
-
-// Reinitialize starts up a backend gateway that was shutdown before
-func (gw *BackendGateway) Reinitialize() error {
-	if gw.State != BackendStateShuttered {
-		return errors.New("backend must be in BackendStateshuttered state to Reinitialize")
-	}
-	err := gw.Initialize(gw.config)
-	if err != nil {
-		return fmt.Errorf("error while initializing the backend: %s", err)
-	}
-	gw.State = BackendStateRunning
-	return err
-}
-
-func (gw *BackendGateway) Initialize(cfg BackendConfig) error {
-	err := gw.b.Initialize(cfg)
-	if err == nil {
-		workersSize := gw.b.getNumberOfWorkers()
-		if workersSize < 1 {
-			gw.State = BackendStateError
-			return errors.New("Must have at least 1 worker")
-		}
-		if err := gw.b.testSettings(); err != nil {
-			gw.State = BackendStateError
-			return err
-		}
-		gw.saveMailChan = make(chan *savePayload, workersSize)
-		// start our savemail workers
-		gw.wg.Add(workersSize)
-		for i := 0; i < workersSize; i++ {
-			go func() {
-				gw.b.saveMailWorker(gw.saveMailChan)
-				gw.wg.Done()
-			}()
-		}
-	} else {
-		gw.State = BackendStateError
-	}
-	return err
-}

backends/debugger.go

@@ -0,0 +1,15 @@
+package backends
+
+import (
+	"github.com/flashmob/go-guerrilla/envelope"
+)
+
+func Debugger() Decorator {
+	return func(c Processor) Processor {
+		return ProcessorFunc(func(e *envelope.Envelope) (BackendResult, error) {
+			mainlog.Infof("Mail from: %s / to: %v", e.MailFrom.String(), e.RcptTo)
+			mainlog.Info("So, Headers are: %s", e.Header)
+			return c.Process(e)
+		})
+	}
+}

backends/decorate.go

@@ -0,0 +1,13 @@
+package backends
+
+// We define what a decorator to our processor will look like
+type Decorator func(Processor) Processor
+
+// Decorate will decorate a processor with a slice of passed decorators
+func Decorate(c Processor, ds ...Decorator) Processor {
+	decorated := c
+	for _, decorate := range ds {
+		decorated = decorate(decorated)
+	}
+	return decorated
+}

backends/dummy.go

@@ -1,10 +1,12 @@
 package backends
 
 func init() {
-	// decorator pattern
 	backends["dummy"] = &AbstractBackend{
-		extend: &DummyBackend{},
+		Extend: &DummyBackend{},
 	}
+
+	backends["dummy"].SetProcessors(Debugger(), HeadersParser())
+
 }
 
 // custom configuration we will parse from the json

backends/gateway.go

@@ -0,0 +1,136 @@
+package backends
+
+import (
+	"errors"
+	"fmt"
+	"strconv"
+	"sync"
+	"time"
+
+	"github.com/flashmob/go-guerrilla/envelope"
+	"github.com/flashmob/go-guerrilla/log"
+	"github.com/flashmob/go-guerrilla/response"
+)
+
+// A backend gateway is a proxy that implements the Backend interface.
+// It is used to start multiple goroutine workers for saving mail, and then distribute email saving to the workers
+// via a channel. Shutting down via Shutdown() will stop all workers.
+// The rest of this program always talks to the backend via this gateway.
+type BackendGateway struct {
+	AbstractBackend
+	saveMailChan chan *savePayload
+	// waits for backend workers to start/stop
+	wg sync.WaitGroup
+	b  Worker
+	// controls access to state
+	stateGuard sync.Mutex
+	State      backendState
+	config     BackendConfig
+}
+
+// possible values for state
+const (
+	BackendStateRunning = iota
+	BackendStateShuttered
+	BackendStateError
+)
+
+type backendState int
+
+func (s backendState) String() string {
+	return strconv.Itoa(int(s))
+}
+
+// New retrieve a backend specified by the backendName, and initialize it using
+// backendConfig
+func New(backendName string, backendConfig BackendConfig, l log.Logger) (Backend, error) {
+	backend, found := backends[backendName]
+	mainlog = l
+	if !found {
+		return nil, fmt.Errorf("backend %q not found", backendName)
+	}
+	gateway := &BackendGateway{b: backend, config: backendConfig}
+	err := gateway.Initialize(backendConfig)
+	if err != nil {
+		return nil, fmt.Errorf("error while initializing the backend: %s", err)
+	}
+	gateway.State = BackendStateRunning
+	return gateway, nil
+}
+
+// Process distributes an envelope to one of the backend workers
+func (gw *BackendGateway) Process(e *envelope.Envelope) BackendResult {
+	if gw.State != BackendStateRunning {
+		return NewBackendResult(response.Canned.FailBackendNotRunning + gw.State.String())
+	}
+	// place on the channel so that one of the save mail workers can pick it up
+	savedNotify := make(chan *saveStatus)
+	gw.saveMailChan <- &savePayload{e, savedNotify}
+	// wait for the save to complete
+	// or timeout
+	select {
+	case status := <-savedNotify:
+		if status.err != nil {
+			return NewBackendResult(response.Canned.FailBackendTransaction + status.err.Error())
+		}
+		return NewBackendResult(response.Canned.SuccessMessageQueued + status.hash)
+
+	case <-time.After(time.Second * 30):
+		mainlog.Infof("Backend has timed out")
+		return NewBackendResult(response.Canned.FailBackendTimeout)
+	}
+}
+func (gw *BackendGateway) Shutdown() error {
+	gw.stateGuard.Lock()
+	defer gw.stateGuard.Unlock()
+	if gw.State != BackendStateShuttered {
+		err := gw.b.Shutdown()
+		if err == nil {
+			close(gw.saveMailChan) // workers will stop
+			gw.wg.Wait()
+			gw.State = BackendStateShuttered
+		}
+		return err
+	}
+	return nil
+}
+
+// Reinitialize starts up a backend gateway that was shutdown before
+func (gw *BackendGateway) Reinitialize() error {
+	if gw.State != BackendStateShuttered {
+		return errors.New("backend must be in BackendStateshuttered state to Reinitialize")
+	}
+	err := gw.Initialize(gw.config)
+	if err != nil {
+		return fmt.Errorf("error while initializing the backend: %s", err)
+	}
+	gw.State = BackendStateRunning
+	return err
+}
+
+func (gw *BackendGateway) Initialize(cfg BackendConfig) error {
+	err := gw.b.Initialize(cfg)
+	if err == nil {
+		workersSize := gw.b.getNumberOfWorkers()
+		if workersSize < 1 {
+			gw.State = BackendStateError
+			return errors.New("Must have at least 1 worker")
+		}
+		if err := gw.b.testSettings(); err != nil {
+			gw.State = BackendStateError
+			return err
+		}
+		gw.saveMailChan = make(chan *savePayload, workersSize)
+		// start our savemail workers
+		gw.wg.Add(workersSize)
+		for i := 0; i < workersSize; i++ {
+			go func() {
+				gw.b.saveMailWorker(gw.saveMailChan)
+				gw.wg.Done()
+			}()
+		}
+	} else {
+		gw.State = BackendStateError
+	}
+	return err
+}

backends/guerrilla_db_redis.go

@@ -48,7 +48,7 @@ const GuerrillaDBAndRedisBatchTimeout = time.Second * 3
 
 func init() {
 	backends["guerrilla-db-redis"] = &AbstractBackend{
-		extend: &GuerrillaDBAndRedisBackend{}}
+		Extend: &GuerrillaDBAndRedisBackend{}}
 }
 
 type GuerrillaDBAndRedisBackend struct {
@@ -362,9 +362,9 @@ func (g *GuerrillaDBAndRedisBackend) saveMailWorker(saveMailChan chan *savePaylo
 			return
 		}
 		mainlog.Debug("Got mail from chan", payload.mail.RemoteAddress)
-		to = trimToLimit(strings.TrimSpace(payload.recipient.User)+"@"+g.config.PrimaryHost, 255)
+		to = trimToLimit(strings.TrimSpace(payload.mail.RcptTo[0].User)+"@"+g.config.PrimaryHost, 255)
 		payload.mail.Helo = trimToLimit(payload.mail.Helo, 255)
-		payload.recipient.Host = trimToLimit(payload.recipient.Host, 255)
+		host := trimToLimit(payload.mail.RcptTo[0].Host, 255)
 		ts := fmt.Sprintf("%d", time.Now().UnixNano())
 		payload.mail.ParseHeaders()
 		hash := MD5Hex(
@@ -376,7 +376,7 @@ func (g *GuerrillaDBAndRedisBackend) saveMailWorker(saveMailChan chan *savePaylo
 		var addHead string
 		addHead += "Delivered-To: " + to + "\r\n"
 		addHead += "Received: from " + payload.mail.Helo + " (" + payload.mail.Helo + "  [" + payload.mail.RemoteAddress + "])\r\n"
-		addHead += "	by " + payload.recipient.Host + " with SMTP id " + hash + "@" + payload.recipient.Host + ";\r\n"
+		addHead += "	by " + host + " with SMTP id " + hash + "@" + host + ";\r\n"
 		addHead += "	" + time.Now().Format(time.RFC1123Z) + "\r\n"
 
 		// data will be compressed when printed, with addHead added to beginning

backends/headers_parser.go

@@ -0,0 +1,15 @@
+package backends
+
+import (
+	"github.com/flashmob/go-guerrilla/envelope"
+)
+
+func HeadersParser() Decorator {
+	return func(c Processor) Processor {
+		return ProcessorFunc(func(e *envelope.Envelope) (BackendResult, error) {
+			mainlog.Info("parse headers")
+			e.ParseHeaders()
+			return c.Process(e)
+		})
+	}
+}

backends/logger.go

@@ -0,0 +1,10 @@
+package backends
+
+type loggerConfig struct {
+	LogReceivedMails bool `json:"log_received_mails"`
+}
+
+// putting all the paces we need together
+type LoggerBackend struct {
+	config dummyConfig
+}

backends/processor.go

@@ -0,0 +1,27 @@
+package backends
+
+import (
+	"github.com/flashmob/go-guerrilla/envelope"
+)
+
+// Our processor is defined as something that processes the envelope and returns a result and error
+type Processor interface {
+	Process(*envelope.Envelope) (BackendResult, error)
+}
+
+// Signature of DoFunc
+type ProcessorFunc func(*envelope.Envelope) (BackendResult, error)
+
+// Add method to DoFunc type to satisfy Client interface
+func (f ProcessorFunc) Process(e *envelope.Envelope) (BackendResult, error) {
+	return f(e)
+}
+
+// DefaultProcessor is a undecorated worker that does nothing
+// Notice MockClient has no knowledge of the other decorators that have orthogonal concerns.
+type DefaultProcessor struct{}
+
+// do nothing except return the result
+func (w DefaultProcessor) Process(e *envelope.Envelope) (BackendResult, error) {
+	return NewBackendResult("200 OK"), nil
+}

envelope/envelope.go

@@ -64,7 +64,8 @@ func (e *Envelope) ParseHeaders() error {
 	if len(all) < max {
 		max = len(all) - 1
 	}
-	headerEnd := bytes.Index(all[:max], []byte("\n\n"))
+	str := string(all[:max])
+	headerEnd := strings.Index(str, "\n\n")
 
 	if headerEnd > -1 {
 		headerReader := textproto.NewReader(bufio.NewReader(bytes.NewBuffer(all[0:headerEnd])))