elasticstream/source/elastic/open.go

package elastic

import (
	"context"
	"log"

	"elasticstream/opencdc"
	"elasticstream/source"

	"github.com/elastic/go-elasticsearch/v8"
)

func (c *Client) Open(ctx context.Context, positions []source.Position) error {
	log.Println(">>>> elastic.Open()")
	defer log.Println("<<<< elastic.Open()")

	// Open a connection with ElasticSearch
	cfg := elasticsearch.Config{
		Addresses: []string{
			c.cfg.Host,
		},
	}

	var err error
	c.es, err = elasticsearch.NewClient(cfg)
	if err != nil {
		return err
	}

	// create a buffer channel
	c.ch = make(chan opencdc.Data, c.cfg.BatchSize)
	c.shutdown = make(chan struct{})

	for _, index := range c.cfg.Indexes {
		c.wg.Add(1)

		offset := 0
		for _, position := range positions {
			if index == position.Index {
				offset = position.Pos
			}
		}

		c.offsets[index] = offset
		NewWorker(c, index, offset)
	}

	c.positions = positions

	return nil
}