// ApplyFormatter update message payload
431621
func (format *CollectdToInflux08) 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 JSON lines
	influxData := tio.NewByteStream(len(contentData))
	name := format.createMetricName(collectdData.Plugin,
		collectdData.PluginInstance,
		collectdData.PluginType,
		collectdData.TypeInstance,
		collectdData.Host)

	for _, value := range collectdData.Values {
		fmt.Fprintf(&influxData, `{"name": "%s", "columns": ["time", "value"], "points":[[%d, %f]]},`, name, int32(collectdData.Time), value)
	}

	format.SetAppliedContent(msg, influxData.Bytes())
	return nil
}
899595
func applyStyle(style string, useColor bool, format string, a ...interface{}) (string, error) {
	p := message.NewPrinter(preferredLanguage)
	for i, x := range a {
		if _, ok := x.(int); ok {
			a[i] = number.Decimal(x, number.NoSeparator())
		}
	}
	out := p.Sprintf(format, a...)

	s, ok := styles[style]
	if !s.OmitNewline {
		out += "\n"
	}

	// Similar to CSS styles, if no style matches, output an unformatted string.
	if !ok {
		return p.Sprintf(format, a...), fmt.Errorf("unknown style: %q", style)
	}

	if !useColor {
		return applyPrefix(lowPrefix(s), out), nil
	}
	return applyPrefix(s.Prefix, out), 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
}
1341030
func NewFormatter() *Formatter {
	return &Formatter{
		KeyColor:        color.New(color.FgBlue, color.Bold),
		StringColor:     color.New(color.FgGreen, color.Bold),
		BoolColor:       color.New(color.FgYellow, color.Bold),
		NumberColor:     color.New(color.FgCyan, color.Bold),
		NullColor:       color.New(color.FgBlack, color.Bold),
		StringMaxLength: 0,
		DisabledColor:   false,
		Indent:          2,
		Newline:         "\n",
	}
}
860864
func (format *Timestamp) ApplyFormatter(msg *core.Message) error {
	timestampStr := msg.GetCreationTime().Format(format.timestampFormat)
	content := format.GetAppliedContent(msg)

	dataSize := len(timestampStr) + len(content)
	payload := make([]byte, dataSize)

	offset := copy(payload, []byte(timestampStr))
	copy(payload[offset:], content)

	format.SetAppliedContent(msg, payload)
	return nil
}
1221104
func (format *Envelope) ApplyFormatter(msg *core.Message) error {
	prefixLen := len(format.prefix)
	postfixLen := len(format.postfix)
	content := format.GetAppliedContent(msg)
	offset := 0

	payload := make([]byte, prefixLen+len(content)+postfixLen)

	if prefixLen > 0 {
		offset = copy(payload, format.prefix)
	}

	offset += copy(payload[offset:], content)

	if postfixLen > 0 {
		copy(payload[offset:], format.postfix)
	}

	format.SetAppliedContent(msg, payload)
	return nil
}
1036171
func MakeXMPPFormatter(name string) MessageFormatter {
	return func(m message.Composer) (string, error) {
		return fmt.Sprintf(completeFormatTmpl, name, m.Priority(), m.String()), nil
	}
}
80575
func (format *SplitToJSON) ApplyFormatter(msg *core.Message) error {
	components := bytes.Split(format.GetAppliedContent(msg), format.token)
	maxIdx := tmath.MinI(len(format.keys), len(components))
	jsonData := ""

	switch {
	case maxIdx == 0:
	case maxIdx == 1:
		jsonData = fmt.Sprintf("{%s:\"%s\"}", format.keys[0], components[0])
	default:
		for i := 0; i < maxIdx; i++ {
			key := tstrings.EscapeJSON(format.keys[i])
			value := string(components[i])
			if isJSON, _, _ := tstrings.IsJSON(components[i]); !format.keepJSON || !isJSON {
				value = "\"" + tstrings.EscapeJSON(value) + "\""
			}

			switch {
			case i == 0:
				jsonData = fmt.Sprintf("{\"%s\":%s", key, value)
			case i == maxIdx-1:
				jsonData = fmt.Sprintf("%s,\"%s\":%s}", jsonData, key, value)
			default:
				jsonData = fmt.Sprintf("%s,\"%s\":%s", jsonData, key, value)
			}
		}
	}

	format.SetAppliedContent(msg, []byte(jsonData))
	return nil
}
407116
func (format *GrokToJSON) ApplyFormatter(msg *core.Message) error {
	content := format.GetAppliedContent(msg)

	values, err := format.applyGrok(string(content[:]))
	if err != nil {
		return err
	}

	serialized, err := json.Marshal(values)
	if err != nil {
		return err
	}

	format.SetAppliedContent(msg, serialized)
	return nil
}
869813
func (serviceMap *ServiceMap) apply(changes *ServiceChangeTracker, UDPStaleClusterIP sets.String) {
	changes.lock.Lock()
	defer changes.lock.Unlock()
	for _, change := range changes.items {
		serviceMap.merge(change.current)
		// filter out the Update event of current changes from previous changes before calling unmerge() so that can
		// skip deleting the Update events.
		change.previous.filter(change.current)
		serviceMap.unmerge(change.previous, UDPStaleClusterIP)
	}
	// clear changes after applying them to ServiceMap.
	changes.items = make(map[types.NamespacedName]*serviceChange)
	return
}
974239
func (formatters FormatterArray) ApplyFormatter(msg *Message) error {
	for _, formatter := range formatters {
		if formatter.CanBeApplied(msg) {
			if err := formatter.ApplyFormatter(msg); err != nil {
				return err
			}
		}
	}
	return nil
}
603382
func (format *Trim) ApplyFormatter(msg *core.Message) error {
	content := format.GetAppliedContent(msg)
	offset := len(content)

	if len(format.rightSeparator) > 0 {
		rightIdx := bytes.LastIndex(content, format.rightSeparator)
		if rightIdx > 0 {
			offset = rightIdx
		}
	}
	format.extendContent(&content, offset-format.rightOffset)

	offset = format.leftOffset
	if len(format.leftSeparator) > 0 {
		leftIdx := bytes.Index(msg.GetPayload(), format.leftSeparator)
		leftIdx++
		if leftIdx > 0 {
			offset += leftIdx
		}
	}
	content = content[offset:]

	format.SetAppliedContent(msg, content)
	return nil
}
1250500
func (il IfLogger) SetFormatter(f Formatter) ContextLogger {
	return &contextLogger{
		isTrue:    il.ok(),
		formatter: f,
	}
}
1155417
func (format *Base64Encode) ApplyFormatter(msg *core.Message) error {
	encoded := format.getEncodedContent(format.GetAppliedContent(msg))
	format.SetAppliedContent(msg, encoded)

	return nil
}
125422
func (p Patch) Apply(doc []byte) ([]byte, error) {
	return p.ApplyIndent(doc, "")
}
87370
func (filters FilterArray) ApplyFilter(msg *Message) (FilterResult, error) {
	for _, filter := range filters {
		result, err := filter.ApplyFilter(msg)
		if err != nil {
			return result, err
		}

		if result != FilterResultMessageAccept {
			return result, nil
		}
	}
	return FilterResultMessageAccept, nil
}
243894
func (logger *Logger) SetFormatter(formatter Formatter) {
	logger.mu.Lock()
	defer logger.mu.Unlock()
	logger.Formatter = formatter
}