Refactoring

- rename Service to Svc
- rename BackendResult to Result (the package already has the word backend)
- rename saveStatus to notifyMsg to better reflect what it does
-Rename NewBackendResult to NewResult
- add ValidateRcpt method to Backend interface
- use private vars where possible
- Added the Errors type
- Svc.shutdown returns Errors
- Svc.initialize returns Errors
- renamed savePayload to workerMsg for a more general name
- added a new validateRcptChan channel
- savedNotify renamed to  notifyMe for more fluent code readability
- backend timeout now a constant  ProcessTimeout
- new ValidateRcpt method implimented to add validateRcptChan channel
- getNumberOfWorkers() renamed to workersSize()
- saveMailWorker() renamed to workDispatcher() so the name better reflects what it does
- Change the Processor interface to add a new task param, allows the selecting of what task to do (so far, save or validate)
- Update all processors to use new Processor interface
- Renamed Envelope.Info to Envelope.Values
- New Canned.FailRcptCmd response message for when recipient does not exist
- Added PushRcpt and PopRcpt to Envelope

backends/backend.go

@@ -12,19 +12,23 @@ import (
 )
 
 var (
-	Service *BackendService
+	Svc *Service
+
 	// deprecated backends system
 	backends = map[string]Backend{}
-	// new backends system
-	Processors map[string]ProcessorConstructor
+	// Store the constructor for making an new processor decorator.
+	processors map[string]processorConstructor
+
+	b Backend // todo make sure legacy works
 )
 
 func init() {
-	Service = &BackendService{}
-	Processors = make(map[string]ProcessorConstructor)
+	Svc = &Service{}
+
+	processors = make(map[string]processorConstructor)
 }
 
-type ProcessorConstructor func() Decorator
+type processorConstructor func() Decorator
 
 // Backends process received mail. Depending on the implementation, they can store mail in the database,
 // write to a file, check for spam, re-transmit to another server, etc.
@@ -32,7 +36,8 @@ type ProcessorConstructor func() Decorator
 // whether the message was processed successfully.
 type Backend interface {
 	// Public methods
-	Process(*envelope.Envelope) BackendResult
+	Process(*envelope.Envelope) Result
+	ValidateRcpt(e *envelope.Envelope) RcptError
 	Initialize(BackendConfig) error
 	Shutdown() error
 }
@@ -42,30 +47,30 @@ type BackendConfig map[string]interface{}
 // All config structs extend from this
 type BaseConfig interface{}
 
-type saveStatus struct {
+type notifyMsg struct {
 	err      error
 	queuedID string
 }
 
-// BackendResult represents a response to an SMTP client after receiving DATA.
+// Result represents a response to an SMTP client after receiving DATA.
 // The String method should return an SMTP message ready to send back to the
 // client, for example `250 OK: Message received`.
-type BackendResult interface {
+type Result interface {
 	fmt.Stringer
 	// Code should return the SMTP code associated with this response, ie. `250`
 	Code() int
 }
 
 // Internal implementation of BackendResult for use by backend implementations.
-type backendResult string
+type result string
 
-func (br backendResult) String() string {
+func (br result) String() string {
 	return string(br)
 }
 
 // Parses the SMTP code from the first 3 characters of the SMTP message.
 // Returns 554 if code cannot be parsed.
-func (br backendResult) Code() int {
+func (br result) Code() int {
 	trimmed := strings.TrimSpace(string(br))
 	if len(trimmed) < 3 {
 		return 554
@@ -77,8 +82,8 @@ func (br backendResult) Code() int {
 	return code
 }
 
-func NewBackendResult(message string) BackendResult {
-	return backendResult(message)
+func NewResult(message string) Result {
+	return result(message)
 }
 
 type ProcessorInitializer interface {
@@ -120,76 +125,110 @@ func (e Errors) Error() string {
 	return msg
 }
 
-type BackendService struct {
-	Initializers []ProcessorInitializer
-	Shutdowners  []ProcessorShutdowner
+// New retrieve a backend specified by the backendName, and initialize it using
+// backendConfig
+func New(backendName string, backendConfig BackendConfig, l log.Logger) (Backend, error) {
+	Svc.StoreMainlog(l)
+	if backend, found := backends[backendName]; found {
+		b = backend
+	} else {
+		gateway := &BackendGateway{config: backendConfig}
+		err := gateway.Initialize(backendConfig)
+		if err != nil {
+			return nil, fmt.Errorf("error while initializing the backend: %s", err)
+		}
+		gateway.State = BackendStateRunning
+		b = Backend(gateway)
+	}
+	return b, nil
+}
+
+func GetBackend() Backend {
+	return b
+}
+
+type Service struct {
+	initializers []ProcessorInitializer
+	shutdowners  []ProcessorShutdowner
 	sync.Mutex
-	mainlog    atomic.Value
-	initErrors Errors
+	mainlog atomic.Value
 }
 
 // Get loads the log.logger in an atomic operation. Returns a stderr logger if not able to load
 func Log() log.Logger {
-	if v, ok := Service.mainlog.Load().(log.Logger); ok {
+	if v, ok := Svc.mainlog.Load().(log.Logger); ok {
 		return v
 	}
 	l, _ := log.GetLogger(log.OutputStderr.String())
 	return l
 }
 
-func (b *BackendService) StoreMainlog(l log.Logger) {
-	b.mainlog.Store(l)
+func (s *Service) StoreMainlog(l log.Logger) {
+	s.mainlog.Store(l)
 }
 
-// AddInitializer adds a function that impliments ProcessorShutdowner to be called when initializing
-func (b *BackendService) AddInitializer(i ProcessorInitializer) {
-	b.Lock()
-	defer b.Unlock()
-	b.Initializers = append(b.Initializers, i)
+// AddInitializer adds a function that implements ProcessorShutdowner to be called when initializing
+func (s *Service) AddInitializer(i ProcessorInitializer) {
+	s.Lock()
+	defer s.Unlock()
+	s.initializers = append(s.initializers, i)
 }
 
-// AddShutdowner adds a function that impliments ProcessorShutdowner to be called when shutting down
-func (b *BackendService) AddShutdowner(i ProcessorShutdowner) {
-	b.Lock()
-	defer b.Unlock()
-	b.Shutdowners = append(b.Shutdowners, i)
+// AddShutdowner adds a function that implements ProcessorShutdowner to be called when shutting down
+func (s *Service) AddShutdowner(sh ProcessorShutdowner) {
+	s.Lock()
+	defer s.Unlock()
+	s.shutdowners = append(s.shutdowners, sh)
 }
 
 // Initialize initializes all the processors one-by-one and returns any errors.
-func (b *BackendService) Initialize(backend BackendConfig) Errors {
-	b.Lock()
-	defer b.Unlock()
-	b.initErrors = nil
-	for i := range b.Initializers {
-		err := b.Initializers[i].Initialize(backend)
-		if err != nil {
-			b.initErrors = append(b.initErrors, err)
+// Subsequent calls to Initialize will not call the initializer again unless it failed on the previous call
+// so Initialize may be called again to retry after getting errors
+func (s *Service) initialize(backend BackendConfig) Errors {
+	s.Lock()
+	defer s.Unlock()
+	var errors Errors
+	failed := make([]ProcessorInitializer, 0)
+	for i := range s.initializers {
+		if err := s.initializers[i].Initialize(backend); err != nil {
+			errors = append(errors, err)
+			failed = append(failed, s.initializers[i])
 		}
 	}
-	return b.initErrors
+	// keep only the failed initializers
+	s.initializers = failed
+	return errors
 }
 
-// Shutdown shuts down all the processor by calling their shutdowners
-// It also clears the initializers and shutdowners that were set with AddInitializer and AddShutdowner
-func (b *BackendService) Shutdown() {
-	b.Lock()
-	defer b.Unlock()
-	for i := range b.Shutdowners {
-		b.Shutdowners[i].Shutdown()
+// Shutdown shuts down all the processors by calling their shutdowners (if any)
+// Subsequent calls to Shutdown will not call the shutdowners again unless it failed on the previous call
+// so Shutdown may be called again to retry after getting errors
+func (s *Service) shutdown() Errors {
+	s.Lock()
+	defer s.Unlock()
+	var errors Errors
+	failed := make([]ProcessorShutdowner, 0)
+	for i := range s.shutdowners {
+		if err := s.shutdowners[i].Shutdown(); err != nil {
+			errors = append(errors, err)
+			failed = append(failed, s.shutdowners[i])
+		}
 	}
-	b.Initializers = make([]ProcessorInitializer, 0)
-	b.Shutdowners = make([]ProcessorShutdowner, 0)
+	s.shutdowners = failed
+	return errors
 }
 
 // AddProcessor adds a new processor, which becomes available to the backend_config.process_stack option
-func (b *BackendService) AddProcessor(name string, p ProcessorConstructor) {
+// Use to add your own custom processor when using backends as a package, or after importing an external
+// processor.
+func (s *Service) AddProcessor(name string, p processorConstructor) {
 	// wrap in a constructor since we want to defer calling it
-	var c ProcessorConstructor
+	var c processorConstructor
 	c = func() Decorator {
 		return p()
 	}
 	// add to our processors list
-	Processors[strings.ToLower(name)] = c
+	processors[strings.ToLower(name)] = c
 }
 
 // extractConfig loads the backend config. It has already been unmarshalled
@@ -198,15 +237,15 @@ func (b *BackendService) AddProcessor(name string, p ProcessorConstructor) {
 // The reason why using reflection is because we'll get a nice error message if the field is missing
 // the alternative solution would be to json.Marshal() and json.Unmarshal() however that will not give us any
 // error messages
-func (b *BackendService) ExtractConfig(configData BackendConfig, configType BaseConfig) (interface{}, error) {
+func (s *Service) ExtractConfig(configData BackendConfig, configType BaseConfig) (interface{}, error) {
 	// Use reflection so that we can provide a nice error message
-	s := reflect.ValueOf(configType).Elem() // so that we can set the values
-	m := reflect.ValueOf(configType).Elem()
+	v := reflect.ValueOf(configType).Elem() // so that we can set the values
+	//m := reflect.ValueOf(configType).Elem()
 	t := reflect.TypeOf(configType).Elem()
-	typeOfT := s.Type()
+	typeOfT := v.Type()
 
-	for i := 0; i < m.NumField(); i++ {
-		f := s.Field(i)
+	for i := 0; i < v.NumField(); i++ {
+		f := v.Field(i)
 		// read the tags of the config struct
 		field_name := t.Field(i).Tag.Get("json")
 		if len(field_name) > 0 {
@@ -222,23 +261,23 @@ func (b *BackendService) ExtractConfig(configData BackendConfig, configType Base
 		if f.Type().Name() == "int" {
 			// in json, there is no int, only floats...
 			if intVal, converted := configData[field_name].(float64); converted {
-				s.Field(i).SetInt(int64(intVal))
+				v.Field(i).SetInt(int64(intVal))
 			} else if intVal, converted := configData[field_name].(int); converted {
-				s.Field(i).SetInt(int64(intVal))
+				v.Field(i).SetInt(int64(intVal))
 			} else {
 				return configType, convertError("property missing/invalid: '" + field_name + "' of expected type: " + f.Type().Name())
 			}
 		}
 		if f.Type().Name() == "string" {
 			if stringVal, converted := configData[field_name].(string); converted {
-				s.Field(i).SetString(stringVal)
+				v.Field(i).SetString(stringVal)
 			} else {
 				return configType, convertError("missing/invalid: '" + field_name + "' of type: " + f.Type().Name())
 			}
 		}
 		if f.Type().Name() == "bool" {
 			if boolVal, converted := configData[field_name].(bool); converted {
-				s.Field(i).SetBool(boolVal)
+				v.Field(i).SetBool(boolVal)
 			} else {
 				return configType, convertError("missing/invalid: '" + field_name + "' of type: " + f.Type().Name())
 			}

backends/gateway.go

@@ -8,7 +8,6 @@ import (
 	"time"
 
 	"github.com/flashmob/go-guerrilla/envelope"
-	"github.com/flashmob/go-guerrilla/log"
 	"github.com/flashmob/go-guerrilla/response"
 	"strings"
 )
@@ -18,11 +17,14 @@ import (
 // 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 {
-	saveMailChan chan *savePayload
+	// channel for sending envelopes to process and save
+	saveMailChan chan *workerMsg
+	// channel for validating the last recipient added to the envelope
+	validateRcptChan chan *workerMsg
 	// waits for backend workers to start/stop
 	wg sync.WaitGroup
 	w  *Worker
-	b  Backend
+
 	// controls access to state
 	sync.Mutex
 	State    backendState
@@ -36,10 +38,10 @@ type GatewayConfig struct {
 }
 
 // savePayload is what get placed on the BackendGateway.saveMailChan channel
-type savePayload struct {
+type workerMsg struct {
 	mail *envelope.Envelope
 	// savedNotify is used to notify that the save operation completed
-	savedNotify chan *saveStatus
+	notifyMe chan *notifyMsg
 }
 
 // possible values for state
@@ -49,51 +51,60 @@ const (
 	BackendStateError
 )
 
+const ProcessTimeout = time.Second * 30
+
 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) {
-	Service.StoreMainlog(l)
-	gateway := &BackendGateway{config: backendConfig}
-	if backend, found := backends[backendName]; found {
-		gateway.b = backend
+// Process distributes an envelope to one of the backend workers
+func (gw *BackendGateway) Process(e *envelope.Envelope) Result {
+	if gw.State != BackendStateRunning {
+		return NewResult(response.Canned.FailBackendNotRunning + gw.State.String())
 	}
-	err := gateway.Initialize(backendConfig)
-	if err != nil {
-		return nil, fmt.Errorf("error while initializing the backend: %s", err)
+	// place on the channel so that one of the save mail workers can pick it up
+	savedNotify := make(chan *notifyMsg)
+	gw.saveMailChan <- &workerMsg{e, savedNotify}
+	// wait for the save to complete
+	// or timeout
+	select {
+	case status := <-savedNotify:
+		if status.err != nil {
+			return NewResult(response.Canned.FailBackendTransaction + status.err.Error())
+		}
+		return NewResult(response.Canned.SuccessMessageQueued + status.queuedID)
+
+	case <-time.After(ProcessTimeout):
+		Log().Infof("Backend has timed out")
+		return NewResult(response.Canned.FailBackendTimeout)
 	}
-	gateway.State = BackendStateRunning
 
-	return gateway, nil
 }
 
-// Process distributes an envelope to one of the backend workers
-func (gw *BackendGateway) Process(e *envelope.Envelope) BackendResult {
+// ValidateRcpt asks one of the workers to validate the recipient
+// Only the last recipient appended to e.RcptTo will be validated.
+func (gw *BackendGateway) ValidateRcpt(e *envelope.Envelope) RcptError {
 	if gw.State != BackendStateRunning {
-		return NewBackendResult(response.Canned.FailBackendNotRunning + gw.State.String())
+		return StorageNotAvailable
 	}
 	// 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
+	notify := make(chan *notifyMsg)
+	gw.validateRcptChan <- &workerMsg{e, notify}
+	// wait for the validation to complete
 	// or timeout
 	select {
-	case status := <-savedNotify:
+	case status := <-notify:
 		if status.err != nil {
-			return NewBackendResult(response.Canned.FailBackendTransaction + status.err.Error())
+			return status.err
 		}
-		return NewBackendResult(response.Canned.SuccessMessageQueued + status.queuedID)
+		return nil
 
-	case <-time.After(time.Second * 30):
+	case <-time.After(time.Second):
 		Log().Infof("Backend has timed out")
-		return NewBackendResult(response.Canned.FailBackendTimeout)
+		return StorageTimeout
 	}
-
 }
 
 // Shutdown shuts down the backend and leaves it in BackendStateShuttered state
@@ -104,7 +115,7 @@ func (gw *BackendGateway) Shutdown() error {
 		close(gw.saveMailChan) // workers will stop
 		// wait for workers to stop
 		gw.wg.Wait()
-		Service.Shutdown()
+		Svc.shutdown()
 		gw.State = BackendStateShuttered
 	}
 	return nil
@@ -136,7 +147,7 @@ func (gw *BackendGateway) newProcessorLine() Processor {
 	line := strings.Split(strings.ToLower(gw.gwConfig.ProcessorLine), "|")
 	for i := range line {
 		name := line[len(line)-1-i] // reverse order, since decorators are stacked
-		if makeFunc, ok := Processors[name]; ok {
+		if makeFunc, ok := processors[name]; ok {
 			decorators = append(decorators, makeFunc())
 		}
 	}
@@ -154,7 +165,7 @@ func (gw *BackendGateway) loadConfig(cfg BackendConfig) error {
 	if _, ok := cfg["save_workers_size"]; !ok {
 		cfg["save_workers_size"] = 1
 	}
-	bcfg, err := Service.ExtractConfig(cfg, configType)
+	bcfg, err := Svc.ExtractConfig(cfg, configType)
 	if err != nil {
 		return err
 	}
@@ -168,7 +179,7 @@ func (gw *BackendGateway) Initialize(cfg BackendConfig) error {
 	defer gw.Unlock()
 	err := gw.loadConfig(cfg)
 	if err == nil {
-		workersSize := gw.getNumberOfWorkers()
+		workersSize := gw.workersSize()
 		if workersSize < 1 {
 			gw.State = BackendStateError
 			return errors.New("Must have at least 1 worker")
@@ -178,15 +189,16 @@ func (gw *BackendGateway) Initialize(cfg BackendConfig) error {
 			lines = append(lines, gw.newProcessorLine())
 		}
 		// initialize processors
-		if err := Service.Initialize(cfg); err != nil {
+		if err := Svc.initialize(cfg); err != nil {
 			return err
 		}
-		gw.saveMailChan = make(chan *savePayload, workersSize)
-		// start our savemail workers
+		gw.saveMailChan = make(chan *workerMsg, workersSize)
+		gw.validateRcptChan = make(chan *workerMsg, workersSize)
+		// start our workers
 		gw.wg.Add(workersSize)
 		for i := 0; i < workersSize; i++ {
 			go func(workerId int) {
-				gw.w.saveMailWorker(gw.saveMailChan, lines[workerId], workerId+1)
+				gw.w.workDispatcher(gw.saveMailChan, gw.validateRcptChan, lines[workerId], workerId+1)
 				gw.wg.Done()
 			}(i)
 		}
@@ -196,9 +208,9 @@ func (gw *BackendGateway) Initialize(cfg BackendConfig) error {
 	return err
 }
 
-// getNumberOfWorkers gets the number of workers to use for saving email by reading the save_workers_size config value
+// workersSize gets the number of workers to use for saving email by reading the save_workers_size config value
 // Returns 1 if no config value was set
-func (gw *BackendGateway) getNumberOfWorkers() int {
+func (gw *BackendGateway) workersSize() int {
 	if gw.gwConfig.WorkersSize == 0 {
 		return 1
 	}

backends/guerrilla_db_redis.go

@@ -85,7 +85,7 @@ func convertError(name string) error {
 
 func (g *GuerrillaDBAndRedisBackend) loadConfig(backendConfig BackendConfig) (err error) {
 	configType := BaseConfig(&guerrillaDBAndRedisConfig{})
-	bcfg, err := Service.ExtractConfig(backendConfig, configType)
+	bcfg, err := Svc.ExtractConfig(backendConfig, configType)
 	if err != nil {
 		return err
 	}
@@ -98,6 +98,11 @@ func (g *GuerrillaDBAndRedisBackend) getNumberOfWorkers() int {
 	return g.config.NumberOfWorkers
 }
 
+// ValidateRcpt not implemented
+func (g *GuerrillaDBAndRedisBackend) ValidateRcpt(e *envelope.Envelope) RcptError {
+	return nil
+}
+
 type redisClient struct {
 	isConnected bool
 	conn        redis.Conn
@@ -309,7 +314,7 @@ func (g *GuerrillaDBAndRedisBackend) mysqlConnect() (*sql.DB, error) {
 
 }
 
-func (g *GuerrillaDBAndRedisBackend) saveMailWorker(saveMailChan chan *savePayload) {
+func (g *GuerrillaDBAndRedisBackend) saveMailWorker(saveMailChan chan *workerMsg) {
 	var to, body string
 
 	var redisErr error
@@ -413,7 +418,7 @@ func (g *GuerrillaDBAndRedisBackend) saveMailWorker(saveMailChan chan *savePaylo
 			trimToLimit(payload.mail.MailFrom.String(), 255),
 			payload.mail.TLS)
 		feeder <- vals
-		payload.savedNotify <- &saveStatus{nil, hash}
+		payload.notifyMe <- &notifyMsg{nil, hash}
 
 	}
 }
@@ -459,8 +464,8 @@ func (g *GuerrillaDBAndRedisBackend) Initialize(config BackendConfig) error {
 }
 
 // does nothing
-func (g *GuerrillaDBAndRedisBackend) Process(mail *envelope.Envelope) BackendResult {
-	return NewBackendResult("250 OK")
+func (g *GuerrillaDBAndRedisBackend) Process(mail *envelope.Envelope) Result {
+	return NewResult("250 OK")
 }
 
 // does nothing

backends/p_compressor.go

@@ -25,7 +25,7 @@ import (
 //               : after being printed
 // ----------------------------------------------------------------------------------
 func init() {
-	Processors["compressor"] = func() Decorator {
+	processors["compressor"] = func() Decorator {
 		return Compressor()
 	}
 }
@@ -91,12 +91,17 @@ func (c *compressor) clear() {
 
 func Compressor() Decorator {
 	return func(c Processor) Processor {
-		return ProcessorFunc(func(e *envelope.Envelope) (BackendResult, error) {
-			compressor := newCompressor()
-			compressor.set([]byte(e.DeliveryHeader), &e.Data)
-			// put the pinter in there for other processors to use later in the line
-			e.Info["zlib-compressor"] = compressor
-			return c.Process(e)
+		return ProcessorFunc(func(e *envelope.Envelope, task SelectTask) (Result, error) {
+			if task == TaskSaveMail {
+				compressor := newCompressor()
+				compressor.set([]byte(e.DeliveryHeader), &e.Data)
+				// put the pointer in there for other processors to use later in the line
+				e.Values["zlib-compressor"] = compressor
+				// continue to the next Processor in the decorator stack
+				return c.Process(e, task)
+			} else {
+				return c.Process(e, task)
+			}
 		})
 	}
 }

backends/p_debugger.go

@@ -16,7 +16,7 @@ import (
 // Output        : none (only output to the log if enabled)
 // ----------------------------------------------------------------------------------
 func init() {
-	Processors["debugger"] = func() Decorator {
+	processors["debugger"] = func() Decorator {
 		return Debugger()
 	}
 }
@@ -29,22 +29,26 @@ func Debugger() Decorator {
 	var config *debuggerConfig
 	initFunc := Initialize(func(backendConfig BackendConfig) error {
 		configType := BaseConfig(&debuggerConfig{})
-		bcfg, err := Service.ExtractConfig(backendConfig, configType)
+		bcfg, err := Svc.ExtractConfig(backendConfig, configType)
 		if err != nil {
 			return err
 		}
 		config = bcfg.(*debuggerConfig)
 		return nil
 	})
-	Service.AddInitializer(initFunc)
+	Svc.AddInitializer(initFunc)
 	return func(c Processor) Processor {
-		return ProcessorFunc(func(e *envelope.Envelope) (BackendResult, error) {
-			if config.LogReceivedMails {
-				Log().Infof("Mail from: %s / to: %v", e.MailFrom.String(), e.RcptTo)
-				Log().Info("Headers are:", e.Header)
+		return ProcessorFunc(func(e *envelope.Envelope, task SelectTask) (Result, error) {
+			if task == TaskSaveMail {
+				if config.LogReceivedMails {
+					Log().Infof("Mail from: %s / to: %v", e.MailFrom.String(), e.RcptTo)
+					Log().Info("Headers are:", e.Header)
+				}
+				// continue to the next Processor in the decorator stack
+				return c.Process(e, task)
+			} else {
+				return c.Process(e, task)
 			}
-			// continue to the next Processor in the decorator chain
-			return c.Process(e)
 		})
 	}
 }

backends/p_hasher.go

@@ -23,7 +23,7 @@ import (
 // Output        : Checksum stored in e.Hash
 // ----------------------------------------------------------------------------------
 func init() {
-	Processors["hasher"] = func() Decorator {
+	processors["hasher"] = func() Decorator {
 		return Hasher()
 	}
 }
@@ -32,24 +32,27 @@ func init() {
 // It appends the hashes to envelope's Hashes slice.
 func Hasher() Decorator {
 	return func(c Processor) Processor {
-		return ProcessorFunc(func(e *envelope.Envelope) (BackendResult, error) {
+		return ProcessorFunc(func(e *envelope.Envelope, task SelectTask) (Result, error) {
 
-			// base hash
-			h := md5.New()
-			ts := fmt.Sprintf("%d", time.Now().UnixNano())
-			io.Copy(h, strings.NewReader(e.MailFrom.String()))
-			io.Copy(h, strings.NewReader(e.Subject))
-			io.Copy(h, strings.NewReader(ts))
-
-			// using the base hash, calculate a unique hash for each recipient
-			for i := range e.RcptTo {
-				h2 := h // copy
-				io.Copy(h2, strings.NewReader(e.RcptTo[i].String()))
-				sum := h2.Sum([]byte{})
-				e.Hashes = append(e.Hashes, fmt.Sprintf("%x", sum))
+			if task == TaskSaveMail {
+				// base hash, use subject from and timestamp-nano
+				h := md5.New()
+				ts := fmt.Sprintf("%d", time.Now().UnixNano())
+				io.Copy(h, strings.NewReader(e.MailFrom.String()))
+				io.Copy(h, strings.NewReader(e.Subject))
+				io.Copy(h, strings.NewReader(ts))
+				// using the base hash, calculate a unique hash for each recipient
+				for i := range e.RcptTo {
+					h2 := h
+					io.Copy(h2, strings.NewReader(e.RcptTo[i].String()))
+					sum := h2.Sum([]byte{})
+					e.Hashes = append(e.Hashes, fmt.Sprintf("%x", sum))
+				}
+				return c.Process(e, task)
+			} else {
+				return c.Process(e, task)
 			}
 
-			return c.Process(e)
 		})
 	}
 }

backends/p_header.go

@@ -25,7 +25,7 @@ type HeaderConfig struct {
 // Output        : Sets e.DeliveryHeader with additional delivery info
 // ----------------------------------------------------------------------------------
 func init() {
-	Processors["header"] = func() Decorator {
+	processors["header"] = func() Decorator {
 		return Header()
 	}
 }
@@ -36,9 +36,9 @@ func Header() Decorator {
 
 	var config *HeaderConfig
 
-	Service.AddInitializer(Initialize(func(backendConfig BackendConfig) error {
+	Svc.AddInitializer(Initialize(func(backendConfig BackendConfig) error {
 		configType := BaseConfig(&HeaderConfig{})
-		bcfg, err := Service.ExtractConfig(backendConfig, configType)
+		bcfg, err := Svc.ExtractConfig(backendConfig, configType)
 		if err != nil {
 			return err
 		}
@@ -47,23 +47,28 @@ func Header() Decorator {
 	}))
 
 	return func(c Processor) Processor {
-		return ProcessorFunc(func(e *envelope.Envelope) (BackendResult, error) {
-			to := strings.TrimSpace(e.RcptTo[0].User) + "@" + config.PrimaryHost
-			hash := "unknown"
-			if len(e.Hashes) > 0 {
-				hash = e.Hashes[0]
-			}
-			var addHead string
-			addHead += "Delivered-To: " + to + "\n"
-			addHead += "Received: from " + e.Helo + " (" + e.Helo + "  [" + e.RemoteAddress + "])\n"
-			if len(e.RcptTo) > 0 {
-				addHead += "	by " + e.RcptTo[0].Host + " with SMTP id " + hash + "@" + e.RcptTo[0].Host + ";\n"
+		return ProcessorFunc(func(e *envelope.Envelope, task SelectTask) (Result, error) {
+			if task == TaskSaveMail {
+				to := strings.TrimSpace(e.RcptTo[0].User) + "@" + config.PrimaryHost
+				hash := "unknown"
+				if len(e.Hashes) > 0 {
+					hash = e.Hashes[0]
+				}
+				var addHead string
+				addHead += "Delivered-To: " + to + "\n"
+				addHead += "Received: from " + e.Helo + " (" + e.Helo + "  [" + e.RemoteAddress + "])\n"
+				if len(e.RcptTo) > 0 {
+					addHead += "	by " + e.RcptTo[0].Host + " with SMTP id " + hash + "@" + e.RcptTo[0].Host + ";\n"
+				}
+				addHead += "	" + time.Now().Format(time.RFC1123Z) + "\n"
+				// save the result
+				e.DeliveryHeader = addHead
+				// next processor
+				return c.Process(e, task)
+
+			} else {
+				return c.Process(e, task)
 			}
-			addHead += "	" + time.Now().Format(time.RFC1123Z) + "\n"
-			// save the result
-			e.DeliveryHeader = addHead
-			// next processor
-			return c.Process(e)
 		})
 	}
 }

backends/p_headers_parser.go

@@ -16,16 +16,22 @@ import (
 // Output        : Headers will be populated in e.Header
 // ----------------------------------------------------------------------------------
 func init() {
-	Processors["headersparser"] = func() Decorator {
+	processors["headersparser"] = func() Decorator {
 		return HeadersParser()
 	}
 }
 
 func HeadersParser() Decorator {
 	return func(c Processor) Processor {
-		return ProcessorFunc(func(e *envelope.Envelope) (BackendResult, error) {
-			e.ParseHeaders()
-			return c.Process(e)
+		return ProcessorFunc(func(e *envelope.Envelope, task SelectTask) (Result, error) {
+			if task == TaskSaveMail {
+				e.ParseHeaders()
+				// next processor
+				return c.Process(e, task)
+			} else {
+				// next processor
+				return c.Process(e, task)
+			}
 		})
 	}
 }

backends/p_mysql.go

@@ -8,6 +8,7 @@ import (
 	"github.com/flashmob/go-guerrilla/envelope"
 	"github.com/go-sql-driver/mysql"
 
+	"github.com/flashmob/go-guerrilla/response"
 	"runtime/debug"
 )
 
@@ -33,7 +34,7 @@ import (
 // Output        : Sets e.QueuedId with the first item fromHashes[0]
 // ----------------------------------------------------------------------------------
 func init() {
-	Processors["mysql"] = func() Decorator {
+	processors["mysql"] = func() Decorator {
 		return MySql()
 	}
 }
@@ -133,9 +134,9 @@ func MySql() Decorator {
 	var db *sql.DB
 	mp := &MysqlProcessor{}
 
-	Service.AddInitializer(Initialize(func(backendConfig BackendConfig) error {
+	Svc.AddInitializer(Initialize(func(backendConfig BackendConfig) error {
 		configType := BaseConfig(&MysqlProcessorConfig{})
-		bcfg, err := Service.ExtractConfig(backendConfig, configType)
+		bcfg, err := Svc.ExtractConfig(backendConfig, configType)
 		if err != nil {
 			return err
 		}
@@ -150,7 +151,7 @@ func MySql() Decorator {
 	}))
 
 	// shutdown
-	Service.AddShutdowner(Shutdown(func() error {
+	Svc.AddShutdowner(Shutdown(func() error {
 		if db != nil {
 			return db.Close()
 		}
@@ -158,55 +159,73 @@ func MySql() Decorator {
 	}))
 
 	return func(c Processor) Processor {
-		return ProcessorFunc(func(e *envelope.Envelope) (BackendResult, error) {
-			var to, body string
-			to = trimToLimit(strings.TrimSpace(e.RcptTo[0].User)+"@"+config.PrimaryHost, 255)
-			hash := ""
-			if len(e.Hashes) > 0 {
-				hash = e.Hashes[0]
-				e.QueuedId = e.Hashes[0]
-			}
+		return ProcessorFunc(func(e *envelope.Envelope, task SelectTask) (Result, error) {
+
+			if task == TaskSaveMail {
+				var to, body string
+				to = trimToLimit(strings.TrimSpace(e.RcptTo[0].User)+"@"+config.PrimaryHost, 255)
+				hash := ""
+				if len(e.Hashes) > 0 {
+					hash = e.Hashes[0]
+					e.QueuedId = e.Hashes[0]
+				}
 
-			var co *compressor
-			// a compressor was set
-			if c, ok := e.Info["zlib-compressor"]; ok {
-				body = "gzip"
-				co = c.(*compressor)
-			}
-			// was saved in redis
-			if _, ok := e.Info["redis"]; ok {
-				body = "redis"
-			}
+				var co *compressor
+				// a compressor was set
+				if c, ok := e.Values["zlib-compressor"]; ok {
+					body = "gzip"
+					co = c.(*compressor)
+				}
+				// was saved in redis
+				if _, ok := e.Values["redis"]; ok {
+					body = "redis"
+				}
+
+				// build the values for the query
+				vals = []interface{}{} // clear the vals
+				vals = append(vals,
+					to,
+					trimToLimit(e.MailFrom.String(), 255),
+					trimToLimit(e.Subject, 255),
+					body)
+				if body == "redis" {
+					// data already saved in redis
+					vals = append(vals, "")
+				} else if co != nil {
+					// use a compressor (automatically adds e.DeliveryHeader)
+					vals = append(vals, co.String())
+
+				} else {
+					vals = append(vals, e.String())
+				}
 
-			// build the values for the query
-			vals = []interface{}{} // clear the vals
-			vals = append(vals,
-				to,
-				trimToLimit(e.MailFrom.String(), 255),
-				trimToLimit(e.Subject, 255),
-				body)
-			if body == "redis" {
-				// data already saved in redis
-				vals = append(vals, "")
-			} else if co != nil {
-				// use a compressor (automatically adds e.DeliveryHeader)
-				vals = append(vals, co.String())
-				//co.clear()
+				vals = append(vals,
+					hash,
+					to,
+					e.RemoteAddress,
+					trimToLimit(e.MailFrom.String(), 255),
+					e.TLS)
+
+				stmt := mp.prepareInsertQuery(1, db)
+				mp.doQuery(1, db, stmt, &vals)
+				// continue to the next Processor in the decorator chain
+				return c.Process(e, task)
+			} else if task == TaskValidateRcpt {
+				// if you need to validate the e.Rcpt then change to:
+				if len(e.RcptTo) > 0 {
+					// since this is called each time a recipient is added
+					// validate only the _last_ recipient that was appended
+					last := e.RcptTo[len(e.RcptTo)-1]
+					if len(last.User) > 255 {
+						// TODO what kind of response to send?
+						return NewResult(response.Canned.FailNoSenderDataCmd), NoSuchUser
+					}
+				}
+				return c.Process(e, task)
 			} else {
-				vals = append(vals, e.String())
+				return c.Process(e, task)
 			}
 
-			vals = append(vals,
-				hash,
-				to,
-				e.RemoteAddress,
-				trimToLimit(e.MailFrom.String(), 255),
-				e.TLS)
-
-			stmt := mp.prepareInsertQuery(1, db)
-			mp.doQuery(1, db, stmt, &vals)
-			// continue to the next Processor in the decorator chain
-			return c.Process(e)
 		})
 	}
 }

backends/p_redis.go

@@ -27,7 +27,7 @@ import (
 // ----------------------------------------------------------------------------------
 func init() {
 
-	Processors["redis"] = func() Decorator {
+	processors["redis"] = func() Decorator {
 		return Redis()
 	}
 }
@@ -61,9 +61,9 @@ func Redis() Decorator {
 	var config *RedisProcessorConfig
 	redisClient := &RedisProcessor{}
 	// read the config into RedisProcessorConfig
-	Service.AddInitializer(Initialize(func(backendConfig BackendConfig) error {
+	Svc.AddInitializer(Initialize(func(backendConfig BackendConfig) error {
 		configType := BaseConfig(&RedisProcessorConfig{})
-		bcfg, err := Service.ExtractConfig(backendConfig, configType)
+		bcfg, err := Svc.ExtractConfig(backendConfig, configType)
 		if err != nil {
 			return err
 		}
@@ -75,7 +75,7 @@ func Redis() Decorator {
 		return nil
 	}))
 	// When shutting down
-	Service.AddShutdowner(Shutdown(func() error {
+	Svc.AddShutdowner(Shutdown(func() error {
 		if redisClient.isConnected {
 			return redisClient.conn.Close()
 		}
@@ -85,40 +85,46 @@ func Redis() Decorator {
 	var redisErr error
 
 	return func(c Processor) Processor {
-		return ProcessorFunc(func(e *envelope.Envelope) (BackendResult, error) {
-			hash := ""
-
-			if len(e.Hashes) > 0 {
-				e.QueuedId = e.Hashes[0]
-				hash = e.Hashes[0]
-
-				var stringer fmt.Stringer
-				// a compressor was set
-				if c, ok := e.Info["zlib-compressor"]; ok {
-					stringer = c.(*compressor)
-				} else {
-					stringer = e
-				}
-				redisErr = redisClient.redisConnection(config.RedisInterface)
+		return ProcessorFunc(func(e *envelope.Envelope, task SelectTask) (Result, error) {
+
+			if task == TaskSaveMail {
+				hash := ""
+				if len(e.Hashes) > 0 {
+					e.QueuedId = e.Hashes[0]
+					hash = e.Hashes[0]
+
+					var stringer fmt.Stringer
+					// a compressor was set
+					if c, ok := e.Values["zlib-compressor"]; ok {
+						stringer = c.(*compressor)
+					} else {
+						stringer = e
+					}
+					redisErr = redisClient.redisConnection(config.RedisInterface)
 
-				if redisErr == nil {
-					_, doErr := redisClient.conn.Do("SETEX", hash, config.RedisExpireSeconds, stringer)
-					if doErr != nil {
-						redisErr = doErr
+					if redisErr == nil {
+						_, doErr := redisClient.conn.Do("SETEX", hash, config.RedisExpireSeconds, stringer)
+						if doErr != nil {
+							redisErr = doErr
+						}
+					}
+					if redisErr != nil {
+						Log().WithError(redisErr).Warn("Error while talking to redis")
+						result := NewResult(response.Canned.FailBackendTransaction)
+						return result, redisErr
+					} else {
+						e.Values["redis"] = "redis" // the backend system will know to look in redis for the message data
 					}
-				}
-				if redisErr != nil {
-					Log().WithError(redisErr).Warn("Error while talking to redis")
-					result := NewBackendResult(response.Canned.FailBackendTransaction)
-					return result, redisErr
 				} else {
-					e.Info["redis"] = "redis" // the backend system will know to look in redis for the message data
+					Log().Error("Redis needs a Hash() process before it")
 				}
+
+				return c.Process(e, task)
 			} else {
-				Log().Error("Redis needs a Hash() process before it")
+				// nothing to do for this task
+				return c.Process(e, task)
 			}
 
-			return c.Process(e)
 		})
 	}
 }

backends/processor.go

@@ -4,24 +4,45 @@ import (
 	"github.com/flashmob/go-guerrilla/envelope"
 )
 
+type SelectTask int
+
+const (
+	TaskSaveMail SelectTask = iota
+	TaskValidateRcpt
+)
+
+func (o SelectTask) String() string {
+	switch o {
+	case TaskSaveMail:
+		return "save mail"
+	case TaskValidateRcpt:
+		return "validate recipient"
+	}
+	return "[unnamed task]"
+}
+
+var BackendResultOK = NewResult("200 OK")
+
 // Our processor is defined as something that processes the envelope and returns a result and error
 type Processor interface {
-	Process(*envelope.Envelope) (BackendResult, error)
+	Process(*envelope.Envelope, SelectTask) (Result, error)
 }
 
-// Signature of DoFunc
-type ProcessorFunc func(*envelope.Envelope) (BackendResult, error)
+// Signature of Processor
+type ProcessorFunc func(*envelope.Envelope, SelectTask) (Result, error)
 
 // Make ProcessorFunc will satisfy the Processor interface
-func (f ProcessorFunc) Process(e *envelope.Envelope) (BackendResult, error) {
-	return f(e)
+func (f ProcessorFunc) Process(e *envelope.Envelope, task SelectTask) (Result, error) {
+	// delegate to the anonymous function
+	return f(e, task)
 }
 
 // DefaultProcessor is a undecorated worker that does nothing
-// Notice MockClient has no knowledge of the other decorators that have orthogonal concerns.
+// Notice DefaultProcessor 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
+// (this is the last call in the decorator stack, if it got here, then all is good)
+func (w DefaultProcessor) Process(e *envelope.Envelope, task SelectTask) (Result, error) {
+	return BackendResultOK, nil
 }

backends/validate.go

@@ -0,0 +1,16 @@
+package backends
+
+import (
+	"errors"
+)
+
+type RcptError error
+
+var (
+	NoSuchUser          = RcptError(errors.New("no such iser"))
+	StorageNotAvailable = RcptError(errors.New("storage not available"))
+	StorageTooBusy      = RcptError(errors.New("stoarge too busy"))
+	StorageTimeout      = RcptError(errors.New("stoarge too busy"))
+	QuotaExceeded       = RcptError(errors.New("quota exceeded"))
+	UserSuspended       = RcptError(errors.New("user suspended"))
+)

backends/worker.go

@@ -2,38 +2,49 @@ package backends
 
 import (
 	"errors"
-	"fmt"
 	"runtime/debug"
 )
 
 type Worker struct{}
 
-func (w *Worker) saveMailWorker(saveMailChan chan *savePayload, p Processor, workerId int) {
+func (w *Worker) workDispatcher(workIn chan *workerMsg, validateRcpt chan *workerMsg, p Processor, workerId int) {
 
 	defer func() {
 		if r := recover(); r != nil {
 			// recover form closed channel
-			fmt.Println("Recovered in f", r, string(debug.Stack()))
 			Log().Error("Recovered form panic:", r, string(debug.Stack()))
 		}
 		// close any connections / files
-		Service.Shutdown()
+		Svc.shutdown()
 
 	}()
 	Log().Infof("Save mail worker started (#%d)", workerId)
 	for {
-		payload := <-saveMailChan
-		if payload == nil {
-			Log().Debug("No more saveMailChan payload")
-			return
-		}
-		// process the email here
-		result, _ := p.Process(payload.mail)
-		// if all good
-		if result.Code() < 300 {
-			payload.savedNotify <- &saveStatus{nil, payload.mail.QueuedId}
-		} else {
-			payload.savedNotify <- &saveStatus{errors.New(result.String()), ""}
+		select {
+		case msg := <-workIn:
+			if msg == nil {
+				Log().Debug("No more messages from saveMail")
+				return
+			}
+			// process the email here
+			// TODO we should check the err
+			result, _ := p.Process(msg.mail, TaskSaveMail)
+			if result.Code() < 300 {
+				// if all good, let the gateway know that it was queued
+				msg.notifyMe <- &notifyMsg{nil, msg.mail.QueuedId}
+			} else {
+				// notify the gateway about the error
+				msg.notifyMe <- &notifyMsg{err: errors.New(result.String())}
+			}
+		case msg := <-validateRcpt:
+			_, err := p.Process(msg.mail, TaskValidateRcpt)
+			if err != nil {
+				// validation failed
+				msg.notifyMe <- &notifyMsg{err: err}
+			} else {
+				// all good.
+				msg.notifyMe <- &notifyMsg{err: nil}
+			}
 		}
 
 	}

cmd/guerrillad/serve.go

@@ -7,6 +7,7 @@ import (
 	"github.com/flashmob/go-guerrilla"
 	"github.com/flashmob/go-guerrilla/backends"
 	"github.com/flashmob/go-guerrilla/log"
+	_ "github.com/flashmob/maildiranasaurus"
 	"github.com/spf13/cobra"
 	"io/ioutil"
 	"os"
@@ -97,8 +98,7 @@ func subscribeBackendEvent(event guerrilla.Event, backend backends.Backend, app
 		newBackend, newErr := backends.New(cmdConfig.BackendName, cmdConfig.BackendConfig, logger)
 		if newErr != nil {
 			// this will continue using old backend
-			logger.WithError(newErr).Error("Error while loading the backend %q",
-				cmdConfig.BackendName)
+			logger.WithError(newErr).Error("Error while loading the backend")
 		} else {
 			// swap to the bew backend (assuming old backend was shutdown so it can be safely swapped)
 			backend = newBackend
@@ -134,9 +134,23 @@ func serve(cmd *cobra.Command, args []string) {
 	var backend backends.Backend
 	backend, err = backends.New(cmdConfig.BackendName, cmdConfig.BackendConfig, mainlog)
 	if err != nil {
-		mainlog.WithError(err).Fatalf("Error while loading the backend %q",
-			cmdConfig.BackendName)
+		mainlog.WithError(err).Fatalf("Error while loading the backend")
 	}
+	/*
+		// add our custom processor to the backend
+		backends.Service.AddProcessor("MailDir", maildiranasaurus.MaildirProcessor)
+		config := guerrilla.AppConfig {
+			LogFile: log.OutputStderr,
+			LogLevel: "info",
+			AllowedHosts: []string{"example.com"},
+			PidFile: "./pidfile.pid",
+			Servers: []guerrilla.ServerConfig{
+
+			}
+		}
+	*/
+
+	//	g := guerrilla.NewSMTPD{config: cmdConfig}
 
 	app, err := guerrilla.New(&cmdConfig.AppConfig, backend, mainlog)
 	if err != nil {

cmd/guerrillad/serve_test.go

@@ -3,7 +3,6 @@ package main
 import (
 	"crypto/tls"
 	"encoding/json"
-	"fmt"
 	"github.com/flashmob/go-guerrilla"
 	"github.com/flashmob/go-guerrilla/backends"
 	"github.com/flashmob/go-guerrilla/log"
@@ -996,7 +995,7 @@ func TestSetTimeoutEvent(t *testing.T) {
 	fd, _ := os.Open("../../tests/testlog")
 	if read, err := ioutil.ReadAll(fd); err == nil {
 		logOutput := string(read)
-		fmt.Println(logOutput)
+		//fmt.Println(logOutput)
 		if i := strings.Index(logOutput, "i/o timeout"); i < 0 {
 			t.Error("Connection to 127.0.0.1:2552 didn't timeout as expected")
 		}

envelope/envelope.go

@@ -51,8 +51,8 @@ type Envelope struct {
 	TLS bool
 	// Header stores the results from ParseHeaders()
 	Header textproto.MIMEHeader
-	// Hold the information generated when processing the envelope by the backend
-	Info map[string]interface{}
+	// Values hold the values generated when processing the envelope by the backend
+	Values map[string]interface{}
 	// Hashes of each email on the rcpt
 	Hashes []string
 	// additional delivery header that may be added
@@ -65,7 +65,7 @@ func NewEnvelope(remoteAddr string, clientID uint64) *Envelope {
 
 	return &Envelope{
 		RemoteAddress: remoteAddr,
-		Info:          make(map[string]interface{}),
+		Values:        make(map[string]interface{}),
 		QueuedId:      queuedID(clientID),
 	}
 }
@@ -140,10 +140,20 @@ func (e *Envelope) Reseed(remoteAddr string, clientID uint64) {
 	e.TLS = false
 	e.Hashes = make([]string, 0)
 	e.DeliveryHeader = ""
-	e.Info = make(map[string]interface{})
+	e.Values = make(map[string]interface{})
 	e.QueuedId = queuedID(clientID)
 }
 
+func (e *Envelope) PushRcpt(addr EmailAddress) {
+	e.RcptTo = append(e.RcptTo, addr)
+}
+
+func (e *Envelope) PopRcpt() EmailAddress {
+	ret := e.RcptTo[len(e.RcptTo)-1]
+	e.RcptTo = e.RcptTo[:len(e.RcptTo)-1]
+	return ret
+}
+
 var mimeRegex, _ = regexp.Compile(`=\?(.+?)\?([QBqp])\?(.+?)\?=`)
 
 // Decode strings in Mime header format

guerrilla.go

@@ -297,7 +297,7 @@ func (g *guerrilla) subscribeEvents() {
 			var l log.Logger
 			if l, err = log.GetLogger(sc.LogFile); err == nil {
 				g.storeMainlog(l)
-				backends.Service.StoreMainlog(l)
+				backends.Svc.StoreMainlog(l)
 				// it will change to the new logger on the next accepted client
 				server.logStore.Store(l)
 
@@ -402,5 +402,5 @@ func (g *guerrilla) Shutdown() {
 func (g *guerrilla) SetLogger(l log.Logger) {
 	l.SetLevel(g.Config.LogLevel)
 	g.storeMainlog(l)
-	backends.Service.StoreMainlog(l)
+	backends.Svc.StoreMainlog(l)
 }

response/enhanced.go

@@ -134,6 +134,7 @@ type Responses struct {
 	FailBackendNotRunning        string
 	FailBackendTransaction       string
 	FailBackendTimeout           string
+	FailRcptCmd                  string
 
 	// The 400's
 	ErrorTooManyRecipients string
@@ -155,7 +156,6 @@ type Responses struct {
 // Called automatically during package load to build up the Responses struct
 func init() {
 
-	// There's even a Wikipedia page for canned responses: https://en.wikipedia.org/wiki/Canned_response
 	Canned = Responses{}
 
 	Canned.FailLineTooLong = (&Response{
@@ -337,6 +337,13 @@ func init() {
 		Comment:      "Error: transaction timeout",
 	}).String()
 
+	Canned.FailRcptCmd = (&Response{
+		EnhancedCode: BadDestinationMailboxAddress,
+		BasicCode:    550,
+		Class:        ClassPermanentFailure,
+		Comment:      "User unknown in local recipient table",
+	}).String()
+
 }
 
 // DefaultMap contains defined default codes (RfC 3463)

server.go

@@ -409,8 +409,15 @@ func (server *server) handleClient(client *client) {
 					if !server.allowsHost(to.Host) {
 						client.sendResponse(response.Canned.ErrorRelayDenied, to.Host)
 					} else {
-						client.RcptTo = append(client.RcptTo, to)
-						client.sendResponse(response.Canned.SuccessRcptCmd)
+						client.PushRcpt(to)
+						rcptError := server.backend.ValidateRcpt(client.Envelope)
+						if rcptError != nil {
+							client.PopRcpt()
+							client.sendResponse(response.Canned.FailRcptCmd)
+						} else {
+							client.sendResponse(response.Canned.SuccessRcptCmd)
+						}
+
 					}
 				}