Make event acknowledgment asynchronous in shipper output

libbeat/outputs/shipper/api/shipper_mock.go

@@ -19,6 +19,8 @@ package api
 
 import (
 	context "context"
+	"errors"
+	"time"
 
 	pb "github.com/elastic/elastic-agent-shipper-client/pkg/proto"
 	"github.com/elastic/elastic-agent-shipper-client/pkg/proto/messages"
@@ -29,23 +31,29 @@ import (
 func NewProducerMock(cap int) *ProducerMock {
 	id, _ := uuid.NewV4()
 	return &ProducerMock{
-		UUID: id.String(),
+		uuid: id.String(),
 		Q:    make([]*messages.Event, 0, cap),
 	}
 }
 
 type ProducerMock struct {
 	pb.UnimplementedProducerServer
-	Q     []*messages.Event
-	UUID  string
-	Error error
+	Q              []*messages.Event
+	uuid           string
+	AcceptedCount  uint32
+	persistedIndex uint64
+	Error          error
 }
 
 func (p *ProducerMock) PublishEvents(ctx context.Context, r *messages.PublishRequest) (*messages.PublishReply, error) {
 	if p.Error != nil {
 		return nil, p.Error
 	}
 
+	if r.Uuid != p.uuid {
+		return nil, errors.New("UUID does not match")
+	}
+
 	resp := &messages.PublishReply{}
 
 	for _, e := range r.Events {
@@ -55,16 +63,44 @@ func (p *ProducerMock) PublishEvents(ctx context.Context, r *messages.PublishReq
 
 		p.Q = append(p.Q, e)
 		resp.AcceptedCount++
+		if resp.AcceptedCount == p.AcceptedCount {
+			break
+		}
 	}
 
 	resp.AcceptedIndex = uint64(len(p.Q))
 
 	return resp, nil
 }
 
+func (p *ProducerMock) Persist(count uint64) {
+	p.persistedIndex = count
+}
+
 func (p *ProducerMock) PersistedIndex(req *messages.PersistedIndexRequest, producer pb.Producer_PersistedIndexServer) error {
-	return producer.Send(&messages.PersistedIndexReply{
-		Uuid:           p.UUID,
-		PersistedIndex: uint64(len(p.Q)),
+	err := producer.Send(&messages.PersistedIndexReply{
+		Uuid:           p.uuid,
+		PersistedIndex: p.persistedIndex,
 	})
+	if err != nil {
+		return err
+	}
+
+	if !req.PollingInterval.IsValid() || req.PollingInterval.AsDuration() == 0 {
+		return nil
+	}
+
+	ticker := time.NewTicker(req.PollingInterval.AsDuration())
+	defer ticker.Stop()
+
+	for range ticker.C {
+		err = producer.Send(&messages.PersistedIndexReply{
+			Uuid:           p.uuid,
+			PersistedIndex: p.persistedIndex,
+		})
+		if err != nil {
+			return err
+		}
+	}
+	return nil
 }