// Configure initializes this producer with values from a plugin config.
965564
func (prod *KafkaProducer) Configure(conf core.PluginConfigReader) error {
	prod.SetStopCallback(prod.close)

	prod.streamToTopic = conf.GetStreamMap("Topic", "default")
	prod.topic = make(map[core.MessageStreamID]*topicHandle)

	prod.topicGuard = new(sync.RWMutex)
	prod.streamToTopic = conf.GetStreamMap("Topic", "default")

	prod.topicHandles = make(map[string]*topicHandle)

	// Init librdkafka
	prod.config = kafka.NewConfig()

	/*kafkaVer := conf.GetString("ProtocolVersion", "0.8.2")
	verParts := strings.Split(kafkaVer, ".")
	multiplicator := 1000000
	verNumber := 0
	for _, n := range verParts {
		part, _ := strconv.Atoi(n)
		verNumber += part * multiplicator
		multiplicator /= 100
	}*/

	prod.config.Set("metadata.broker.list", strings.Join(prod.servers, ","))
	prod.config.Set("sasl.mechanisms", conf.GetString("SaslMechanism", ""))
	prod.config.Set("sasl.password", conf.GetString("SaslPassword", ""))
	prod.config.Set("sasl.username", conf.GetString("SaslUsername", ""))
	prod.config.Set("ssl.ca.location", conf.GetString("SslCaLocation", ""))
	prod.config.Set("ssl.certificate.location", conf.GetString("SslCertificateLocation", ""))
	prod.config.Set("ssl.cipher.suites", conf.GetString("SslCipherSuites", ""))
	prod.config.Set("ssl.crl.location", conf.GetString("SslCrlLocation", ""))
	prod.config.Set("ssl.key.location", conf.GetString("SslKeyLocation", ""))
	prod.config.Set("ssl.key.password", conf.GetString("SslKeyPassword", ""))
	prod.config.SetI("message.max.bytes", int(conf.GetInt("Batch/SizeMaxKB", 1<<10))<<10)
	prod.config.SetI("metadata.request.timeout.ms", int(conf.GetInt("MetadataTimeoutMs", 1500)))
	prod.config.SetI("topic.metadata.refresh.interval.ms", int(conf.GetInt("MetadataRefreshMs", 300000)))
	prod.config.SetI("socket.max.fails", int(conf.GetInt("ServerMaxFails", 3)))
	prod.config.SetI("socket.timeout.ms", int(conf.GetInt("ServerTimeoutSec", 60)*1000))
	prod.config.SetB("socket.keepalive.enable", true)
	prod.config.SetI("message.send.max.retries", int(conf.GetInt("SendRetries", 0)))
	prod.config.SetI("queue.buffering.max.messages", int(conf.GetInt("Batch/MaxMessages", 100000)))
	prod.config.SetI("queue.buffering.max.ms", int(conf.GetInt("Batch/TimeoutMs", 1000)))
	prod.config.SetI("batch.num.messages", int(conf.GetInt("Batch/MinMessages", 1000)))
	//prod.config.SetI("protocol.version", verNumber)

	securityProtocol := strings.ToLower(conf.GetString("SecurityProtocol", protocolPlaintext))
	switch securityProtocol {
	default:
		return fmt.Errorf("%s is not a valid security protocol", securityProtocol)
	case protocolPlaintext:
		prod.config.Set("security.protocol", protocolPlaintext)
	case protocolSsl:
		prod.config.Set("security.protocol", protocolSsl)
	case protocolSaslPlaintext:
		prod.config.Set("security.protocol", protocolSaslPlaintext)
	case protocolSaslSsl:
		prod.config.Set("security.protocol", protocolSaslSsl)
	}

	switch strings.ToLower(conf.GetString("Compression", compressNone)) {
	default:
		prod.config.Set("compression.codec", "none")
	case compressGZIP:
		prod.config.Set("compression.codec", "gzip")
	case compressSnappy:
		prod.config.Set("compression.codec", "snappy")
	}

	tgo.Metric.New(kafkaMetricAllocations)
	return nil
}
1086355
func (format *ProcessTSV) Configure(conf core.PluginConfigReader) {
	directives := conf.GetStringArray("Directives", []string{})

	format.directives = make([]tsvDirective, 0, len(directives))
	for _, d := range directives {
		directive := strings.Replace(d, "\\:", "\r", -1)
		parts := strings.Split(directive, ":")
		for i, value := range parts {
			parts[i] = strings.Replace(value, "\r", ":", -1)
		}

		if len(parts) >= 2 {
			index, err := strconv.Atoi(parts[0])
			conf.Errors.Push(err)

			newDirective := tsvDirective{
				index:     index,
				operation: strings.ToLower(parts[1]),
			}

			for i := 2; i < len(parts); i++ {
				newDirective.parameters = append(newDirective.parameters, tstrings.Unescape(parts[i]))
			}

			format.directives = append(format.directives, newDirective)
		}
	}
}
436301
func (format *Aggregate) Configure(conf core.PluginConfigReader) {
	applyTo := conf.GetString("ApplyTo", "")

	// init modulator array
	modulatorSettings := format.getModulatorSettings(applyTo, conf)

	config := core.NewPluginConfig("", "format.Aggregate.Modulators")
	config.Override("Modulators", modulatorSettings)
	modulatorReader := core.NewPluginConfigReaderWithError(&config)

	modulatorArray, err := modulatorReader.GetModulatorArray("Modulators", format.Logger, core.ModulatorArray{})
	if err != nil {
		conf.Errors.Push(err)
	}

	format.modulators = modulatorArray
}
685764
func (w *unixConfigure) Configure() error {
	if err := w.ConfigureBasic(); err != nil {
		return err
	}
	if err := w.ConfigureJuju(); err != nil {
		return err
	}
	return w.ConfigureCustomOverrides()
}
63854
func (format *Double) Configure(conf core.PluginConfigReader) {
	format.applyTo = conf.GetString("ApplyTo", "")
}
436619
func (format *Identifier) Configure(conf core.PluginConfigReader) {
	idType := strings.ToLower(conf.GetString("Generator", "time"))
	format.seq = new(uint64)

	switch idType {
	case "hash":
		format.hash = format.idHash
	case "seq":
		format.hash = format.idSeq
	case "seqhex":
		format.hash = format.idSeqHex
	case "time":
		format.hash = format.idTime
	default:
		format.hash = func(msg *core.Message) []byte {
			return []byte(idType)
		}
	}
}
641838
func (n *Node) configure(ctx context.Context, cc raftpb.ConfChange) error {
	cc.ID = n.reqIDGen.Next()

	ctx, cancel := context.WithCancel(ctx)
	ch := n.wait.register(cc.ID, nil, cancel)

	if err := n.raftNode.ProposeConfChange(ctx, cc); err != nil {
		n.wait.cancel(cc.ID)
		return err
	}

	select {
	case x := <-ch:
		if err, ok := x.(error); ok {
			return err
		}
		if x != nil {
			log.G(ctx).Panic("raft: configuration change error, return type should always be error")
		}
		return nil
	case <-ctx.Done():
		n.wait.cancel(cc.ID)
		return ctx.Err()
	}
}
1321822
func (cons *SystemdConsumer) Configure(conf core.PluginConfigReader) error {
	var err error
	cons.journal, err = sdjournal.NewJournal()
	if err != nil {
		return err
	}

	if sdUnit := conf.GetString("SystemdUnit", ""); sdUnit != "" {
		err = cons.journal.AddMatch("_SYSTEMD_UNIT=" + sdUnit)
		if err != nil {
			return err
		}
	}

	// Offset
	offsetValue := strings.ToLower(conf.GetString("DefaultOffset", sdOffsetTail))

	if cons.offsetFile != "" {
		fileContents, err := ioutil.ReadFile(cons.offsetFile)
		if err != nil {
			cons.Logger.WithError(err).Error("Error reading offset file")
		}
		if string(fileContents) != "" {
			offsetValue = string(fileContents)
		}
	}

	switch offsetValue {
	case sdOffsetHead:
		if err = cons.journal.SeekHead(); err != nil {
			return err
		}

	case sdOffsetTail:
		if err = cons.journal.SeekTail(); err != nil {
			return err
		}
		// start *after* the newest record
		if _, err = cons.journal.Next(); err != nil {
			return err
		}

	default:
		offset, err := strconv.ParseUint(offsetValue, 10, 64)
		if err != nil {
			return err
		}
		if err = cons.journal.SeekRealtimeUsec(offset); err != nil {
			return err
		}
		// start *after* the specified time
		if _, err = cons.journal.Next(); err != nil {
			return err
		}
	}

	// Register close to the control message handler
	cons.SetStopCallback(cons.close)
	return conf.Errors.OrNil()
}
435912
func (filter *JSON) Configure(conf core.PluginConfigReader) {
	rejectValues := conf.GetStringMap("Reject", make(map[string]string))
	acceptValues := conf.GetStringMap("Accept", 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 !conf.Errors.Push(err) {
			filter.rejectValues[key] = exp
		}
	}

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

	applyTo := conf.GetString("ApplyTo", "")
	filter.getAppliedContent = core.GetAppliedContentGetFunction(applyTo)
}
1273916
func (prod *Proxy) Configure(conf core.PluginConfigReader) {
	prod.SetStopCallback(prod.close)

	prod.protocol, prod.address = tnet.ParseAddress(conf.GetString("Address", ":5880"), "tcp")
	if prod.protocol == "udp" {
		conf.Errors.Pushf("Proxy does not support UDP")
	}

	delimiter := tstrings.Unescape(conf.GetString("Delimiter", "\n"))
	offset := int(conf.GetInt("Offset", 0))
	flags := tio.BufferedReaderFlagEverything // pass all messages as-is

	partitioner := strings.ToLower(conf.GetString("Partitioner", "delimiter"))
	switch partitioner {
	case "binary_be":
		flags |= tio.BufferedReaderFlagBigEndian
		fallthrough

	case "binary", "binary_le":
		switch conf.GetInt("Size", 4) {
		case 1:
			flags |= tio.BufferedReaderFlagMLE8
		case 2:
			flags |= tio.BufferedReaderFlagMLE16
		case 4:
			flags |= tio.BufferedReaderFlagMLE32
		case 8:
			flags |= tio.BufferedReaderFlagMLE64
		default:
			conf.Errors.Pushf("Size only supports the value 1,2,4 and 8")
		}

	case "fixed":
		flags |= tio.BufferedReaderFlagMLEFixed
		offset = int(conf.GetInt("Size", 1))

	case "ascii":
		flags |= tio.BufferedReaderFlagMLE

	case "delimiter":
		// Nothing to add

	default:
		conf.Errors.Pushf("Unknown partitioner: %s", partitioner)
	}

	prod.reader = tio.NewBufferedReader(prod.bufferSizeKB, flags, offset, delimiter)
}
338136
func Configure(b *bootstrap.Bootstrapper) {
	h := New(b)
	b.UseGlobal(h)
}
1341916
func (_m *Communicator) Configure(channel string, members []cluster.RemoteNode) {
	_m.Called(channel, members)
}
1138199
func (sqlite3 *Sqlite3DB) Configure(cfg map[string]string) {
	sqlite3.DBPath = cfg["DBPath"]
	sqlite3.TableName = cfg["TableName"]
}
1224428
func (app *Application) Configure(configurators ...func(*Application)) *Application {
	for _, c := range configurators {
		c(app)
	}
	return app
}
209838
func (r *Router) Configure(opts ...RouterOption) {
	for _, o := range opts {
		o(r)
	}
}
773018
func (prod *StatsdMetrics) Configure(conf core.PluginConfigReader) {
	prod.SetStopCallback(prod.close)

	prod.streamMap = conf.GetStreamMap("StreamMapping", "")
	prod.batch = core.NewMessageBatch(int(conf.GetInt("Batch/MaxMessages", 500)))

	server := conf.GetString("Server", "localhost:8125")
	prefix := conf.GetString("Prefix", "gollum.")
	prod.client = statsd.NewStatsdClient(server, prefix)
	conf.Errors.Push(prod.client.CreateSocket())
}
1127596
func Configure(b *bootstrap.Bootstrapper) {
	b.Get("/", GetIndexHandler)
	b.Get("/follower/{id:long}", GetFollowerHandler)
	b.Get("/following/{id:long}", GetFollowingHandler)
	b.Get("/like/{id:long}", GetLikeHandler)
}