<commit_before>package publisher

import (
	"time"

	"github.com/elastic/libbeat/common"
	"github.com/elastic/libbeat/publisher"
)

// PlusPublisher is a Publisher that publishes Nginx Plus status.
type PlusPublisher struct {
	client publisher.Client
}

// NewPlusPublisher constructs a new PlusPublisher.
func NewPlusPublisher(c publisher.Client) *PlusPublisher {
	return &PlusPublisher{client: c}
}

// Publish Nginx Plus status.
func (p *PlusPublisher) Publish(s map[string]interface{}) {
	zones := s["server_zones"].([]interface{})
	delete(s, "server_zones")

	upstreams := s["upstreams"].([]interface{})
	delete(s, "upstreams")

	caches := s["caches"].([]interface{})
	delete(s, "caches")

	stream := s["stream"]
	delete(s, "stream")

	p.client.PublishEvent(common.MapStr{
		"@timestamp": common.Time(time.Now()),
		"type":       "nginx",
		"nginx":      s,
	})

	for _, i := range zones {
		p.client.PublishEvent(common.MapStr{
			"@timestamp": common.Time(time.Now()),
			"type":       "zone",
			"zone":       i,
		})
	}

	for _, i := range upstreams {
		p.client.PublishEvent(common.MapStr{
			"@timestamp": common.Time(time.Now()),
			"type":       "upstream",
			"upstream":   i,
		})
	}

	for _, i := range caches {
		p.client.PublishEvent(common.MapStr{
			"@timestamp": common.Time(time.Now()),
			"type":       "cache",
			"cache":      i,
		})
	}

	p.client.PublishEvent(common.MapStr{
		"@timestamp": common.Time(time.Now()),
		"type":       "stream",
		"stream":     stream,
	})
}<commit_msg>Use a shared timestamp for all events published in Nginx Plus at once<commit_after>package publisher

import (
	"time"

	"github.com/elastic/libbeat/common"
	"github.com/elastic/libbeat/publisher"
)

// PlusPublisher is a Publisher that publishes Nginx Plus status.
type PlusPublisher struct {
	client publisher.Client
}

// NewPlusPublisher constructs a new PlusPublisher.
func NewPlusPublisher(c publisher.Client) *PlusPublisher {
	return &PlusPublisher{client: c}
}

// Publish Nginx Plus status.
func (p *PlusPublisher) Publish(s map[string]interface{}) {
	zones := s["server_zones"].([]interface{})
	delete(s, "server_zones")

	upstreams := s["upstreams"].([]interface{})
	delete(s, "upstreams")

	caches := s["caches"].([]interface{})
	delete(s, "caches")

	stream := s["stream"]
	delete(s, "stream")

	now := common.Time(time.Now())

	p.client.PublishEvent(common.MapStr{
		"@timestamp": now,
		"type":       "nginx",
		"nginx":      s,
	})

	for _, i := range zones {
		p.client.PublishEvent(common.MapStr{
			"@timestamp": now,
			"type":       "zone",
			"zone":       i,
		})
	}

	for _, i := range upstreams {
		p.client.PublishEvent(common.MapStr{
			"@timestamp": now,
			"type":       "upstream",
			"upstream":   i,
		})
	}

	for _, i := range caches {
		p.client.PublishEvent(common.MapStr{
			"@timestamp": now,
			"type":       "cache",
			"cache":      i,
		})
	}

	p.client.PublishEvent(common.MapStr{
		"@timestamp": now,
		"type":       "stream",
		"stream":     stream,
	})
}
{'fraction_non_alphanumeric': 0.17347956131605186, 'fraction_numerical': 0.0, 'mean_word_length': 3.6165644171779143, 'pattern_counts': {'":': 30, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3577524', 'n_tokens_mistral': 1182, 'n_tokens_neox': 1060, 'n_words': 222}