func (s *SingleConnectionMode) PublishEvents(
	signaler outputs.Signaler,
	opts outputs.Options,
	events []common.MapStr,
) error {
	return s.publish(signaler, opts, func() (bool, bool) {
		for len(events) > 0 {
			var err error
			total := len(events)
			events, err = s.conn.PublishEvents(events)
			if err != nil {
				logp.Info("Error publishing events (retrying): %s", err)
				madeProgress := len(events) < total
				return false, madeProgress
			}
		}
		return true, false
	})
}
PublishEvents tries to publish the events with retries if connection becomes
unavailable. On failure PublishEvents tries to reconnect.
package mode
import (
	"errors"
	"time"
	"github.com/elastic/beats/libbeat/common"
	"github.com/elastic/beats/libbeat/logp"
	"github.com/elastic/beats/libbeat/outputs"
)
type SingleConnectionMode struct {
	conn ProtocolClient
	closed bool  
	timeout time.Duration  
	backoff *backoff
	maxAttempts int
}
var (
	errNeedBackoff = errors.New("need to backoff")
)
func NewSingleConnectionMode(
	client ProtocolClient,
	maxAttempts int,
	waitRetry, timeout, maxWaitRetry time.Duration,
) (*SingleConnectionMode, error) {
	s := &SingleConnectionMode{
		conn: client,
		timeout:     timeout,
		backoff:     newBackoff(nil, waitRetry, maxWaitRetry),
		maxAttempts: maxAttempts,
	}
	_ = s.connect()  
	return s, nil
}
func (s *SingleConnectionMode) connect() error {
	if s.conn.IsConnected() {
		return nil
	}
	return s.conn.Connect(s.timeout)
}
func (s *SingleConnectionMode) Close() error {
	s.closed = true
	return s.conn.Close()
}