// Configure initializes this filter with values from a plugin config.
func (filter *JSON) Configure(conf core.PluginConfig) error {
	rejectValues := conf.GetStringMap("FilterReject", make(map[string]string))
	acceptValues := conf.GetStringMap("FilterAccept", make(map[string]string))

	// Compile regexp from map[string]string to map[string]*regexp.Regexp
	filter.rejectValues = make(map[string]*regexp.Regexp)
	filter.acceptValues = make(map[string]*regexp.Regexp)

	for key, val := range rejectValues {
		exp, err := regexp.Compile(val)
		if err != nil {
			return err
		}
		filter.rejectValues[key] = exp
	}

	for key, val := range acceptValues {
		exp, err := regexp.Compile(val)
		if err != nil {
			return err
		}
		filter.acceptValues[key] = exp
	}

	return nil
}

// Configure initializes this consumer with values from a plugin config.
func (cons *File) Configure(conf core.PluginConfig) error {
	err := cons.ConsumerBase.Configure(conf)
	if err != nil {
		return err
	}

	if !conf.HasValue("File") {
		return core.NewConsumerError("No file configured for consumer.File")
	}

	cons.file = nil
	cons.fileName = conf.GetString("File", "")
	cons.offsetFileName = conf.GetString("OffsetFile", "")
	cons.delimiter = shared.Unescape(conf.GetString("Delimiter", "\n"))

	switch strings.ToLower(conf.GetString("DefaultOffset", fileOffsetEnd)) {
	default:
		fallthrough
	case fileOffsetEnd:
		cons.seek = 2
		cons.seekOffset = 0

	case fileOffsetStart:
		cons.seek = 1
		cons.seekOffset = 0
	}

	return nil
}

// Configure sets the database connection values
func (writer *influxDBWriter10) configure(conf core.PluginConfig, prod *InfluxDB) error {
	writer.host = conf.GetString("Host", "localhost:8086")
	writer.username = conf.GetString("User", "")
	writer.password = conf.GetString("Password", "")
	writer.databaseTemplate = conf.GetString("Database", "default")
	writer.buffer = shared.NewByteStream(4096)
	writer.connectionUp = false
	writer.timeBasedDBName = conf.GetBool("TimeBasedName", true)
	writer.Control = prod.Control

	writer.writeURL = fmt.Sprintf("http://%s/write", writer.host)
	writer.queryURL = fmt.Sprintf("http://%s/query", writer.host)
	writer.pingURL = fmt.Sprintf("http://%s/ping", writer.host)
	writer.separator = '?'

	if writer.username != "" {
		credentials := fmt.Sprintf("?u=%s&p=%s", url.QueryEscape(writer.username), url.QueryEscape(writer.password))
		writer.writeURL += credentials
		writer.queryURL += credentials
		writer.separator = '&'
	}

	writer.writeURL = fmt.Sprintf("%s%cprecision=ms", writer.writeURL, writer.separator)
	prod.SetCheckFuseCallback(writer.isConnectionUp)
	return nil
}

// Configure initializes this consumer with values from a plugin config.
func (cons *Http) Configure(conf core.PluginConfig) error {
	err := cons.ConsumerBase.Configure(conf)
	if err != nil {
		return err
	}

	cons.address = conf.GetString("Address", ":80")
	cons.readTimeoutSec = time.Duration(conf.GetInt("ReadTimeoutSec", 3)) * time.Second
	cons.withHeaders = conf.GetBool("WithHeaders", true)

	certificateFile := conf.GetString("Certificate", "")
	keyFile := conf.GetString("PrivateKey", "")

	if certificateFile != "" || keyFile != "" {
		if certificateFile == "" || keyFile == "" {
			return fmt.Errorf("There must always be a certificate and a private key or none of both")
		}

		cons.certificate = new(tls.Config)
		cons.certificate.NextProtos = []string{"http/1.1"}

		keypair, err := tls.LoadX509KeyPair(certificateFile, keyFile)
		if err != nil {
			return err
		}

		cons.certificate.Certificates = []tls.Certificate{keypair}
	}

	return err
}

// Configure initializes this formatter with values from a plugin config.
func (format *CollectdToInflux08) Configure(conf core.PluginConfig) error {
	plugin, err := core.NewPluginWithType(conf.GetString("CollectdToInfluxFormatter", "format.Forward"), conf)
	if err != nil {
		return err
	}
	format.base = plugin.(core.Formatter)
	return nil
}

// Configure initializes this formatter with values from a plugin config.
func (format *Runlength) Configure(conf core.PluginConfig) error {
	plugin, err := core.NewPluginWithType(conf.GetString("RunlengthFormatter", "format.Forward"), conf)
	if err != nil {
		return err
	}

	format.base = plugin.(core.Formatter)
	return nil
}

// Configure initializes this formatter with values from a plugin config.
func (format *Serialize) Configure(conf core.PluginConfig) error {
	plugin, err := core.NewPluginWithType(conf.GetString("SerializeFormatter", "format.Forward"), conf)
	if err != nil {
		return err
	}

	format.encode = conf.GetBool("SerializeStringEncode", true)
	format.base = plugin.(core.Formatter)
	return nil
}

// Configure initializes this formatter with values from a plugin config.
func (format *CollectdToInflux10) Configure(conf core.PluginConfig) error {
	plugin, err := core.NewPluginWithType(conf.GetString("CollectdToInflux1009", "format.Forward"), conf)
	if err != nil {
		return err
	}
	format.base = plugin.(core.Formatter)
	format.tagString = strings.NewReplacer(",", "\\,", " ", "\\ ")
	format.stringString = strings.NewReplacer("\"", "\\\"")
	return nil
}

// Configure initializes this formatter with values from a plugin config.
func (format *Hostname) Configure(conf core.PluginConfig) error {
	plugin, err := core.NewPluginWithType(conf.GetString("HostnameFormatter", "format.Forward"), conf)
	if err != nil {
		return err
	}

	format.separator = conf.GetString("HostnameSeparator", " ")
	format.base = plugin.(core.Formatter)
	return nil
}

// Configure initializes this formatter with values from a plugin config.
func (format *Timestamp) Configure(conf core.PluginConfig) error {
	plugin, err := core.NewPluginWithType(conf.GetString("TimestampFormatter", "format.Forward"), conf)
	if err != nil {
		return err
	}

	format.base = plugin.(core.Formatter)
	format.timestampFormat = conf.GetString("Timestamp", "2006-01-02 15:04:05 MST | ")

	return nil
}

// Configure initializes this formatter with values from a plugin config.
func (format *StreamRoute) Configure(conf core.PluginConfig) error {
	plugin, err := core.NewPluginWithType(conf.GetString("StreamRouteFormatter", "format.Forward"), conf)
	if err != nil {
		return err
	}

	format.delimiter = []byte(conf.GetString("StreamRouteDelimiter", ":"))
	format.base = plugin.(core.Formatter)

	return nil
}

// Configure initializes this formatter with values from a plugin config.
func (format *Base64Decode) Configure(conf core.PluginConfig) error {
	dict := conf.GetString("Dictionary", "")
	if dict == "" {
		format.dictionary = base64.StdEncoding
	} else {
		if len(dict) != 64 {
			return fmt.Errorf("Base64 dictionary must contain 64 characters.")
		}
		format.dictionary = base64.NewEncoding(dict)
	}
	return nil
}

// Configure initializes this distributor with values from a plugin config.
func (stream *Route) Configure(conf core.PluginConfig) error {
	if err := stream.StreamBase.ConfigureStream(conf, stream.Broadcast); err != nil {
		return err // ### return, base stream error ###
	}

	routes := conf.GetStringArray("Routes", []string{})
	for _, streamName := range routes {
		targetStream := newStreamWithID(streamName)
		stream.routes = append(stream.routes, targetStream)
	}

	return nil
}

// Configure initializes this consumer with values from a plugin config.
func (cons *File) Configure(conf core.PluginConfig) error {
	err := cons.ConsumerBase.Configure(conf)
	if err != nil {
		return err
	}

	cons.SetRollCallback(cons.onRoll)

	cons.file = nil
	cons.fileName = conf.GetString("File", "/var/run/system.log")
	cons.offsetFileName = conf.GetString("OffsetFile", "")
	cons.delimiter = shared.Unescape(conf.GetString("Delimiter", "\n"))

	switch strings.ToLower(conf.GetString("DefaultOffset", fileOffsetEnd)) {
	default:
		fallthrough
	case fileOffsetEnd:
		cons.seek = 2
		cons.seekOffset = 0

	case fileOffsetStart:
		cons.seek = 1
		cons.seekOffset = 0
	}

	return nil
}

// Configure initializes this consumer with values from a plugin config.
func (cons *Profiler) Configure(conf core.PluginConfig) error {
	err := cons.ConsumerBase.Configure(conf)
	if err != nil {
		return err
	}
	numTemplates := conf.GetInt("TemplateCount", 10)

	cons.profileRuns = conf.GetInt("Runs", 10000)
	cons.batches = conf.GetInt("Batches", 10)
	cons.chars = conf.GetString("Characters", profilerDefaultCharacters)
	cons.message = conf.GetString("Message", "%# %256s")
	cons.templates = make([][]byte, numTemplates)

	return nil
}

// Configure initializes this consumer with values from a plugin config.
func (cons *LoopBack) Configure(conf core.PluginConfig) error {
	err := cons.ConsumerBase.Configure(conf)
	if err != nil {
		return err
	}

	routes := conf.GetStreamRoutes("Routes")
	cons.routes = make(map[core.MessageStreamID]map[core.MessageStreamID]core.Stream)

	for sourceID, targetIDs := range routes {
		for _, targetID := range targetIDs {
			cons.addRoute(sourceID, targetID)
		}
	}

	core.EnableRetryQueue(conf.GetInt("Channel", 8192))
	return nil
}

// Configure initializes this formatter with values from a plugin config.
func (format *Base64Decode) Configure(conf core.PluginConfig) error {
	plugin, err := core.NewPluginWithType(conf.GetString("Base64Formatter", "format.Forward"), conf)
	if err != nil {
		return err
	}
	format.base = plugin.(core.Formatter)

	dict := conf.GetString("Base64Dictionary", "")
	if dict == "" {
		format.dictionary = base64.StdEncoding
	} else {
		if len(dict) != 64 {
			return fmt.Errorf("Base64 dictionary must contain 64 characters.")
		}
		format.dictionary = base64.NewEncoding(dict)
	}
	return nil
}

// Configure initializes this producer with values from a plugin config.
func (prod *Console) Configure(conf core.PluginConfig) error {
	err := prod.ProducerBase.Configure(conf)
	if err != nil {
		return err
	}

	console := conf.GetString("Console", "stdout")

	switch strings.ToLower(console) {
	default:
		fallthrough
	case "stdout":
		prod.console = os.Stdout
	case "stderr":
		prod.console = os.Stderr
	}

	return nil
}

// Configure initializes this consumer with values from a plugin config.
func (cons *Syslogd) Configure(conf core.PluginConfig) error {
	err := cons.ConsumerBase.Configure(conf)
	if err != nil {
		return err
	}

	cons.address, cons.protocol = shared.ParseAddress(conf.GetString("Address", "udp://0.0.0.0:514"))
	format := conf.GetString("Format", "RFC6587")

	switch cons.protocol {
	case "udp", "tcp", "unix":
	default:
		return fmt.Errorf("Syslog: unknown protocol type %s", cons.protocol) // ### return, unknown protocol ###
	}

	switch format {
	// http://www.ietf.org/rfc/rfc3164.txt
	case "RFC3164":
		cons.format = syslog.RFC3164
		if cons.protocol == "tcp" {
			Log.Warning.Print("Syslog: RFC3164 demands UDP")
			cons.protocol = "udp"
		}

	// https://tools.ietf.org/html/rfc5424
	case "RFC5424":
		cons.format = syslog.RFC5424
		if cons.protocol == "tcp" {
			Log.Warning.Print("Syslog: RFC5424 demands UDP")
			cons.protocol = "udp"
		}

	// https://tools.ietf.org/html/rfc6587
	case "RFC6587":
		cons.format = syslog.RFC6587

	default:
		err = fmt.Errorf("Syslog: Format %s is not supported", format)
	}

	cons.sequence = new(uint64)
	return err
}

// Configure initializes this filter with values from a plugin config.
func (filter *RegExp) Configure(conf core.PluginConfig) error {
	var err error

	exp := conf.GetString("FilterExpression", "")
	if exp != "" {
		filter.exp, err = regexp.Compile(exp)
		if err != nil {
			return err // ### return, regex parser error ###
		}
	}

	exp = conf.GetString("FilterExpressionNot", "")
	if exp != "" {
		filter.expNot, err = regexp.Compile(exp)
		if err != nil {
			return err // ### return, regex parser error ###
		}
	}

	return nil
}