// ApplyFormatter update message payload
968041
func (format *ExtractJSON) ApplyFormatter(msg *core.Message) error {
	content := format.GetAppliedContent(msg)

	value, err := format.extractJSON(content)
	if err != nil {
		return err
	}

	format.SetAppliedContent(msg, value)
	return nil
}
987433
func (f *Int64SliceFlag) Apply(set *flag.FlagSet) {
	f.set = set
	f.Int64SliceFlag.Apply(set)
}
214723
func MakeCallSiteFormatter(depth int) MessageFormatter {
	depth++
	return func(m message.Composer) (string, error) {
		file, line := callerInfo(depth)
		return fmt.Sprintf(callSiteTmpl, m.Priority(), file, line, m), nil
	}
}
222157
func (b *MultiHandler) SetFormatter(f Formatter) {
	for _, h := range b.handlers {
		h.SetFormatter(f)
	}
}
367197
func applyTransform(str string, transform sf) (out string) {
	out = ""

	for idx, line := range strings.Split(str, "\n") {
		out += transform(idx, line)
	}

	return
}
37042
func (format *MetadataCopy) ApplyFormatter(msg *core.Message) error {

	if len(format.metaDataKeys) > 0 {
		// DEPRECATED
		// This codepath will be removed in 0.6
		meta := msg.GetMetadata()
		data := format.GetAppliedContent(msg)

		for _, key := range format.metaDataKeys {
			bufferCopy := make([]byte, len(data))
			copy(bufferCopy, data)
			meta.SetValue(key, bufferCopy)
		}
		return nil
	}

	getSourceData := core.GetAppliedContentGetFunction(format.key)
	srcData := getSourceData(msg)

	cloneSrcData := make([]byte, len(srcData))
	copy(cloneSrcData, srcData)

	switch format.mode {
	case metadataCopyModeReplace:
		format.SetAppliedContent(msg, cloneSrcData)

	case metadataCopyModePrepend:
		dstData := format.GetAppliedContent(msg)
		if len(format.separator) != 0 {
			cloneSrcData = append(cloneSrcData, format.separator...)
		}
		format.SetAppliedContent(msg, append(cloneSrcData, dstData...))

	case metadataCopyModeAppend:
		dstData := format.GetAppliedContent(msg)
		if len(format.separator) != 0 {
			dstData = append(dstData, format.separator...)
		}
		format.SetAppliedContent(msg, append(dstData, cloneSrcData...))
	}

	return nil
}
1250500
func (il IfLogger) SetFormatter(f Formatter) ContextLogger {
	return &contextLogger{
		isTrue:    il.ok(),
		formatter: f,
	}
}
1199074
func applySlice(dst reflect.Value, src reflect.Value) error {
	dst.Set(reflect.MakeSlice(dst.Type(), src.Len(), src.Len()))
	dstElemType := dst.Type().Elem()
	for i := 0; i < src.Len(); i++ {
		v, err := convert(src.Index(i), dstElemType)
		if err != nil {
			return fmt.Errorf("Invalid %dth value: %s", i, err)
		}
		dst.Index(i).Set(v)
	}
	return nil
}
1170581
func NewFormattedMessage(p level.Priority, base string, args ...interface{}) Composer {
	m := &formatMessenger{
		base: base,
		args: args,
	}
	_ = m.SetPriority(p)

	return m
}
467143
func (format *Clear) ApplyFormatter(msg *core.Message) error {
	format.SetAppliedContent(msg, nil)
	return nil
}
936927
func (format *RegExp) ApplyFormatter(msg *core.Message) error {
	content := format.GetAppliedContent(msg)
	matches := format.expression.FindSubmatchIndex(content)
	transformed := format.expression.Expand([]byte{}, format.template, content, matches)

	format.SetAppliedContent(msg, transformed)
	return nil
}
930608
func simpleMessageFormatter(entry *JournalEntry) (string, error) {
	msg, ok := entry.Fields["MESSAGE"]
	if !ok {
		return "", fmt.Errorf("no MESSAGE field present in journal entry")
	}

	usec := entry.RealtimeTimestamp
	timestamp := time.Unix(0, int64(usec)*int64(time.Microsecond))

	return fmt.Sprintf("%s %s\n", timestamp, msg), nil
}
920625
func (format *CollectdToInflux10) ApplyFormatter(msg *core.Message) error {
	contentData := format.GetAppliedContent(msg)

	collectdData, err := parseCollectdPacket(contentData)
	if err != nil {
		format.Logger.Error("Collectd parser error: ", err)
		return err
	}

	// Manually convert to line protocol
	influxData := tio.NewByteStream(len(contentData))
	timestamp := int64(collectdData.Time * 1000)
	fixedPart := fmt.Sprintf(
		`%s,plugin_instance=%s,type=%s,type_instance=%s,host=%s`,
		format.escapeTag(collectdData.Plugin),
		format.escapeTag(collectdData.PluginInstance),
		format.escapeTag(collectdData.PluginType),
		format.escapeTag(collectdData.TypeInstance),
		format.escapeTag(collectdData.Host))

	setSize := tmath.Min3I(len(collectdData.Dstypes), len(collectdData.Dsnames), len(collectdData.Values))
	for i := 0; i < setSize; i++ {
		fmt.Fprintf(&influxData,
			`%s,dstype=%s,dsname=%s value=%f %d\n`,
			fixedPart,
			format.escapeTag(collectdData.Dstypes[i]),
			format.escapeTag(collectdData.Dsnames[i]),
			collectdData.Values[i],
			timestamp)
	}

	format.SetAppliedContent(msg, influxData.Bytes())
	return nil
}
154284
func (format *JSONToInflux10) ApplyFormatter(msg *core.Message) error {
	content := format.GetAppliedContent(msg)

	values := make(tcontainer.MarshalMap)
	err := json.Unmarshal(content, &values)

	if err != nil {
		format.Logger.Warningf("JSON parser error: %s, Message: %s", err, content)
		return err
	}

	var timestamp int64
	if val, ok := values[format.timeField]; ok {
		timestamp, err = format.toUnixTime(val.(string))
		if err != nil {
			format.Logger.Error("Invalid time format in message:", err)
		}
		delete(values, format.timeField)
	} else {
		timestamp = time.Now().Unix()
	}

	measurement, measurementFound := values[format.measurement]
	if !measurementFound {
		return fmt.Errorf("required field for measurement (%s) not found in payload", format.measurement)
	}

	delete(values, format.measurement)

	fields := make(map[string]interface{})
	tags := make(map[string]interface{})
	for k, v := range values {
		if _, ignore := format.ignore[k]; ignore {
			continue
		}
		key := format.escapeMetadata(k)
		if _, isTag := format.tags[k]; isTag {
			tags[key] = format.escapeMetadata(v.(string))
		} else {
			fields[key] = format.escapeFieldValue(v.(string))
		}
	}

	measurementString := format.escapeMeasurement(measurement.(string))

	tagsString := format.joinMap(tags)
	if len(tagsString) > 0 {
		// Only add comma between measurement and tags if tags are not empty.
		// See https://docs.influxdata.com/influxdb/v1.2/write_protocols/line_protocol_reference/
		tagsString = "," + tagsString
	}

	fieldsString := format.joinMap(fields)

	line := fmt.Sprintf(
		`%s%s %s %d`,
		measurementString,
		tagsString,
		fieldsString,
		timestamp)

	format.SetAppliedContent(msg, []byte(line))
	return nil
}
1344638
func (format *JSONToArray) ApplyFormatter(msg *core.Message) error {
	content, err := format.getCsvContent(format.GetAppliedContent(msg))
	if err != nil {
		return err
	}

	format.SetAppliedContent(msg, content)
	return nil
}
892002
func (format *ProcessTSV) ApplyFormatter(msg *core.Message) error {
	if len(format.directives) == 0 {
		return nil // continue
	}

	values := make([]tsvValue, 0)
	if format.quotedValues {
		remainder := string(format.GetAppliedContent(msg))
		for {
			if remainder[:1] != `"` {
				split := strings.SplitN(remainder, format.delimiter+`"`, 2)
				tsvValues := stringsToTSVValues(strings.Split(split[0], format.delimiter))
				values = append(values, tsvValues...)
				if len(split) == 2 {
					remainder = split[1]
				} else {
					break
				}
			} else {
				remainder = remainder[1:]
			}
			split := strings.SplitN(remainder, `"`+format.delimiter, 2)
			if len(split) == 2 {
				values = append(values, tsvValue{
					quoted: true,
					value:  split[0],
				})
				remainder = split[1]
			} else {
				if split[0][len(split[0])-1:] != `"` {
					// unmatched quote, abort processing this message
					format.Logger.Warning("ProcessTSV failed to parse a message: unmatched quote")
					return nil // continue
				}
				values = append(values, tsvValue{
					quoted: true,
					value:  split[0][:len(split[0])-1],
				})
				break
			}
		}
	} else {
		values = stringsToTSVValues(strings.Split(string(format.GetAppliedContent(msg)), format.delimiter))
	}

	for _, directive := range format.directives {
		values = format.processTSVDirective(directive, values)
	}

	stringValues := make([]string, len(values))
	for index, value := range values {
		if value.quoted {
			stringValues[index] = `"` + value.value + `"`
		} else {
			stringValues[index] = value.value
		}
	}

	format.SetAppliedContent(msg, []byte(strings.Join(stringValues, format.delimiter)))
	return nil // continue
}
1253606
func (b *Base) SetFormatter(mf MessageFormatter) error {
	if mf == nil {
		return errors.New("cannot set message formatter to nil")
	}

	b.mutex.Lock()
	defer b.mutex.Unlock()
	b.formatter = mf

	return nil
}