func (p *syncPublisher) send(m message) bool {
	if p.pub.disabled {
		debug("publisher disabled")
		outputs.SignalCompleted(m.context.signal)
		return true
	}

	signal := m.context.signal
	sync := outputs.NewSyncSignal()
	if len(p.pub.Output) > 1 {
		m.context.signal = outputs.NewSplitSignaler(sync, len(p.pub.Output))
	} else {
		m.context.signal = sync
	}

	for _, o := range p.pub.Output {
		o.send(m)
	}

	ok := sync.Wait()
	if ok {
		outputs.SignalCompleted(signal)
	} else if signal != nil {
		signal.Failed()
	}
	return ok
}

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

	if c.config.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:
	if opts.Guaranteed {
		logp.Critical("Unable to publish events to console: %v", err)
	}
	outputs.SignalFailed(s, err)
	return err
}

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:
	outputs.SignalFailed(s, err)
	return err
}

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

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

	err = out.rotator.WriteLine(jsonEvent)
	if err != nil {
		if opts.Guaranteed {
			logp.Critical("Unable to write events to file: %s", err)
		} else {
			logp.Err("Error when writing line to file: %s", err)
		}
	}
	outputs.Signal(trans, err)
	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:]
	}
}

// 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,
	opts outputs.Options,
	send func() (ok bool, resetFail bool),
) error {
	fails := 0
	var backoffCount uint
	var err error

	guaranteed := opts.Guaranteed || s.maxAttempts == 0
	for !s.closed && (guaranteed || 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
		}

		debug("send completed")
		outputs.SignalCompleted(signaler)
		return nil

	sendFail:
		fails++
		if resetFail {
			debug("reset fails")
			fails = 0
		}

		if !guaranteed && (s.maxAttempts > 0 && fails == s.maxAttempts) {
			// max number of attempts reached
			debug("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)
	}

	debug("messages dropped")
	messagesDropped.Add(1)
	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
					dropping(msg)
					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 handlePublishEventsResult(
	m *AsyncLoadBalancerMode,
	msg eventsMessage,
) func([]common.MapStr, error) {
	total := len(msg.events)
	return func(events []common.MapStr, err error) {
		debug("handlePublishEventsResult")

		if err != nil {
			debug("handle publish error: %v", err)

			if msg.attemptsLeft > 0 {
				msg.attemptsLeft--
			}

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

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

			if m.maxAttempts > 0 && msg.attemptsLeft == 0 {
				// no more attempts left => drop
				dropping(msg, &m.ws)
				return
			}

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

		// re-insert non-published events into pipeline
		if len(events) != 0 {
			debug("add non-published events back into pipeline: %v", len(events))
			msg.events = events
			if ok := m.forwardEvent(m.retries, msg); !ok {
				dropping(msg, &m.ws)
			}
			return
		}

		// all events published -> signal success
		debug("async bulk publish success")
		outputs.SignalCompleted(msg.signaler)
		m.ws.DoneEvent()
	}
}

func handlePublishEventResult(m *AsyncLoadBalancerMode, msg eventsMessage) func(error) {
	return func(err error) {
		if err != nil {
			if msg.attemptsLeft > 0 {
				msg.attemptsLeft--
			}
			m.onFail(false, msg, err)
		} else {
			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,
	opts outputs.Options,
	send func() (ok bool, resetFail bool),
) error {
	fails := 0
	var err error

	guaranteed := opts.Guaranteed || s.maxAttempts == 0
	for !s.closed && (guaranteed || 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
		}

		debug("send completed")
		s.backoff.Reset()
		outputs.SignalCompleted(signaler)
		return nil

	sendFail:
		logp.Info("send fail")
		s.backoff.Wait()

		fails++
		if resetFail {
			debug("reset fails")
			fails = 0
		}

		if !guaranteed && (s.maxAttempts > 0 && fails == s.maxAttempts) {
			// max number of attempts reached
			debug("max number of attempts reached")
			break
		}
	}

	debug("messages dropped")
	messagesDropped.Add(1)
	outputs.SignalFailed(signaler, 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,
	opts outputs.Options,
	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.

	guaranteed := opts.Guaranteed || f.maxAttempts == 0
	for !f.closed && (guaranteed || 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
		}

		debug("send completed")
		outputs.SignalCompleted(signaler)
		return nil

	sendFail:
		fails++
		if resetFail {
			debug("reset fails")
			fails = 0
		}
		if !guaranteed && (f.maxAttempts > 0 && fails == f.maxAttempts) {
			// max number of attempts reached
			debug("max number of attempts reached")
			break
		}

		time.Sleep(f.waitRetry)
	}

	debug("messages dropped")
	messagesDropped.Add(1)
	outputs.SignalFailed(signaler, err)
	return nil
}

func TestPublisherModes(t *testing.T) {
	tests := []struct {
		title string
		async bool
		order []int
	}{
		{"sync", false, []int{1, 2, 3, 4, 5, 6}},
		{"async ordered signal", true, []int{1, 2, 3, 4, 5, 6}},
		{"async out of order signal", true, []int{5, 2, 3, 1, 4, 6}},
	}

	for i, test := range tests {
		t.Logf("run publisher test (%v): %v", i, test.title)

		pubChan := make(chan []*input.FileEvent, len(test.order)+1)
		regChan := make(chan []*input.FileEvent, len(test.order)+1)
		client := publisher.ExtChanClient{make(chan publisher.PublishMessage)}

		pub := newPublisher(test.async, pubChan, regChan, client)
		pub.Start()

		var events [][]*input.FileEvent
		for i := range test.order {
			tmp := makeEvents(fmt.Sprintf("msg: %v", i), 1)
			pubChan <- tmp
			events = append(events, tmp)
		}

		var msgs []publisher.PublishMessage
		for _ = range test.order {
			m := <-client.Channel
			msgs = append(msgs, m)
		}

		for _, i := range test.order {
			outputs.SignalCompleted(msgs[i-1].Context.Signal)
		}

		var regEvents [][]*input.FileEvent
		for _ = range test.order {
			regEvents = append(regEvents, <-regChan)
		}
		pub.Stop()

		// validate order
		for i := range events {
			assert.Equal(t, events[i], regEvents[i])
		}
	}
}

func (p *syncPublisher) forward(m message) bool {
	sync := outputs.NewSyncSignal()
	signal := m.context.signal
	m.context.signal = sync
	p.send(m)
	if sync.Wait() {
		outputs.SignalCompleted(signal)
		return true
	}
	if signal != nil {
		signal.Failed()
	}
	return false
}

// 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
}

func (p *asyncPublisher) send(m message) {
	if p.pub.disabled {
		debug("publisher disabled")
		outputs.SignalCompleted(m.context.signal)
		return
	}

	// m.signal is not set yet. But a async client type supporting signals might
	// be implemented in the future.
	// If m.signal is nil, NewSplitSignaler will return nil -> signaler will
	// only set if client did send one
	if m.context.signal != nil && len(p.outputs) > 1 {
		m.context.signal = outputs.NewSplitSignaler(m.context.signal, len(p.outputs))
	}
	for _, o := range p.outputs {
		o.send(m)
	}
}

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
		}
	}
}

// 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 (c *cmp) PublishEvent(
	s outputs.Signaler,
	opts outputs.Options,
	event common.MapStr,
) error {
	var err error

	if err = postToCmp(c, event); err != nil {
		goto fail
	}

	outputs.SignalCompleted(s)
	return nil
fail:
	if opts.Guaranteed {
		logp.Critical("Unable to publish events to cmp: %v", err)
	}
	outputs.SignalFailed(s, err)
	return err
}