amocrm/internal/workers/data_updater/data_updater.go

package data_updater

import (
	"amocrm/internal/brokers"
	"amocrm/internal/models"
	"amocrm/pkg/timer"
	"context"
	"go.uber.org/zap"
	"time"
	"fmt"
)

type Deps struct {
	Logger   *zap.Logger
	Producer *brokers.Producer
}

type DataUpdater struct {
	logger   *zap.Logger
	producer *brokers.Producer
}

func NewDataUpdaterWC(deps Deps) *DataUpdater {
	return &DataUpdater{
		logger:   deps.Logger,
		producer: deps.Producer,
	}
}

func (wc *DataUpdater) Start(ctx context.Context) {
	nextStart := timer.CalculateTime(4)
	ticker := time.NewTicker(time.Nanosecond * time.Duration(nextStart))
	//ticker := time.NewTicker(10 * time.Second)
	defer ticker.Stop()

	for {
		func () { 
			defer func() {
				if v := recover();v!=nil{
					fmt.Println("RECOVERING in DataUpdater",v)
				}
			} ()
			select {
			case <-ticker.C:
				wc.processTasks(ctx)
				ticker.Reset(time.Nanosecond * time.Duration(nextStart))
			case <-ctx.Done():
				return
			}
		}()
	}
}

func (wc *DataUpdater) processTasks(ctx context.Context) {
	err := wc.producer.ToKafkaUpdate(ctx, models.KafkaMessage{
		Type:      models.AllDataUpdate,
		AccountID: "",
	})
	if err != nil {
		wc.logger.Error("error send task in kafka ro update all data", zap.Error(err))
		return
	}
}

func (wc *DataUpdater) Stop(_ context.Context) error {
	return nil
}