func (w *Worker) UpdateLastSave() {
	w.LastSave = time.Now()
}
Update last save time
package main
import (
	"fmt"
	"github.com/wunderlist/hamustro/src/dialects"
	"log"
	"sync"
	"time"
)
type Worker struct {
	ID             int
	WorkerPool     chan *Worker
	JobChannel     chan Job
	BufferSize     int
	BufferedEvents []*dialects.Event
	Penalty        float32
	RetryAttempt   int
	LastSave       time.Time
	quit           chan *sync.WaitGroup
}
type WorkerOptions struct {
	BufferSize   int
	RetryAttempt int
	SpreadBuffer bool
}
func NewWorker(id int, options *WorkerOptions, workerPool chan *Worker) *Worker {
	return &Worker{
		ID:             id,
		WorkerPool:     workerPool,
		JobChannel:     make(chan Job),
		BufferSize:     options.BufferSize,
		BufferedEvents: []*dialects.Event{},
		Penalty:        1.0,
		RetryAttempt:   options.RetryAttempt,
		LastSave:       time.Now(),
		quit:           make(chan *sync.WaitGroup)}
}
func (w *Worker) Start() {
	if storageClient.IsBufferedStorage() {
		log.Printf("(%d worker) Started with %d buffer", w.ID, w.BufferSize)
	} else {
		log.Printf("(%d worker) Started", w.ID)
	}
	go func() {
		for {
			w.WorkerPool <- w
			select {
			case action := <-w.JobChannel:
				switch action.GetAction() {
				case ACTION_EVENT:
					if verbose {
						log.Printf("(%d worker) Received an add new event request!\n", w.ID)
					}
					if err := w.Work(action.(*EventAction)); err != nil {
						log.Print(err)
					}
				case ACTION_FLUSH:
					if verbose {
						log.Printf("(%d worker) Received a flush request!\n", w.ID)
					}
					if err := w.Flush(); err != nil {
						log.Print(err)
					}
				}
			case wg := <-w.quit:
				defer wg.Done()
				if err := w.Rescue(); err != nil {
					log.Print(err)
				}
				return
			}
		}
	}()
}
func (w *Worker) Work(action *EventAction) error {
	if !storageClient.IsBufferedStorage() {
		if err := w.Save(action); err != nil {
			return err
		}
	} else {
		w.AddEventToBuffer(action.GetEvent())
		if !w.IsBufferFull() {
			return nil
		}
		if verbose {
			log.Printf("(%d worker) Saving buffered messages started", w.ID)
		}
		if err := w.SaveBatch(); err != nil {
			return err
		}
		if verbose {
			log.Printf("(%d worker) Saving buffered messages was finished", w.ID)
		}
	}
	return nil
}
func (w *Worker) SaveBatch() error {
	msg, err := storageClient.GetBatchConverter()(w.BufferedEvents)
	if err != nil {
		w.IncreasePenalty()
		return fmt.Errorf("(%d worker) Batch converting buffered messages is failed with %d records: %s", w.ID, len(w.BufferedEvents), err.Error())
	}
	if err := storageClient.Save(msg); err != nil {
		w.IncreasePenalty()
		return fmt.Errorf("(%d worker) Saving buffered messages is failed with %d records: %s", w.ID, len(w.BufferedEvents), err.Error())
	}
	w.ResetBuffer()
	w.UpdateLastSave()
	return nil
}
func (w *Worker) Save(action *EventAction) error {
	msg, err := storageClient.GetConverter()(action.GetEvent())
	if err != nil {
		rerr := fmt.Errorf("(%d worker) Encoding message is failed (%d attempt): %s", w.ID, action.Attempt, err.Error())
		action.MarkAsFailed(w.RetryAttempt)
		return rerr
	}
	if err := storageClient.Save(msg); err != nil {
		rerr := fmt.Errorf("(%d worker) Saving message is failed (%d attempt): %s", w.ID, action.Attempt, err.Error())
		action.MarkAsFailed(w.RetryAttempt)
		return rerr
	}
	w.UpdateLastSave()
	return nil
}
func (w *Worker) Rescue() error {
	log.Printf("(%d worker) Received a signal to stop", w.ID)
	if err := w.Flush(); err != nil {
		return err
	}
	log.Printf("(%d worker) Stopped successfully", w.ID)
	return nil
}
func (w *Worker) Flush() error {
	if storageClient.IsBufferedStorage() && len(w.BufferedEvents) != 0 {
		log.Printf("(%d worker) Flushing %d buffered messages", w.ID, len(w.BufferedEvents))
		if err := w.SaveBatch(); err != nil {
			return err
		}
	} else if len(w.BufferedEvents) == 0 {
		w.UpdateLastSave()
	}
	return nil
}
func (w *Worker) Stop(wg *sync.WaitGroup) {
	go func() {
		if storageClient.IsBufferedStorage() {
			log.Printf("(%d worker) Sending stop signal to worker with %d buffered events", w.ID, len(w.BufferedEvents))
		} else {
			log.Printf("(%d worker) Sending stop signal to worker", w.ID)
		}
		w.quit <- wg
	}()
}
func (w *Worker) IncreasePenalty() {
	w.Penalty *= 1.5
}
func (w *Worker) GetBufferSize() int {
	return int(float32(w.BufferSize) * w.Penalty)
}
func (w *Worker) IsBufferFull() bool {
	return len(w.BufferedEvents) >= w.GetBufferSize()
}
func (w *Worker) ResetBuffer() {
	w.BufferedEvents = w.BufferedEvents[:0]
	w.Penalty = 1.0
}
func (w *Worker) AddEventToBuffer(event *dialects.Event) {
	w.BufferedEvents = append(w.BufferedEvents, event)
}