multiple backends, needs debugging

api.go

@@ -13,9 +13,9 @@ import (
 // Daemon provides a convenient API when using go-guerrilla as a package in your Go project.
 // Is's facade for Guerrilla, AppConfig, backends.Backend and log.Logger
 type Daemon struct {
-	Config  *AppConfig
-	Logger  log.Logger
-	Backend backends.Backend
+	Config   *AppConfig
+	Logger   log.Logger
+	Backends []backends.Backend
 
 	// Guerrilla will be managed through the API
 	g Guerrilla
@@ -51,13 +51,7 @@ func (d *Daemon) Start() (err error) {
 				return err
 			}
 		}
-		if d.Backend == nil {
-			d.Backend, err = backends.New(d.Config.BackendConfig, d.Logger)
-			if err != nil {
-				return err
-			}
-		}
-		d.g, err = New(d.Config, d.Backend, d.Logger)
+		d.g, err = New(d.Config, d.Logger, d.Backends...)
 		if err != nil {
 			return err
 		}
@@ -72,7 +66,6 @@ func (d *Daemon) Start() (err error) {
 		if err := d.resetLogger(); err == nil {
 			d.Log().Infof("main log configured to %s", d.Config.LogFile)
 		}
-
 	}
 	return err
 }
@@ -155,7 +148,7 @@ func (d *Daemon) ReopenLogs() error {
 	if d.Config == nil {
 		return errors.New("d.Config nil")
 	}
-	d.Config.EmitLogReopenEvents(d.g)
+	d.Config.emitLogReopenEvents(d.g)
 	return nil
 }
 
@@ -217,7 +210,9 @@ func (d *Daemon) configureDefaults() error {
 	if err != nil {
 		return err
 	}
-	if d.Backend == nil {
+	if d.Backends == nil {
+		d.Backends = make([]backends.Backend, 0)
+		// the config will be used to make backends
 		err = d.Config.setBackendDefaults()
 		if err != nil {
 			return err

api_test.go

@@ -102,11 +102,22 @@ func TestSMTPCustomBackend(t *testing.T) {
 	}
 	cfg.Servers = append(cfg.Servers, sc)
 	bcfg := backends.BackendConfig{
-		"save_workers_size":  3,
-		"save_process":       "HeadersParser|Header|Hasher|Debugger",
-		"log_received_mails": true,
-		"primary_mail_host":  "example.com",
+		"processors": {
+			"debugger": {
+				"log_received_mails": true,
+			},
+			"header": {
+				"primary_mail_host": "example.com",
+			},
+		},
+		"gateways": {
+			"default": {
+				"save_workers_size": 3,
+				"save_process":      "HeadersParser|Header|Hasher|Debugger",
+			},
+		},
 	}
+
 	cfg.BackendConfig = bcfg
 	d := Daemon{Config: cfg}
 
@@ -126,12 +137,20 @@ func TestSMTPLoadFile(t *testing.T) {
     "log_level" : "debug",
     "pid_file" : "tests/go-guerrilla.pid",
     "allowed_hosts": ["spam4.me","grr.la"],
-    "backend_config" :
-        {
-            "log_received_mails" : true,
-            "save_process": "HeadersParser|Header|Hasher|Debugger",
-            "save_workers_size":  3
-        },
+	"backend" : {
+		"processors" : {
+			"debugger" {
+				"log_received_mails" : true
+			}
+		},
+		"gateways" : {
+			"default" : {
+				"save_workers_size" : 3,
+    			"save_process": "HeadersParser|Header|Hasher|Debugger"
+			}
+		}
+	},
+
     "servers" : [
         {
             "is_enabled" : true,
@@ -156,12 +175,19 @@ func TestSMTPLoadFile(t *testing.T) {
     "log_level" : "debug",
     "pid_file" : "tests/go-guerrilla2.pid",
     "allowed_hosts": ["spam4.me","grr.la"],
-    "backend_config" :
-        {
-            "log_received_mails" : true,
-            "save_process": "HeadersParser|Header|Hasher|Debugger",
-            "save_workers_size":  3
-        },
+    "backend" : {
+		"processors" : {
+			"debugger" {
+				"log_received_mails" : true
+			}
+		},
+		"gateways" : {
+			"default" : {
+				"save_workers_size" : 3,
+    			"save_process": "HeadersParser|Header|Hasher|Debugger"
+			}
+		}
+	},
     "servers" : [
         {
             "is_enabled" : true,
@@ -455,8 +481,12 @@ func TestSetAddProcessor(t *testing.T) {
 		LogFile:      "tests/testlog",
 		AllowedHosts: []string{"grr.la"},
 		BackendConfig: backends.BackendConfig{
-			"save_process":     "HeadersParser|Debugger|FunkyLogger",
-			"validate_process": "FunkyLogger",
+			"gateways": {
+				"default": {
+					"save_process":     "HeadersParser|Debugger|FunkyLogger",
+					"validate_process": "FunkyLogger",
+				},
+			},
 		},
 	}
 	d := Daemon{Config: cfg}
@@ -597,8 +627,12 @@ func TestReloadConfig(t *testing.T) {
 		LogFile:      "tests/testlog",
 		AllowedHosts: []string{"grr.la"},
 		BackendConfig: backends.BackendConfig{
-			"save_process":     "HeadersParser|Debugger|FunkyLogger",
-			"validate_process": "FunkyLogger",
+			"gateways": {
+				"default": {
+					"save_process":     "HeadersParser|Debugger|FunkyLogger",
+					"validate_process": "FunkyLogger",
+				},
+			},
 		},
 	}
 	// Look mom, reloading the config without shutting down!
@@ -628,8 +662,12 @@ func TestPubSubAPI(t *testing.T) {
 		LogFile:      "tests/testlog",
 		AllowedHosts: []string{"grr.la"},
 		BackendConfig: backends.BackendConfig{
-			"save_process":     "HeadersParser|Debugger|FunkyLogger",
-			"validate_process": "FunkyLogger",
+			"gateways": {
+				"default": {
+					"save_process":     "HeadersParser|Debugger|FunkyLogger",
+					"validate_process": "FunkyLogger",
+				},
+			},
 		},
 	}
 
@@ -765,10 +803,15 @@ func TestCustomBackendResult(t *testing.T) {
 		LogFile:      "tests/testlog",
 		AllowedHosts: []string{"grr.la"},
 		BackendConfig: backends.BackendConfig{
-			"save_process":     "HeadersParser|Debugger|Custom",
-			"validate_process": "Custom",
+			"gateways": {
+				"default": {
+					"save_process":     "HeadersParser|Debugger|Custom",
+					"validate_process": "Custom",
+				},
+			},
 		},
 	}
+
 	d := Daemon{Config: cfg}
 	d.AddProcessor("Custom", customBackend2)
 
@@ -806,8 +849,12 @@ func TestStreamProcessor(t *testing.T) {
 		LogFile:      "tests/testlog",
 		AllowedHosts: []string{"grr.la"},
 		BackendConfig: backends.BackendConfig{
-			"save_process":        "HeadersParser|Debugger",
-			"stream_save_process": "Header|headersparser|compress|Decompress|debug",
+			"gateways": {
+				"default": {
+					"save_process":        "HeadersParser|Debugger",
+					"stream_save_process": "Header|headersparser|compress|Decompress|debug",
+				},
+			},
 		},
 	}
 	d := Daemon{Config: cfg}
@@ -1113,8 +1160,12 @@ func TestStreamMimeProcessor(t *testing.T) {
 		LogFile:      "tests/testlog",
 		AllowedHosts: []string{"grr.la"},
 		BackendConfig: backends.BackendConfig{
-			"save_process":        "HeadersParser|Debugger",
-			"stream_save_process": "mimeanalyzer|headersparser|compress|Decompress|debug",
+			"gateways": {
+				"default": {
+					"save_process":        "HeadersParser|Debugger",
+					"stream_save_process": "mimeanalyzer|headersparser|compress|Decompress|debug",
+				},
+			},
 		},
 	}
 	d := Daemon{Config: cfg}
@@ -1253,10 +1304,19 @@ func TestStreamChunkSaver(t *testing.T) {
 		LogFile:      "tests/testlog",
 		AllowedHosts: []string{"grr.la"},
 		BackendConfig: backends.BackendConfig{
-			"stream_save_process":       "mimeanalyzer|chunksaver",
-			"chunksaver_chunk_size":     1024 * 32,
-			"stream_buffer_size":        1024 * 16,
-			"chunksaver_storage_engine": "memory",
+			"stream_processors": {
+				"chunksaver": {
+					"chunksaver_chunk_size":     1024 * 32,
+					"stream_buffer_size":        1024 * 16,
+					"chunksaver_storage_engine": "memory",
+				},
+			},
+			"gateways": {
+				"default": {
+					"save_process":        "HeadersParser|Debugger",
+					"stream_save_process": "mimeanalyzer|chunksaver",
+				},
+			},
 		},
 	}
 

backends/backend.go

@@ -31,6 +31,8 @@ func init() {
 	Streamers = make(map[string]StreamProcessorConstructor)
 }
 
+const DefaultGateway = "default"
+
 type ProcessorConstructor func() Decorator
 
 type StreamProcessorConstructor func() *StreamDecorator
@@ -56,13 +58,10 @@ type Backend interface {
 	Shutdown() error
 	// Start Starts a backend that has been initialized
 	Start() error
+	// returns the name of the backend
+	Name() string
 }
 
-type BackendConfig map[string]interface{}
-
-// All config structs extend from this
-type BaseConfig interface{}
-
 type notifyMsg struct {
 	err      error
 	queuedID string
@@ -262,12 +261,21 @@ func (s *service) AddStreamProcessor(name string, p StreamProcessorConstructor)
 }
 
 // extractConfig loads the backend config. It has already been unmarshalled
-// configData contains data from the main config file's "backend_config" value
+// "group" refers
+// cfg contains data from the main config file's "backend_config" value
 // configType is a Processor's specific config value.
 // 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 (s *service) ExtractConfig(configData BackendConfig, configType BaseConfig) (interface{}, error) {
+func (s *service) ExtractConfig(ns configNameSpace, group string, cfg BackendConfig, configType BaseConfig) (interface{}, error) {
+	group = strings.ToLower(group)
+
+	var configData ConfigGroup
+	if v, ok := cfg[ns.String()][group]; ok {
+		configData = v
+	} else {
+		return configData, nil
+	}
 	// Use reflection so that we can provide a nice error message
 	v := reflect.ValueOf(configType).Elem() // so that we can set the values
 	//m := reflect.ValueOf(configType).Elem()

backends/backend_test.go

@@ -0,0 +1,16 @@
+package backends
+
+import (
+	"encoding/json"
+	"fmt"
+	"testing"
+)
+
+func TestSetProcessorValue(t *testing.T) {
+
+	var test BackendConfig
+	test = make(map[string]map[string]interface{}, 0)
+	test.SetValue("processors", "ABC", "key", "value")
+	out, _ := json.MarshalIndent(test, "", "   ")
+	fmt.Println(string(out))
+}

backends/config.go

@@ -0,0 +1,193 @@
+package backends
+
+import (
+	"errors"
+	"fmt"
+	"reflect"
+	"strings"
+)
+
+type ConfigGroup map[string]interface{}
+type BackendConfig map[string]map[string]ConfigGroup
+
+func (c *BackendConfig) SetValue(ns configNameSpace, name string, key string, value interface{}) {
+	nsKey := ns.String()
+	if *c == nil {
+		*c = make(BackendConfig, 0)
+	}
+	if (*c)[nsKey] == nil {
+		(*c)[nsKey] = make(map[string]ConfigGroup)
+	}
+	if (*c)[nsKey][name] == nil {
+		(*c)[nsKey][name] = make(ConfigGroup)
+	}
+	(*c)[nsKey][name] = map[string]interface{}{key: value}
+}
+
+func (c *BackendConfig) GetValue(ns configNameSpace, name string, key string) interface{} {
+	nsKey := ns.String()
+	if (*c)[nsKey] == nil {
+		return nil
+	}
+	if (*c)[nsKey][name] == nil {
+		return nil
+	}
+	if v, ok := (*c)[nsKey][name][key]; ok {
+		return &v
+	}
+	return nil
+}
+
+type configNameSpace int
+
+const (
+	ConfigProcessors configNameSpace = iota
+	ConfigStreamProcessors
+	ConfigGateways
+)
+
+func (o configNameSpace) String() string {
+	switch o {
+	case ConfigProcessors:
+		return "processors"
+	case ConfigStreamProcessors:
+		return "stream_processors"
+	case ConfigGateways:
+		return "gateways"
+	}
+	return "unknown"
+}
+
+// All config structs extend from this
+type BaseConfig interface{}
+
+type stackConfigExpression struct {
+	alias string
+	name  string
+}
+
+type stackConfig struct {
+	list     []stackConfigExpression
+	notFound func(s string) error
+}
+
+func NewStackConfig(config string) (ret *stackConfig) {
+	ret = new(stackConfig)
+	cfg := strings.ToLower(strings.TrimSpace(config))
+	if cfg == "" {
+		return
+	}
+	items := strings.Split(cfg, "|")
+	ret.list = make([]stackConfigExpression, len(items))
+	pos := 0
+	for i := range items {
+		pos = len(items) - 1 - i // reverse order, since decorators are stacked
+		ret.list[pos] = stackConfigExpression{alias: "", name: items[pos]}
+	}
+	return ret
+}
+
+func newStackProcessorConfig(config string) (ret *stackConfig) {
+	ret = NewStackConfig(config)
+	ret.notFound = func(s string) error {
+		return errors.New(fmt.Sprintf("processor [%s] not found", s))
+	}
+	return ret
+}
+
+func newStackStreamProcessorConfig(config string) (ret *stackConfig) {
+	ret = NewStackConfig(config)
+	ret.notFound = func(s string) error {
+		return errors.New(fmt.Sprintf("stream processor [%s] not found", s))
+	}
+	return ret
+}
+
+// Changes returns a list of gateways whose config changed
+func (c BackendConfig) Changes(oldConfig BackendConfig) (changed, added, removed map[string]bool) {
+	// check the processors if changed
+
+	changed = make(map[string]bool, 0)
+	added = make(map[string]bool, 0)
+	removed = make(map[string]bool, 0)
+
+	changedProcessors := changedConfigGroups(oldConfig[string(ConfigProcessors)], c[string(ConfigProcessors)])
+	changedStreamProcessors := changedConfigGroups(oldConfig[string(ConfigProcessors)], c[string(ConfigProcessors)])
+	configType := BaseConfig(&GatewayConfig{})
+
+	// go through all the gateway configs,
+	// make a list of all the ones that have processors whose config had changed
+	for key, _ := range c[string(ConfigGateways)] {
+		e, _ := Svc.ExtractConfig(ConfigGateways, key, c, configType)
+		bcfg := e.(*GatewayConfig)
+		config := NewStackConfig(bcfg.SaveProcess)
+		for _, v := range config.list {
+			if _, ok := changedProcessors[v.name]; ok {
+				changed[key] = true
+			}
+		}
+		config = NewStackConfig(bcfg.StreamSaveProcess)
+		for _, v := range config.list {
+			if _, ok := changedStreamProcessors[v.name]; ok {
+				changed[key] = true
+			}
+		}
+
+		if o, ok := oldConfig[key]; ok {
+			delete(oldConfig, key)
+			if !reflect.DeepEqual(c[key], o) {
+				// whats changed
+				changed[key] = true
+			}
+		} else {
+			// whats been added
+			added[key] = true
+		}
+	}
+
+	// whats been removed
+	for p := range oldConfig {
+		removed[p] = true
+	}
+
+	return
+
+}
+
+func changedConfigGroups(old map[string]ConfigGroup, new map[string]ConfigGroup) map[string]bool {
+	diff, added, removed := compareConfigGroup(old, new)
+	var all []string
+	all = append(all, diff...)
+	all = append(all, removed...)
+	all = append(all, added...)
+	changed := make(map[string]bool, 0)
+	for p := range all {
+		changed[all[p]] = true
+	}
+	return changed
+}
+
+// compareConfigGroup compares two config groups
+// returns a list of keys that changed, been added or removed to new
+func compareConfigGroup(old map[string]ConfigGroup, new map[string]ConfigGroup) (diff, added, removed []string) {
+	diff = make([]string, 0)
+	added = make([]string, 0)
+	removed = make([]string, 0)
+	for p := range new {
+		if o, ok := old[p]; ok {
+			delete(old, p)
+			if !reflect.DeepEqual(new[p], o) {
+				// whats changed
+				diff = append(diff, p)
+			}
+		} else {
+			// whats been added
+			added = append(added, p)
+		}
+	}
+	// whats been removed
+	for p := range old {
+		removed = append(removed, p)
+	}
+	return
+}

backends/gateway.go

@@ -8,12 +8,10 @@ import (
 	"sync"
 	"time"
 
-	"runtime/debug"
-	"strings"
-
 	"github.com/flashmob/go-guerrilla/log"
 	"github.com/flashmob/go-guerrilla/mail"
 	"github.com/flashmob/go-guerrilla/response"
+	"runtime/debug"
 )
 
 var ErrProcessorNotFound error
@@ -23,6 +21,8 @@ var ErrProcessorNotFound error
 // 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 {
+	// name is the name of the gateway given in the config
+	name string
 	// channel for distributing envelopes to workers
 	conveyor chan *workerMsg
 
@@ -152,17 +152,15 @@ func (s backendState) String() string {
 
 // New makes a new default BackendGateway backend, and initializes it using
 // backendConfig and stores the logger
-func New(backendConfig BackendConfig, l log.Logger) (Backend, error) {
+func New(name string, backendConfig BackendConfig, l log.Logger) (Backend, error) {
 	Svc.SetMainlog(l)
-	gateway := &BackendGateway{}
+	gateway := &BackendGateway{name: name}
 	err := gateway.Initialize(backendConfig)
 	if err != nil {
 		return nil, fmt.Errorf("error while initializing the backend: %s", err)
 	}
-	// keep the config known to be good.
+	// keep the a copy of the config
 	gateway.config = backendConfig
-
-	b = Backend(gateway)
 	return b, nil
 }
 
@@ -182,6 +180,10 @@ func (w *workerMsg) reset(e *mail.Envelope, task SelectTask) {
 	w.task = task
 }
 
+func (gw *BackendGateway) Name() string {
+	return gw.name
+}
+
 // Process distributes an envelope to one of the backend workers with a TaskSaveMail task
 func (gw *BackendGateway) Process(e *mail.Envelope) Result {
 	if gw.State != BackendStateRunning {
@@ -370,18 +372,15 @@ func (gw *BackendGateway) Reinitialize() error {
 // This function uses the config value save_process or validate_process to figure out which Decorator to use
 func (gw *BackendGateway) newStack(stackConfig string) (Processor, error) {
 	var decorators []Decorator
-	cfg := strings.ToLower(strings.TrimSpace(stackConfig))
-	if len(cfg) == 0 {
+	c := newStackProcessorConfig(stackConfig)
+	if len(c.list) == 0 {
 		return NoopProcessor{}, nil
 	}
-	items := strings.Split(cfg, "|")
-	for i := range items {
-		name := items[len(items)-1-i] // reverse order, since decorators are stacked
-		if makeFunc, ok := processors[name]; ok {
+	for i := range c.list {
+		if makeFunc, ok := processors[c.list[i].name]; ok {
 			decorators = append(decorators, makeFunc())
 		} else {
-			ErrProcessorNotFound = fmt.Errorf("processor [%s] not found", name)
-			return nil, ErrProcessorNotFound
+			return nil, c.notFound(c.list[i].name)
 		}
 	}
 	// build the call-stack of decorators
@@ -391,21 +390,17 @@ func (gw *BackendGateway) newStack(stackConfig string) (Processor, error) {
 
 func (gw *BackendGateway) newStreamStack(stackConfig string) (streamer, error) {
 	var decorators []*StreamDecorator
-	cfg := strings.ToLower(strings.TrimSpace(stackConfig))
-	if len(cfg) == 0 {
+	c := newStackStreamProcessorConfig(stackConfig)
+	if len(c.list) == 0 {
 		return streamer{NoopStreamProcessor{}, decorators}, nil
 	}
-	items := strings.Split(cfg, "|")
-	for i := range items {
-		name := items[len(items)-1-i] // reverse order, since decorators are stacked
-		if makeFunc, ok := Streamers[name]; ok {
+	for i := range c.list {
+		if makeFunc, ok := Streamers[c.list[i].name]; ok {
 			decorators = append(decorators, makeFunc())
 		} else {
-			ErrProcessorNotFound = errors.New(fmt.Sprintf("stream processor [%s] not found", name))
-			return streamer{nil, decorators}, ErrProcessorNotFound
+			return streamer{nil, decorators}, c.notFound(c.list[i].name)
 		}
 	}
-
 	// build the call-stack of decorators
 	sp, decorators := DecorateStream(&DefaultStreamProcessor{}, decorators)
 	return streamer{sp, decorators}, nil
@@ -417,7 +412,13 @@ func (gw *BackendGateway) loadConfig(cfg BackendConfig) error {
 	// Note: treat config values as immutable
 	// if you need to change a config value, change in the file then
 	// send a SIGHUP
-	bcfg, err := Svc.ExtractConfig(cfg, configType)
+	if gw.name == "" {
+		gw.name = DefaultGateway
+	}
+	if _, ok := cfg["gateways"][gw.name]; !ok {
+		return errors.New("no such gateway configured: " + gw.name)
+	}
+	bcfg, err := Svc.ExtractConfig(ConfigGateways, gw.name, cfg, configType)
 	if err != nil {
 		return err
 	}

backends/p_debugger.go

@@ -32,7 +32,8 @@ func Debugger() Decorator {
 	var config *debuggerConfig
 	initFunc := InitializeWith(func(backendConfig BackendConfig) error {
 		configType := BaseConfig(&debuggerConfig{})
-		bcfg, err := Svc.ExtractConfig(backendConfig, configType)
+		bcfg, err := Svc.ExtractConfig(
+			ConfigProcessors, defaultProcessor, backendConfig, configType)
 		if err != nil {
 			return err
 		}
@@ -57,7 +58,6 @@ func Debugger() Decorator {
 					if config.SleepSec == 1 {
 						panic("panic on purpose")
 					}
-
 				}
 
 				// continue to the next Processor in the decorator stack

backends/p_guerrilla_db_redis.go

@@ -68,7 +68,7 @@ type guerrillaDBAndRedisConfig struct {
 // Now we need to convert each type and copy into the guerrillaDBAndRedisConfig struct
 func (g *GuerrillaDBAndRedisBackend) loadConfig(backendConfig BackendConfig) (err error) {
 	configType := BaseConfig(&guerrillaDBAndRedisConfig{})
-	bcfg, err := Svc.ExtractConfig(backendConfig, configType)
+	bcfg, err := Svc.ExtractConfig(ConfigProcessors, "guerrillaredisdb", backendConfig, configType)
 	if err != nil {
 		return err
 	}
@@ -360,7 +360,8 @@ func GuerrillaDbRedis() Decorator {
 	Svc.AddInitializer(InitializeWith(func(backendConfig BackendConfig) error {
 
 		configType := BaseConfig(&guerrillaDBAndRedisConfig{})
-		bcfg, err := Svc.ExtractConfig(backendConfig, configType)
+		bcfg, err := Svc.ExtractConfig(
+			ConfigProcessors, "guerrillaredisdb", backendConfig, configType)
 		if err != nil {
 			return err
 		}

backends/p_header.go

@@ -38,7 +38,7 @@ func Header() Decorator {
 
 	Svc.AddInitializer(InitializeWith(func(backendConfig BackendConfig) error {
 		configType := BaseConfig(&HeaderConfig{})
-		bcfg, err := Svc.ExtractConfig(backendConfig, configType)
+		bcfg, err := Svc.ExtractConfig(ConfigProcessors, "header", backendConfig, configType)
 		if err != nil {
 			return err
 		}

backends/p_redis.go

@@ -61,7 +61,7 @@ func Redis() Decorator {
 	// read the config into RedisProcessorConfig
 	Svc.AddInitializer(InitializeWith(func(backendConfig BackendConfig) error {
 		configType := BaseConfig(&RedisProcessorConfig{})
-		bcfg, err := Svc.ExtractConfig(backendConfig, configType)
+		bcfg, err := Svc.ExtractConfig(ConfigProcessors, "redis", backendConfig, configType)
 		if err != nil {
 			return err
 		}

backends/p_sql.go

@@ -194,7 +194,7 @@ func SQL() Decorator {
 	// open the database connection (it will also check if we can select the table)
 	Svc.AddInitializer(InitializeWith(func(backendConfig BackendConfig) error {
 		configType := BaseConfig(&SQLProcessorConfig{})
-		bcfg, err := Svc.ExtractConfig(backendConfig, configType)
+		bcfg, err := Svc.ExtractConfig(ConfigProcessors, "sql", backendConfig, configType)
 		if err != nil {
 			return err
 		}

backends/p_sql_test.go

@@ -31,14 +31,15 @@ func TestSQL(t *testing.T) {
 		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)
+	cfg := BackendConfig{}
+
+	cfg.SetValue(ConfigGateways, DefaultGateway, "save_process", "sql")
+	cfg.SetValue(ConfigProcessors, "sql", "mail_table", *mailTableFlag)
+	cfg.SetValue(ConfigProcessors, "sql", "primary_mail_host", "example.com")
+	cfg.SetValue(ConfigProcessors, "sql", "sql_driver", *sqlDriverFlag)
+	cfg.SetValue(ConfigProcessors, "sql", "sql_dsn", *sqlDSNFlag)
+
+	backend, err := New(DefaultGateway, cfg, logger)
 	if err != nil {
 		t.Fatal("new backend:", err)
 	}

backends/s_header.go

@@ -63,7 +63,8 @@ func StreamHeader() *StreamDecorator {
 
 	Svc.AddInitializer(InitializeWith(func(backendConfig BackendConfig) error {
 		configType := BaseConfig(&HeaderConfig{})
-		bcfg, err := Svc.ExtractConfig(backendConfig, configType)
+		bcfg, err := Svc.ExtractConfig(
+			ConfigStreamProcessors, "header", backendConfig, configType)
 		if err != nil {
 			return err
 		}

backends/s_transformer.go

@@ -225,7 +225,8 @@ func Transformer() *StreamDecorator {
 
 	Svc.AddInitializer(InitializeWith(func(backendConfig BackendConfig) error {
 		configType := BaseConfig(&HeaderConfig{})
-		bcfg, err := Svc.ExtractConfig(backendConfig, configType)
+		bcfg, err := Svc.ExtractConfig(
+			ConfigStreamProcessors, "transformer", backendConfig, configType)
 		if err != nil {
 			return err
 		}

chunk/processor.go

@@ -90,7 +90,8 @@ func Chunksaver() *backends.StreamDecorator {
 			backends.Svc.AddInitializer(backends.InitializeWith(func(backendConfig backends.BackendConfig) error {
 
 				configType := backends.BaseConfig(&Config{})
-				bcfg, err := backends.Svc.ExtractConfig(backendConfig, configType)
+				bcfg, err := backends.Svc.ExtractConfig(
+					backends.ConfigStreamProcessors, "chunksaver", backendConfig, configType)
 				if err != nil {
 					return err
 				}

chunk/reader.go

@@ -64,7 +64,7 @@ type cachedChunks struct {
 	db        Storage
 }
 
-// chunkCachePreload controls how many to pre-load in the
+// chunkCachePreload controls how many chunks to pre-load in the cache
 const chunkCachePreload = 2
 
 // warm allocates the chunk cache, and gets the first few and stores them in the cache

chunk/store_sql.go

@@ -182,7 +182,7 @@ func (s *StoreSQL) CloseMessage(mailID uint64, size int64, partsInfo *PartsInfo,
 // Initialize loads the specific database config, connects to the db, prepares statements
 func (s *StoreSQL) Initialize(cfg backends.BackendConfig) error {
 	configType := backends.BaseConfig(&sqlConfig{})
-	bcfg, err := backends.Svc.ExtractConfig(cfg, configType)
+	bcfg, err := backends.Svc.ExtractConfig(backends.ConfigStreamProcessors, "chunksaver", cfg, configType)
 	if err != nil {
 		return err
 	}

cmd/guerrillad/serve_test.go

@@ -37,11 +37,19 @@ var configJsonA = `
       "guerrillamail.net",
       "guerrillamail.org"
     ],
-    "backend_config": {
-    	"save_workers_size" : 1,
-    	"save_process": "HeadersParser|Debugger",
-        "log_received_mails": true
-    },
+	"backend" : {
+		"processors" : {
+			"debugger" {
+				"log_received_mails" : true
+			}
+		},
+		"gateways" : {
+			"default" : {
+				"save_workers_size" : 1,
+    			"save_process": "HeadersParser|Debugger"
+			}
+		}
+	},
     "servers" : [
         {
             "is_enabled" : true,
@@ -90,11 +98,19 @@ var configJsonB = `
       "guerrillamail.net",
       "guerrillamail.org"
     ],
-    "backend_config": {
-    	"save_workers_size" : 1,
-    	"save_process": "HeadersParser|Debugger",
-        "log_received_mails": false
-    },
+    "backend" : {
+		"processors" : {
+			"debugger" {
+				"log_received_mails" : false
+			}
+		},
+		"gateways" : {
+			"default" : {
+				"save_workers_size" : 1,
+    			"save_process": "HeadersParser|Debugger"
+			}
+		}
+	},
     "servers" : [
         {
             "is_enabled" : true,
@@ -128,19 +144,24 @@ var configJsonC = `
       "guerrillamail.net",
       "guerrillamail.org"
     ],
-    "backend_config" :
-        {
-            "sql_driver": "mysql",
-            "sql_dsn": "root:ok@tcp(127.0.0.1:3306)/gmail_mail?readTimeout=10s&writeTimeout=10s",
-            "mail_table":"new_mail",
-            "redis_interface" : "127.0.0.1:6379",
-            "redis_expire_seconds" : 7200,
-            "save_workers_size" : 3,
-            "primary_mail_host":"sharklasers.com",
-            "save_workers_size" : 1,
-	    	"save_process": "HeadersParser|Debugger",
-	    	"log_received_mails": true
-        },
+	"backend" : {
+		"processors" : {
+			"debugger" {
+				"log_received_mails" : true
+			},
+            "sql" : {
+ 				"sql_dsn": "root:ok@tcp(127.0.0.1:3306)/gmail_mail?readTimeout=10s&writeTimeout=10s",
+            	"mail_table":"new_mail",
+				"primary_mail_host":"sharklasers.com",
+			}
+		},
+		"gateways" : {
+			"default" : {
+				"save_workers_size" : 3,
+    			"save_process": "HeadersParser|Debugger"
+			}
+		}
+	},
     "servers" : [
         {
             "is_enabled" : true,
@@ -189,11 +210,19 @@ var configJsonD = `
       "guerrillamail.net",
       "guerrillamail.org"
     ],
-    "backend_config": {
-        "save_workers_size" : 1,
-    	"save_process": "HeadersParser|Debugger",
-        "log_received_mails": false
-    },
+	"backend" : {
+		"processors" : {
+			"debugger" {
+				"log_received_mails" : false
+			}
+		},
+		"gateways" : {
+			"default" : {
+				"save_workers_size" : 1,
+    			"save_process": "HeadersParser|Debugger"
+			}
+		}
+	},
     "servers" : [
         {
             "is_enabled" : true,
@@ -242,19 +271,31 @@ var configJsonE = `
       "guerrillamail.net",
       "guerrillamail.org"
     ],
-    "backend_config" :
-        {
-            "save_process_old": "HeadersParser|Debugger|Hasher|Header|Compressor|Redis|MySql",
-            "save_process": "GuerrillaRedisDB",
-            "log_received_mails" : true,
-            "sql_driver": "mysql",
-            "sql_dsn": "root:secret@tcp(127.0.0.1:3306)/gmail_mail?readTimeout=10s&writeTimeout=10s",
-            "mail_table":"new_mail",
-            "redis_interface" : "127.0.0.1:6379",
-            "redis_expire_seconds" : 7200,
-            "save_workers_size" : 3,
-            "primary_mail_host":"sharklasers.com"
-        },
+	"backend" : {
+		"processors" : {
+			"debugger" {
+				"log_received_mails" : true
+			},
+            "sql" : {
+ 				"sql_dsn": "root:ok@tcp(127.0.0.1:3306)/gmail_mail?readTimeout=10s&writeTimeout=10s",
+            	"mail_table":"new_mail",
+				"sql_driver": "mysql",
+				"primary_mail_host":"sharklasers.com",
+			},
+			"GuerrillaRedisDB" : {
+				"redis_interface" : "127.0.0.1:6379",
+            	"redis_expire_seconds" : 7200,
+				"primary_mail_host":"sharklasers.com"
+			}
+		},
+		"gateways" : {
+			"default" : {
+				"save_workers_size" : 3,
+    			"save_process_old": "HeadersParser|Debugger|Hasher|Header|Compressor|Redis|MySql",
+				"save_process": "GuerrillaRedisDB",
+			}
+		}
+	},
     "servers" : [
         {
             "is_enabled" : true,
@@ -496,8 +537,8 @@ func TestCmdConfigChangeEvents(t *testing.T) {
 	}
 
 	expectedEvents := map[guerrilla.Event]bool{
-		guerrilla.EventConfigBackendConfig: false,
-		guerrilla.EventConfigServerNew:     false,
+		guerrilla.EventConfigBackendConfigChanged: false,
+		guerrilla.EventConfigServerNew:            false,
 	}
 	mainlog, err = getTestLog()
 	if err != nil {
@@ -505,9 +546,16 @@ func TestCmdConfigChangeEvents(t *testing.T) {
 		t.FailNow()
 	}
 
-	bcfg := backends.BackendConfig{"log_received_mails": true}
-	backend, err := backends.New(bcfg, mainlog)
-	app, err := guerrilla.New(oldconf, backend, mainlog)
+	oldconf.BackendConfig = backends.BackendConfig{
+		"processors": {"debugger": {"log_received_mails": true}},
+	}
+
+	backend, err := backends.New(backends.DefaultGateway, oldconf.BackendConfig, mainlog)
+	if err != nil {
+		t.Error("failed to create backend", err)
+	}
+
+	app, err := guerrilla.New(oldconf, mainlog, backend)
 	if err != nil {
 		t.Error("Failed to create new app", err)
 	}

config.go

@@ -30,7 +30,7 @@ type AppConfig struct {
 	// "info", "debug", "error", "panic". Default "info"
 	LogLevel string `json:"log_level,omitempty"`
 	// BackendConfig configures the email envelope processing backend
-	BackendConfig backends.BackendConfig `json:"backend_config"`
+	BackendConfig backends.BackendConfig `json:"backend"`
 }
 
 // ServerConfig specifies config options for a single server
@@ -60,6 +60,8 @@ type ServerConfig struct {
 	// XClientOn when using a proxy such as Nginx, XCLIENT command is used to pass the
 	// original client's IP address & client's HELO
 	XClientOn bool `json:"xclient_on,omitempty"`
+	// Gateway specifies which backend to use
+	Gateway string `json:"backend"`
 }
 
 type ServerTLSConfig struct {
@@ -190,7 +192,7 @@ func (c *AppConfig) Load(jsonBytes []byte) error {
 func (c *AppConfig) EmitChangeEvents(oldConfig *AppConfig, app Guerrilla) {
 	// has backend changed?
 	if !reflect.DeepEqual((*c).BackendConfig, (*oldConfig).BackendConfig) {
-		app.Publish(EventConfigBackendConfig, c)
+		c.emitBackendChangeEvents(oldConfig, app)
 	}
 	// has config changed, general check
 	if !reflect.DeepEqual(oldConfig, c) {
@@ -234,13 +236,27 @@ func (c *AppConfig) EmitChangeEvents(oldConfig *AppConfig, app Guerrilla) {
 }
 
 // EmitLogReopen emits log reopen events using existing config
-func (c *AppConfig) EmitLogReopenEvents(app Guerrilla) {
+func (c *AppConfig) emitLogReopenEvents(app Guerrilla) {
 	app.Publish(EventConfigLogReopen, c)
 	for _, sc := range c.getServers() {
 		app.Publish(EventConfigServerLogReopen, sc)
 	}
 }
 
+func (c *AppConfig) emitBackendChangeEvents(oldConfig *AppConfig, app Guerrilla) {
+	// check what's changed
+	changed, added, removed := c.BackendConfig.Changes(oldConfig.BackendConfig)
+	for b := range changed {
+		app.Publish(EventConfigBackendConfigChanged, c, b)
+	}
+	for b := range added {
+		app.Publish(EventConfigBackendConfigAdded, c, b)
+	}
+	for b := range removed {
+		app.Publish(EventConfigBackendConfigAdded, c, b)
+	}
+}
+
 // gets the servers in a map (key by interface) for easy lookup
 func (c *AppConfig) getServers() map[string]*ServerConfig {
 	servers := make(map[string]*ServerConfig, len(c.Servers))
@@ -310,6 +326,9 @@ func (c *AppConfig) setDefaults() error {
 			if c.Servers[i].LogFile == "" {
 				c.Servers[i].LogFile = c.LogFile
 			}
+			if c.Servers[i].Gateway == "" {
+				c.Servers[i].Gateway = backends.DefaultGateway
+			}
 			// validate the server config
 			err = c.Servers[i].Validate()
 			if err != nil {
@@ -324,36 +343,22 @@ func (c *AppConfig) setDefaults() error {
 // if no backend config was added before starting, then use a default config
 // otherwise, see what required values were missed in the config and add any missing with defaults
 func (c *AppConfig) setBackendDefaults() error {
-
-	if len(c.BackendConfig) == 0 {
-		h, err := os.Hostname()
-		if err != nil {
-			return err
-		}
-		c.BackendConfig = backends.BackendConfig{
-			"log_received_mails": true,
-			"save_workers_size":  1,
-			"save_process":       "HeadersParser|Header|Debugger",
-			"primary_mail_host":  h,
-		}
-	} else {
-		if _, ok := c.BackendConfig["save_process"]; !ok {
-			c.BackendConfig["save_process"] = "HeadersParser|Header|Debugger"
-		}
-		if _, ok := c.BackendConfig["primary_mail_host"]; !ok {
-			h, err := os.Hostname()
-			if err != nil {
-				return err
-			}
-			c.BackendConfig["primary_mail_host"] = h
-		}
-		if _, ok := c.BackendConfig["save_workers_size"]; !ok {
-			c.BackendConfig["save_workers_size"] = 1
-		}
-
-		if _, ok := c.BackendConfig["log_received_mails"]; !ok {
-			c.BackendConfig["log_received_mails"] = false
-		}
+	h, err := os.Hostname()
+	if err != nil {
+		return err
+	}
+	// set the defaults if no value has been configured
+	if c.BackendConfig.GetValue(backends.ConfigGateways, "default", "save_workers_size") == nil {
+		c.BackendConfig.SetValue(backends.ConfigGateways, "default", "save_workers_size", 1)
+	}
+	if c.BackendConfig.GetValue(backends.ConfigGateways, "default", "save_process") == nil {
+		c.BackendConfig.SetValue(backends.ConfigGateways, "default", "save_process", "HeadersParser|Header|Debugger")
+	}
+	if c.BackendConfig.GetValue(backends.ConfigProcessors, "default", "primary_mail_host") == nil {
+		c.BackendConfig.SetValue(backends.ConfigProcessors, "Header", "primary_mail_host", h)
+	}
+	if c.BackendConfig.GetValue(backends.ConfigProcessors, "default", "log_received_mails") == nil {
+		c.BackendConfig.SetValue(backends.ConfigProcessors, "Debugger", "log_received_mails", true)
 	}
 	return nil
 }
@@ -402,6 +407,10 @@ func (sc *ServerConfig) emitChangeEvents(oldServer *ServerConfig, app Guerrilla)
 		app.Publish(EventConfigServerMaxClients, sc)
 	}
 
+	if _, ok := changes["Gateway"]; ok {
+		app.Publish(EventConfigServerGatewayConfig, sc)
+	}
+
 	if len(tlsChanges) > 0 {
 		app.Publish(EventConfigServerTLSConfig, sc)
 	}

config_test.go

@@ -20,10 +20,19 @@ var configJsonA = `
     "log_level" : "debug",
     "pid_file" : "tests/go-guerrilla.pid",
     "allowed_hosts": ["spam4.me","grr.la"],
-    "backend_config" :
-        {
-            "log_received_mails" : true
-        },
+	"backends" : {
+		"processors" : {
+			"debugger" : {
+				"log_received_mails": true
+			},
+		"gateways" : {
+			"default" : {
+				"save_workers_size":  1,
+				"save_process":  "HeadersParser|Header|Hasher|Debugger"
+			}
+		}
+	},
+   
     "servers" : [
         {
             "is_enabled" : true,
@@ -97,10 +106,13 @@ var configJsonB = `
     "log_level" : "debug",
     "pid_file" : "tests/different-go-guerrilla.pid",
     "allowed_hosts": ["spam4.me","grr.la","newhost.com"],
-    "backend_config" :
-        {
-            "log_received_mails" : true
-        },
+    "backend" : {
+		"processors" : {
+			"debugger" {
+				"log_received_mails" : true
+			}
+		}
+	},
     "servers" : [
         {
             "is_enabled" : true,
@@ -230,12 +242,16 @@ func TestConfigChangeEvents(t *testing.T) {
 		t.Error(err)
 	}
 	logger, _ := log.GetLogger(oldconf.LogFile, oldconf.LogLevel)
-	bcfg := backends.BackendConfig{"log_received_mails": true}
-	backend, err := backends.New(bcfg, logger)
+
+	oldconf.BackendConfig = backends.BackendConfig{
+		"processors": {"debugger": {"log_received_mails": true}},
+	}
+
+	backend, err := backends.New("default", oldconf.BackendConfig, logger)
 	if err != nil {
-		t.Error("cannot create backend", err)
+		t.Error("failed to create backend", err)
 	}
-	app, err := New(oldconf, backend, logger)
+	app, err := New(oldconf, logger, backend)
 	if err != nil {
 		t.Error("cannot create daemon", err)
 	}

event.go

@@ -20,7 +20,9 @@ const (
 	// when log level changed
 	EventConfigLogLevel
 	// when the backend's config changed
-	EventConfigBackendConfig
+	EventConfigBackendConfigChanged
+	EventConfigBackendConfigAdded
+	EventConfigBackendConfigRemoved
 	// when a new server was added
 	EventConfigServerNew
 	// when an existing server was removed
@@ -41,6 +43,8 @@ const (
 	EventConfigServerMaxClients
 	// when a server's TLS config changed
 	EventConfigServerTLSConfig
+	// when the server's backend config changed
+	EventConfigServerGatewayConfig
 )
 
 var eventList = [...]string{
@@ -50,7 +54,7 @@ var eventList = [...]string{
 	"config_change:log_file",
 	"config_change:reopen_log_file",
 	"config_change:log_level",
-	"config_change:backend_config",
+	"config_change:backend_config", // todo change to 'backend;
 	"server_change:new_server",
 	"server_change:remove_server",
 	"server_change:update_config",

guerrilla.go

@@ -3,7 +3,9 @@ package guerrilla
 import (
 	"errors"
 	"fmt"
+	"github.com/sirupsen/logrus"
 	"os"
+	"strings"
 	"sync"
 	"sync/atomic"
 
@@ -52,19 +54,20 @@ type guerrilla struct {
 	state int8
 	EventHandler
 	logStore
-	backendStore
+
+	beGuard  sync.Mutex
+	backends BackendContainer
 }
 
 type logStore struct {
 	atomic.Value
 }
 
-type backendStore struct {
-	atomic.Value
-}
+type BackendContainer map[string]backends.Backend
 
 type daemonEvent func(c *AppConfig)
 type serverEvent func(sc *ServerConfig)
+type backendEvent func(c *AppConfig, gateway string)
 
 // Get loads the log.logger in an atomic operation. Returns a stderr logger if not able to load
 func (ls *logStore) mainlog() log.Logger {
@@ -80,13 +83,43 @@ func (ls *logStore) setMainlog(log log.Logger) {
 	ls.Store(log)
 }
 
+// makeConfiguredBackends makes backends from the config
+func (g *guerrilla) makeConfiguredBackends(l log.Logger) ([]backends.Backend, error) {
+	var list []backends.Backend
+	config := g.Config.BackendConfig[backends.ConfigGateways.String()]
+	count := len(config)
+	if count == 0 {
+		return list, errors.New("no backends configured")
+	}
+	list = make([]backends.Backend, count)
+	for name := range config {
+		if b, err := backends.New(name, g.Config.BackendConfig, l); err != nil {
+			return nil, err
+		} else {
+			list = append(list, b)
+		}
+	}
+	return list, nil
+}
+
+// New creates a new Guerrilla instance configured with backends and a logger
 // Returns a new instance of Guerrilla with the given config, not yet running. Backend started.
-func New(ac *AppConfig, b backends.Backend, l log.Logger) (Guerrilla, error) {
+// b can be nil. If nil. then it will use the config to make the backends
+func New(ac *AppConfig, l log.Logger, b ...backends.Backend) (Guerrilla, error) {
 	g := &guerrilla{
 		Config:  *ac, // take a local copy
 		servers: make(map[string]*server, len(ac.Servers)),
 	}
-	g.backendStore.Store(b)
+	if 0 == len(b) {
+		var err error
+		b, err = g.makeConfiguredBackends(l)
+		if err != nil {
+			return g, err
+		}
+	}
+	for i := range b {
+		g.storeBackend(b[i])
+	}
 	g.setMainlog(l)
 
 	if ac.LogLevel != "" {
@@ -106,8 +139,11 @@ func New(ac *AppConfig, b backends.Backend, l log.Logger) (Guerrilla, error) {
 		return g, err
 	}
 
-	// start backend for processing email
-	err = g.backend().Start()
+	// start backends for processing email
+	_, err = g.mapBackends(func(b backends.Backend) error {
+		return b.Start()
+	})
+
 	if err != nil {
 		return g, err
 	}
@@ -133,7 +169,7 @@ func (g *guerrilla) makeServers() error {
 			continue
 		} else {
 			sc := sc // pin!
-			server, err := newServer(&sc, g.backend(), g.mainlog())
+			server, err := newServer(&sc, g.backend(sc.Gateway), g.mainlog())
 			if err != nil {
 				g.mainlog().WithError(err).Errorf("Failed to create server [%s]", sc.ListenInterface)
 				errs = append(errs, err)
@@ -197,6 +233,28 @@ func (g *guerrilla) mapServers(callback func(*server)) map[string]*server {
 	return g.servers
 }
 
+type mapBackendErrors []error
+
+func (e mapBackendErrors) Error() string {
+	data := make([]string, len(e))
+	for i, s := range e {
+		data[i] = fmt.Sprint(s)
+	}
+	return strings.Join(data, ",")
+}
+
+func (g *guerrilla) mapBackends(callback func(backend backends.Backend) error) (BackendContainer, error) {
+	defer g.beGuard.Unlock()
+	g.beGuard.Lock()
+	var e mapBackendErrors
+	for name := range g.backends {
+		if err := callback(g.backends[name]); err != nil {
+			e = append(e, err)
+		}
+	}
+	return g.backends, e
+}
+
 // subscribeEvents subscribes event handlers for configuration change events
 func (g *guerrilla) subscribeEvents() {
 
@@ -268,7 +326,6 @@ func (g *guerrilla) subscribeEvents() {
 		g.mainlog().Debugf("event fired [%s] %s", EventConfigServerNew, sc.ListenInterface)
 		if _, err := g.findServer(sc.ListenInterface); err != nil {
 			// not found, lets add it
-			//
 			if err := g.makeServers(); err != nil {
 				g.mainlog().WithError(err).Errorf("cannot add server [%s]", sc.ListenInterface)
 				return
@@ -373,38 +430,82 @@ func (g *guerrilla) subscribeEvents() {
 			g.mainlog().Infof("Server [%s] re-opened log file [%s]", sc.ListenInterface, sc.LogFile)
 		}
 	})
-	// when the backend changes
-	events[EventConfigBackendConfig] = daemonEvent(func(appConfig *AppConfig) {
+
+	// when the server's gateway setting changed
+	events[EventConfigServerGatewayConfig] = serverEvent(func(sc *ServerConfig) {
+		b := g.backend(sc.Gateway)
+		if b == nil {
+			g.mainlog().WithField("gateway", sc.Gateway).Error("could not change to gateway, not configured")
+			return
+		}
+		g.storeBackend(b)
+	})
+
+	events[EventConfigBackendConfigChanged] = backendEvent(func(appConfig *AppConfig, name string) {
 		logger, _ := log.GetLogger(appConfig.LogFile, appConfig.LogLevel)
 		// shutdown the backend first.
 		var err error
-		if err = g.backend().Shutdown(); err != nil {
-			logger.WithError(err).Warn("Backend failed to shutdown")
+		fields := logrus.Fields{"error": err, "gateway": name}
+		if err = g.backend(name).Shutdown(); err != nil {
+			logger.WithFields(fields).Warn("gateway failed to shutdown")
 			return
 		}
 		// init a new backend, Revert to old backend config if it fails
-		if newBackend, newErr := backends.New(appConfig.BackendConfig, logger); newErr != nil {
-			logger.WithError(newErr).Error("Error while loading the backend")
-			err = g.backend().Reinitialize()
+		if newBackend, newErr := backends.New(name, appConfig.BackendConfig, logger); newErr != nil {
+			logger.WithFields(fields).Error("error while loading the gateway")
+			err = g.backend(name).Reinitialize()
 			if err != nil {
-				logger.WithError(err).Fatal("failed to revert to old backend config")
+				logger.WithFields(fields).Fatal("failed to revert to old gateway config")
 				return
 			}
-			err = g.backend().Start()
+			err = g.backend(name).Start()
 			if err != nil {
-				logger.WithError(err).Fatal("failed to start backend with old config")
+				logger.WithFields(fields).Fatal("failed to start gateway with old config")
 				return
 			}
-			logger.Info("reverted to old backend config")
+			logger.WithField("gateway", name).Info("reverted to old gateway config")
 		} else {
-			// swap to the bew backend (assuming old backend was shutdown so it can be safely swapped)
+			// swap to the bew backend (assuming old gateway was shutdown so it can be safely swapped)
 			if err := newBackend.Start(); err != nil {
-				logger.WithError(err).Error("backend could not start")
+				logger.WithFields(fields).Error("gateway could not start")
 			}
-			logger.Info("new backend started")
+			logger.WithField("gateway", name).Info("new gateway started")
 			g.storeBackend(newBackend)
 		}
 	})
+
+	// a new gateway was added
+	events[EventConfigBackendConfigAdded] = backendEvent(func(appConfig *AppConfig, name string) {
+		logger, _ := log.GetLogger(appConfig.LogFile, appConfig.LogLevel)
+		// shutdown any old backend first.
+		var err error
+		fields := logrus.Fields{"error": err, "gateway": name}
+		if newBackend, newErr := backends.New(name, appConfig.BackendConfig, logger); newErr != nil {
+			logger.WithFields(fields).Error("Error while loading the gateway")
+		} else {
+			// swap to the bew gateway (assuming old gateway was shutdown so it can be safely swapped)
+			if err := newBackend.Start(); err != nil {
+				logger.WithFields(fields).Error("gateway could not start")
+			}
+			logger.WithField("gateway", name).Info("new gateway started")
+			g.storeBackend(newBackend)
+		}
+
+	})
+
+	// remove a gateway (shut it down)
+	events[EventConfigBackendConfigRemoved] = backendEvent(func(appConfig *AppConfig, name string) {
+		logger, _ := log.GetLogger(appConfig.LogFile, appConfig.LogLevel)
+		// shutdown the backend first.
+		var err error
+		if err = g.backend(name).Shutdown(); err != nil {
+			logger.WithFields(logrus.Fields{"error": err, "gateway": name}).Warn("gateway failed to shutdown")
+			return
+		}
+		g.removeBackend(g.backend(name))
+		logger.WithField("gateway", name).Info("gateway removed")
+	})
+
 	var err error
 	for topic, fn := range events {
 		switch f := fn.(type) {
@@ -418,18 +519,28 @@ func (g *guerrilla) subscribeEvents() {
 			break
 		}
 	}
+}
+
+func (g *guerrilla) removeBackend(b backends.Backend) {
+	g.beGuard.Lock()
+	defer g.beGuard.Unlock()
+	delete(g.backends, b.Name())
 
 }
 
 func (g *guerrilla) storeBackend(b backends.Backend) {
-	g.backendStore.Store(b)
+	g.beGuard.Lock()
+	defer g.beGuard.Unlock()
+	g.backends[b.Name()] = b
 	g.mapServers(func(server *server) {
 		server.setBackend(b)
 	})
 }
 
-func (g *guerrilla) backend() backends.Backend {
-	if b, ok := g.backendStore.Load().(backends.Backend); ok {
+func (g *guerrilla) backend(name string) backends.Backend {
+	g.beGuard.Lock()
+	defer g.beGuard.Unlock()
+	if b, ok := g.backends[name]; ok {
 		return b
 	}
 	return nil
@@ -448,10 +559,12 @@ func (g *guerrilla) Start() error {
 	}
 	if g.state == daemonStateStopped {
 		// when a backend is shutdown, we need to re-initialize before it can be started again
-		if err := g.backend().Reinitialize(); err != nil {
-			startErrors = append(startErrors, err)
-		}
-		if err := g.backend().Start(); err != nil {
+		if _, err := g.mapBackends(func(b backends.Backend) error {
+			if err := b.Reinitialize(); err != nil {
+				return err
+			}
+			return b.Start()
+		}); err != nil {
 			startErrors = append(startErrors, err)
 		}
 	}
@@ -508,7 +621,10 @@ func (g *guerrilla) Shutdown() {
 		g.state = daemonStateStopped
 		defer g.guard.Unlock()
 	}()
-	if err := g.backend().Shutdown(); err != nil {
+
+	if _, err := g.mapBackends(func(b backends.Backend) error {
+		return b.Shutdown()
+	}); err != nil {
 		g.mainlog().WithError(err).Warn("Backend failed to shutdown")
 	} else {
 		g.mainlog().Infof("Backend shutdown completed")

server.go

@@ -180,7 +180,10 @@ func (s *server) configureTLS() error {
 
 // setBackend sets the backend to use for processing email envelopes
 func (s *server) setBackend(b backends.Backend) {
-	s.backendStore.Store(b)
+	sc := s.configStore.Load().(ServerConfig)
+	if b.Name() == sc.Gateway {
+		s.backendStore.Store(b)
+	}
 }
 
 // backend gets the backend used to process email envelopes

server_test.go

@@ -51,8 +51,18 @@ func getMockServerConn(sc *ServerConfig, t *testing.T) (*mocks.Conn, *server) {
 	if logOpenError != nil {
 		mainlog.WithError(logOpenError).Errorf("Failed creating a logger for mock conn [%s]", sc.ListenInterface)
 	}
-	backend, err := backends.New(
-		backends.BackendConfig{"log_received_mails": true, "save_workers_size": 1},
+
+	bcfg := backends.BackendConfig{
+		backends.ConfigProcessors.String(): {
+			"debugger": {"log_received_mails": true},
+		},
+		backends.ConfigGateways.String(): {
+			backends.DefaultGateway: {"save_workers_size": 1},
+		},
+	}
+
+	backend, err := backends.New(backends.DefaultGateway,
+		bcfg,
 		mainlog)
 	if err != nil {
 		t.Error("new dummy backend failed because:", err)
@@ -415,8 +425,12 @@ func TestGithubIssue198(t *testing.T) {
 		mainlog.WithError(logOpenError).Errorf("Failed creating a logger for mock conn [%s]", sc.ListenInterface)
 	}
 	conn, server := getMockServerConn(sc, t)
-	be, err := backends.New(map[string]interface{}{
-		"save_process": "HeadersParser|Header|debugger|custom", "primary_mail_host": "example.com", "log_received_mails": true},
+	cfg := backends.BackendConfig{}
+	cfg.SetValue(backends.ConfigGateways, backends.DefaultGateway, "save_process", "HeadersParser|Header|debugger|custom")
+	cfg.SetValue(backends.ConfigProcessors, "header", "primary_mail_host", "example.com")
+	cfg.SetValue(backends.ConfigProcessors, "debugger", "log_received_mails", true)
+
+	be, err := backends.New("default", cfg,
 		mainlog)
 	if err != nil {
 		t.Error(err)
@@ -923,15 +937,15 @@ func TestXClient(t *testing.T) {
 // a second transaction
 func TestGatewayTimeout(t *testing.T) {
 	defer cleanTestArtifacts(t)
-	bcfg := backends.BackendConfig{
-		"save_workers_size":   1,
-		"save_process":        "HeadersParser|Debugger",
-		"log_received_mails":  true,
-		"primary_mail_host":   "example.com",
-		"gw_save_timeout":     "1s",
-		"gw_val_rcpt_timeout": "1s",
-		"sleep_seconds":       2,
-	}
+
+	bcfg := backends.BackendConfig{}
+	bcfg.SetValue(backends.ConfigGateways, backends.DefaultGateway, "gw_save_timeout", "1s")
+	bcfg.SetValue(backends.ConfigGateways, backends.DefaultGateway, "gw_val_rcpt_timeout", "1s")
+	bcfg.SetValue(backends.ConfigGateways, backends.DefaultGateway, "save_workers_size", 1)
+	bcfg.SetValue(backends.ConfigGateways, backends.DefaultGateway, "save_process", "HeadersParser|Debugger")
+	bcfg.SetValue(backends.ConfigProcessors, "header", "primary_mail_host", "example.com")
+	bcfg.SetValue(backends.ConfigProcessors, "debugger", "log_received_mails", true)
+	bcfg.SetValue(backends.ConfigProcessors, "debugger", "sleep_seconds", 2)
 
 	cfg := &AppConfig{
 		LogFile:      log.OutputOff.String(),
@@ -1010,15 +1024,15 @@ func TestGatewayTimeout(t *testing.T) {
 // The processor will panic and gateway should recover from it
 func TestGatewayPanic(t *testing.T) {
 	defer cleanTestArtifacts(t)
-	bcfg := backends.BackendConfig{
-		"save_workers_size":   1,
-		"save_process":        "HeadersParser|Debugger",
-		"log_received_mails":  true,
-		"primary_mail_host":   "example.com",
-		"gw_save_timeout":     "2s",
-		"gw_val_rcpt_timeout": "2s",
-		"sleep_seconds":       1,
-	}
+
+	bcfg := backends.BackendConfig{}
+	bcfg.SetValue(backends.ConfigGateways, backends.DefaultGateway, "gw_save_timeout", "2s")
+	bcfg.SetValue(backends.ConfigGateways, backends.DefaultGateway, "gw_val_rcpt_timeout", "2s")
+	bcfg.SetValue(backends.ConfigGateways, backends.DefaultGateway, "save_workers_size", 1)
+	bcfg.SetValue(backends.ConfigGateways, backends.DefaultGateway, "save_process", "HeadersParser|Debugger")
+	bcfg.SetValue(backends.ConfigProcessors, "header", "primary_mail_host", "example.com")
+	bcfg.SetValue(backends.ConfigProcessors, "debugger", "log_received_mails", true)
+	bcfg.SetValue(backends.ConfigProcessors, "debugger", "sleep_seconds", 1)
 
 	cfg := &AppConfig{
 		LogFile:      log.OutputOff.String(),

tests/guerrilla_test.go

@@ -41,7 +41,7 @@ import (
 
 type TestConfig struct {
 	guerrilla.AppConfig
-	BackendConfig map[string]interface{} `json:"backend_config"`
+	BackendConfig backends.BackendConfig `json:"backend"`
 }
 
 var (
@@ -68,7 +68,7 @@ func init() {
 			return
 		}
 		backend, _ := getBackend(config.BackendConfig, logger)
-		app, initErr = guerrilla.New(&config.AppConfig, backend, logger)
+		app, initErr = guerrilla.New(&config.AppConfig, logger, backend)
 	}
 
 }
@@ -80,10 +80,14 @@ var configJson = `
     "log_level" : "debug",
     "pid_file" : "go-guerrilla.pid",
     "allowed_hosts": ["spam4.me","grr.la"],
-    "backend_config" :
-        {
-            "log_received_mails" : true
-        },
+	
+    "backend" : {
+		"processors" : {
+			"debugger" {
+				"log_received_mails" : true
+			}
+		}
+	},
     "servers" : [
         {
             "is_enabled" : true,
@@ -120,8 +124,8 @@ var configJson = `
 }
 `
 
-func getBackend(backendConfig map[string]interface{}, l log.Logger) (backends.Backend, error) {
-	b, err := backends.New(backendConfig, l)
+func getBackend(backendConfig backends.BackendConfig, l log.Logger) (backends.Backend, error) {
+	b, err := backends.New(backends.DefaultGateway, backendConfig, l)
 	if err != nil {
 		fmt.Println("backend init error", err)
 		os.Exit(1)