func (c *console) PublishEvent(
	s outputs.Signaler,
	ts time.Time,
	event common.MapStr,
) error {
	var jsonEvent []byte
	var err error

	if c.pretty {
		jsonEvent, err = json.MarshalIndent(event, "", "  ")
	} else {
		jsonEvent, err = json.Marshal(event)
	}
	if err != nil {
		logp.Err("Fail to convert the event to JSON: %s", err)
		outputs.SignalCompleted(s)
		return err
	}

	if err = writeBuffer(jsonEvent); err != nil {
		goto fail
	}
	if err = writeBuffer([]byte{'\n'}); err != nil {
		goto fail
	}

	outputs.SignalCompleted(s)
	return nil
fail:
	logp.Err("Fail to write event: %s", err)
	outputs.SignalFailed(s)
	return err
}

func (mh *testMessageHandler) acknowledgeMessage(m message) {
	if mh.response == CompletedResponse {
		outputs.SignalCompleted(m.context.signal)
	} else {
		outputs.SignalFailed(m.context.signal, nil)
	}
}

func (o *outputWorker) onBulk(ctx *context, events []common.MapStr) {
	if len(events) == 0 {
		debug("output worker: no events to publish")
		outputs.SignalCompleted(ctx.signal)
		return
	}

	var sync *outputs.SyncSignal
	if ctx.sync {
		sync = outputs.NewSyncSignal()
	}

	if o.maxBulkSize < 0 || len(events) <= o.maxBulkSize {
		o.sendBulk(sync, ctx, events)
		return
	}

	// start splitting bulk request
	splits := (len(events) + (o.maxBulkSize - 1)) / o.maxBulkSize
	ctx.signal = outputs.NewSplitSignaler(ctx.signal, splits)
	for len(events) > 0 {
		sz := o.maxBulkSize
		if sz > len(events) {
			sz = len(events)
		}
		o.sendBulk(sync, ctx, events[:sz])
		events = events[sz:]
	}
}

func (m *LoadBalancerMode) onMessage(client ProtocolClient, msg eventsMessage) {
	if msg.event != nil {
		if err := client.PublishEvent(msg.event); err != nil {
			m.onFail(msg, err)
			return
		}
	} else {
		published := 0
		events := msg.events
		send := 0
		for published < len(events) {
			n, err := client.PublishEvents(events[published:])
			if err != nil {
				// retry only non-confirmed subset of events in batch
				msg.events = msg.events[published:]

				// reset attempt count if subset of message has been send
				if send > 0 {
					msg.attemptsLeft = m.maxAttempts + 1
				}
				m.onFail(msg, err)
				return
			}
			published += n
			send++
		}
	}
	outputs.SignalCompleted(msg.signaler)
}

func (s *singleConnectionMode) PublishEvents(
	trans outputs.Signaler,
	events []common.MapStr,
) error {
	published := 0
	fails := 0
	for !s.closed && (s.maxAttempts == 0 || fails < s.maxAttempts) {
		if err := s.Connect(); err != nil {
			fails++
			time.Sleep(s.waitRetry)
			continue
		}

		for published < len(events) {
			n, err := s.conn.PublishEvents(events[published:])
			if err != nil {
				break
			}

			fails = 0
			published += n
		}

		if published == len(events) {
			outputs.SignalCompleted(trans)
			return nil
		}

		time.Sleep(s.waitRetry)
		fails++
	}

	outputs.SignalFailed(trans)
	return nil
}

// PublishEvent forwards a single event. On failure PublishEvent tries to reconnect.
func (s *SingleConnectionMode) PublishEvent(
	signaler outputs.Signaler,
	event common.MapStr,
) error {
	fails := 0

	var err error

	for !s.closed && (s.maxAttempts == 0 || fails < s.maxAttempts) {
		if err = s.connect(); err != nil {
			logp.Info("Connecting error publishing event (retrying): %s", err)

			fails++
			time.Sleep(s.waitRetry)
			continue
		}
		if err := s.conn.PublishEvent(event); err != nil {
			logp.Info("Error publishing event (retrying): %s", err)

			fails++
			continue
		}

		outputs.SignalCompleted(signaler)
		return nil
	}

	outputs.SignalFailed(signaler, err)
	return nil
}

// PublishEvent forwards a single event. On failure PublishEvent tries to reconnect.
func (f *FailOverConnectionMode) PublishEvent(
	signaler outputs.Signaler,
	event common.MapStr,
) error {
	fails := 0
	var err error
	for !f.closed && (f.maxAttempts == 0 || fails < f.maxAttempts) {
		if err := f.connect(f.active); err != nil {
			logp.Info("Connecting error publishing events (retrying): %s", err)
			fails++
			time.Sleep(f.waitRetry)
			continue
		}

		if err := f.conns[f.active].PublishEvent(event); err != nil {
			logp.Info("Error publishing events (retrying): %s", err)
			fails++
			continue
		}

		outputs.SignalCompleted(signaler)
		return nil
	}

	outputs.SignalFailed(signaler, err)
	return nil
}

func (m *LoadBalancerMode) onMessage(client ProtocolClient, msg eventsMessage) {
	if msg.event != nil {
		if err := client.PublishEvent(msg.event); err != nil {
			msg.attemptsLeft--
			m.onFail(msg, err)
			return
		}
	} else {
		events := msg.events
		total := len(events)
		var backoffCount uint

		for len(events) > 0 {
			var err error

			events, err = client.PublishEvents(events)
			if err != nil {
				msg.attemptsLeft--

				// reset attempt count if subset of messages has been processed
				if len(events) < total {
					msg.attemptsLeft = m.maxAttempts
				}

				if err != ErrTempBulkFailure {
					// retry non-published subset of events in batch
					msg.events = events
					m.onFail(msg, err)
					return
				}

				if m.maxAttempts > 0 && msg.attemptsLeft <= 0 {
					// no more attempts left => drop
					outputs.SignalFailed(msg.signaler, err)
					return
				}

				// wait before retry
				backoff := time.Duration(int64(m.waitRetry) * (1 << backoffCount))
				if backoff > m.maxWaitRetry {
					backoff = m.maxWaitRetry
				} else {
					backoffCount++
				}
				select {
				case <-m.done: // shutdown
					outputs.SignalFailed(msg.signaler, err)
					return
				case <-time.After(backoff):
				}

				// reset total count for temporary failure loop
				total = len(events)
			}
		}
	}
	outputs.SignalCompleted(msg.signaler)
}

func (out *redisOutput) BulkPublish(
	signal outputs.Signaler,
	ts time.Time,
	events []common.MapStr,
) error {
	if !out.connected {
		logp.Debug("output_redis", "Droping pkt ...")
		return errors.New("Not connected")
	}

	command := "RPUSH"
	if out.DataType == RedisChannelType {
		command = "PUBLISH"
	}

	if len(events) == 1 { // single event
		event := events[0]
		jsonEvent, err := json.Marshal(event)
		if err != nil {
			logp.Err("Fail to convert the event to JSON: %s", err)
			outputs.SignalCompleted(signal)
			return err
		}

		_, err = out.Conn.Do(command, out.Index, string(jsonEvent))
		outputs.Signal(signal, err)
		out.onFail(err)
		return err
	}

	for _, event := range events {
		jsonEvent, err := json.Marshal(event)
		if err != nil {
			logp.Err("Fail to convert the event to JSON: %s", err)
			continue
		}
		err = out.Conn.Send(command, out.Index, string(jsonEvent))
		if err != nil {
			outputs.SignalFailed(signal, err)
			out.onFail(err)
			return err
		}
	}
	if err := out.Conn.Flush(); err != nil {
		outputs.Signal(signal, err)
		out.onFail(err)
		return err
	}
	_, err := out.Conn.Receive()
	outputs.Signal(signal, err)
	out.onFail(err)
	return err
}

// publish is used to publish events using the configured protocol client.
// It provides general error handling and back off support used on failed
// send attempts. To be used by PublishEvent and PublishEvents.
// The send callback will try to progress sending traffic and returns kind of
// progress made in ok or resetFail. If ok is set to true, send finished
// processing events. If ok is false but resetFail is set, send was partially
// successful. If send was partially successful, the fail counter is reset thus up
// to maxAttempts send attempts without any progress might be executed.
func (s *SingleConnectionMode) publish(
	signaler outputs.Signaler,
	send func() (ok bool, resetFail bool),
) error {
	fails := 0
	var backoffCount uint
	var err error

	for !s.closed && (s.maxAttempts == 0 || fails < s.maxAttempts) {
		ok := false
		resetFail := false

		if err := s.connect(); err != nil {
			logp.Info("Connecting error publishing events (retrying): %s", err)
			goto sendFail
		}

		ok, resetFail = send()
		if !ok {
			goto sendFail
		}

		outputs.SignalCompleted(signaler)
		return nil

	sendFail:
		fails++
		if resetFail {
			fails = 0
		}
		if s.maxAttempts > 0 && fails == s.maxAttempts {
			// max number of attempts reached
			break
		}

		logp.Info("send fail")
		backoff := time.Duration(int64(s.waitRetry) * (1 << backoffCount))
		if backoff > s.maxWaitRetry {
			backoff = s.maxWaitRetry
		} else {
			backoffCount++
		}
		logp.Info("backoff retry: %v", backoff)
		time.Sleep(backoff)
	}

	outputs.SignalFailed(signaler, err)
	return nil
}

// PublishEvents tries to publish the events with retries if connection becomes
// unavailable. On failure PublishEvents tries to reconnect.
func (s *SingleConnectionMode) PublishEvents(
	signaler outputs.Signaler,
	events []common.MapStr,
) error {
	fails := 0
	var backoffCount uint
	var err error

	for !s.closed && (s.maxAttempts == 0 || fails < s.maxAttempts) {
		if err := s.connect(); err != nil {
			logp.Info("Connecting error publishing events (retrying): %s", err)
			goto sendFail
		}

		for len(events) > 0 {
			var err error
			events, err = s.conn.PublishEvents(events)
			if err != nil {
				logp.Info("Error publishing events (retrying): %s", err)
				break
			}

			fails = 0
		}

		if len(events) == 0 {
			outputs.SignalCompleted(signaler)
			return nil
		}

	sendFail:
		logp.Info("send fail")
		backoff := time.Duration(int64(s.waitRetry) * (1 << backoffCount))
		if backoff > s.maxWaitRetry {
			backoff = s.maxWaitRetry
		} else {
			backoffCount++
		}
		logp.Info("backoff retry: %v", backoff)
		time.Sleep(backoff)
		fails++
	}

	outputs.SignalFailed(signaler, err)
	return nil
}

func (o *outputWorker) onMessage(m message) {

	if m.event != nil {
		debug("output worker: publish single event")
		ts := time.Time(m.event["timestamp"].(common.Time))
		_ = o.out.PublishEvent(m.signal, ts, m.event)
	} else {
		if len(m.events) == 0 {
			debug("output worker: no events to publish")
			outputs.SignalCompleted(m.signal)
			return
		}

		debug("output worker: publish %v events", len(m.events))
		ts := time.Time(m.events[0]["timestamp"].(common.Time))
		_ = o.out.BulkPublish(m.signal, ts, m.events)
	}
}

func (o *outputWorker) onEvent(ctx *context, event common.MapStr) {
	debug("output worker: publish single event")
	ts := time.Time(event["@timestamp"].(common.Time)).UTC()

	if !ctx.sync {
		_ = o.out.PublishEvent(ctx.signal, ts, event)
		return
	}

	signal := outputs.NewSyncSignal()
	for {
		o.out.PublishEvent(signal, ts, event)
		if signal.Wait() {
			outputs.SignalCompleted(ctx.signal)
			break
		}
	}
}

func (out *fileOutput) PublishEvent(
	trans outputs.Signaler,
	ts time.Time,
	event common.MapStr,
) error {
	jsonEvent, err := json.Marshal(event)
	if err != nil {
		// mark as success so event is not send again.
		outputs.SignalCompleted(trans)

		logp.Err("Fail to convert the event to JSON: %s", err)
		return err
	}

	err = out.rotator.WriteLine(jsonEvent)
	outputs.Signal(trans, err)
	return err
}

// PublishEvents tries to publish the events with retries if connection becomes
// unavailable. On failure PublishEvents tries to connect to another configured
// connection by random.
func (f *FailOverConnectionMode) PublishEvents(
	trans outputs.Signaler,
	events []common.MapStr,
) error {
	published := 0
	fails := 0
	var err error

	for !f.closed && (f.maxAttempts == 0 || fails < f.maxAttempts) {
		if err = f.connect(f.active); err != nil {
			logp.Info("Connecting error publishing events (retrying): %s", err)
			fails++
			time.Sleep(f.waitRetry)
			continue
		}

		// loop until all events have been send in case client supports partial sends
		for published < len(events) {
			conn := f.conns[f.active]
			n, err := conn.PublishEvents(events[published:])
			if err != nil {
				logp.Info("Error publishing events (retrying): %s", err)
				break
			}
			published += n
		}

		if published == len(events) {
			outputs.SignalCompleted(trans)
			return nil
		}

		// TODO(sissel): Track how frequently we timeout and reconnect. If we're
		// timing out too frequently, there's really no point in timing out since
		// basically everything is slow or down. We'll want to ratchet up the
		// timeout value slowly until things improve, then ratchet it down once
		// things seem healthy.
		time.Sleep(f.waitRetry)
		fails++
	}

	outputs.SignalFailed(trans, err)
	return nil
}

// publish is used to publish events using the configured protocol client.
// It provides general error handling and back off support used on failed
// send attempts. To be used by PublishEvent and PublishEvents.
// The send callback will try to progress sending traffic and returns kind of
// progress made in ok or resetFail. If ok is set to true, send finished
// processing events. If ok is false but resetFail is set, send was partially
// successful. If send was partially successful, the fail counter is reset thus up
// to maxAttempts send attempts without any progress might be executed.
func (f *FailOverConnectionMode) publish(
	signaler outputs.Signaler,
	send func() (ok bool, resetFail bool),
) error {
	fails := 0
	var err error

	// TODO: we want back off support here? Fail over normally will try another
	// connection.

	for !f.closed && (f.maxAttempts == 0 || fails < f.maxAttempts) {
		ok := false
		resetFail := false

		if err = f.connect(f.active); err != nil {
			logp.Info("Connecting error publishing events (retrying): %s", err)
			goto sendFail
		}

		ok, resetFail = send()
		if !ok {
			goto sendFail
		}

		outputs.SignalCompleted(signaler)
		return nil

	sendFail:
		fails++
		if resetFail {
			fails = 0
		}
		if f.maxAttempts > 0 && fails == f.maxAttempts {
			// max number of attempts reached
			break
		}

		time.Sleep(f.waitRetry)
	}

	outputs.SignalFailed(signaler, err)
	return nil
}

func (o *outputWorker) sendBulk(
	sync *outputs.SyncSignal,
	ctx *context,
	events []common.MapStr,
) {
	debug("output worker: publish %v events", len(events))
	ts := time.Time(events[0]["@timestamp"].(common.Time)).UTC()

	if sync == nil {
		err := o.out.BulkPublish(ctx.signal, ts, events)
		if err != nil {
			logp.Info("Error bulk publishing events: %s", err)
		}
		return
	}

	for done := false; !done; done = sync.Wait() {
		o.out.BulkPublish(sync, ts, events)
	}
	outputs.SignalCompleted(ctx.signal)
}

// PublishEvents tries to publish the events with retries if connection becomes
// unavailable. On failure PublishEvents tries to reconnect.
func (s *SingleConnectionMode) PublishEvents(
	signaler outputs.Signaler,
	events []common.MapStr,
) error {
	published := 0
	fails := 0
	var err error

	for !s.closed && (s.maxAttempts == 0 || fails < s.maxAttempts) {
		if err := s.connect(); err != nil {
			logp.Info("Connecting error publishing events (retrying): %s", err)

			fails++
			time.Sleep(s.waitRetry)
			continue
		}

		for published < len(events) {
			n, err := s.conn.PublishEvents(events[published:])
			if err != nil {
				logp.Info("Error publishing events (retrying): %s", err)
				break
			}

			fails = 0
			published += n
		}

		if published == len(events) {
			outputs.SignalCompleted(signaler)
			return nil
		}

		time.Sleep(s.waitRetry)
		fails++
	}

	outputs.SignalFailed(signaler, err)
	return nil
}

func (o *outputWorker) onMessage(m message) {

	if m.event != nil {
		debug("output worker: publish single event")
		ts := time.Time(m.event["@timestamp"].(common.Time)).UTC()
		_ = o.out.PublishEvent(m.signal, ts, m.event)
	} else {
		if len(m.events) == 0 {
			debug("output worker: no events to publish")
			outputs.SignalCompleted(m.signal)
			return
		}

		debug("output worker: publish %v events", len(m.events))
		ts := time.Time(m.events[0]["@timestamp"].(common.Time)).UTC()
		err := o.out.BulkPublish(m.signal, ts, m.events)

		if err != nil {
			logp.Info("Error bulk publishing events: %s", err)
		}
	}
}

func (p *preprocessor) onMessage(m message) {
	publisher := p.pub
	single := false
	events := m.events
	if m.event != nil {
		single = true
		events = []common.MapStr{m.event}
	}

	var ignore []int // indices of events to be removed from events

	debug("preprocessor")

	for i, event := range events {
		// validate some required field
		if err := filterEvent(event); err != nil {
			logp.Err("Publishing event failed: %v", err)
			ignore = append(ignore, i)
			continue
		}

		// update address and geo-ip information. Ignore event
		// if address is invalid or event is found to be a duplicate
		ok := updateEventAddresses(publisher, event)
		if !ok {
			ignore = append(ignore, i)
			continue
		}

		// add additional Beat meta data
		event["beat"] = common.MapStr{
			"name":     publisher.name,
			"hostname": publisher.hostname,
			"version":  publisher.version,
		}
		if len(publisher.tags) > 0 {
			event["tags"] = publisher.tags
		}

		if logp.IsDebug("publish") {
			PrintPublishEvent(event)
		}
	}

	// return if no event is left
	if len(ignore) == len(events) {
		debug("no event left, complete send")
		outputs.SignalCompleted(m.context.signal)
		return
	}

	// remove invalid events.
	// TODO: is order important? Removal can be turned into O(len(ignore)) by
	//       copying last element into idx and doing
	//       events=events[:len(events)-len(ignore)] afterwards
	// Alternatively filtering could be implemented like:
	//   https://github.com/golang/go/wiki/SliceTricks#filtering-without-allocating
	for i := len(ignore) - 1; i >= 0; i-- {
		idx := ignore[i]
		debug("remove event[%v]", idx)
		events = append(events[:idx], events[idx+1:]...)
	}

	if publisher.disabled {
		debug("publisher disabled")
		outputs.SignalCompleted(m.context.signal)
		return
	}

	debug("preprocessor forward")
	if single {
		p.handler.onMessage(message{context: m.context, event: events[0]})
	} else {
		p.handler.onMessage(message{context: m.context, events: events})
	}
}