elasticstream/source/elastic/ack.go

package elastic

import (
	"context"
	"fmt"

	"elasticstream/source"
)

func (c *Client) Ack(ctx context.Context, position source.Position) error {
	curr := c.offsets[position.Index]

	fmt.Println("curr:", curr)
	fmt.Println("asked:", position.Pos)

	for _, p := range c.positions {
		if p.Index == position.Index {
			fmt.Println("initial:", p.Pos)
			if p.Pos > position.Pos {
				return fmt.Errorf("not acknowledged pos less than initial position")
			}
		}
	}

	if curr < position.Pos {
		return fmt.Errorf("not acknowledged pos more than current position")
	}
	return nil
}