<commit_before>package processor

import (
	"errors"
	"fmt"
	"log"
	"sync"
	"sync/atomic"
	"time"

	"github.com/bsm/redis-lock"

	"github.com/go-msgqueue/msgqueue"
	"github.com/go-msgqueue/msgqueue/internal"
)

const consumerBackoff = time.Second
const maxBackoff = 12 * time.Hour
const stopTimeout = 30 * time.Second

type Delayer interface {
	Delay() time.Duration
}

type Stats struct {
	InFlight    uint32
	Deleting    uint32
	Processed   uint32
	Retries     uint32
	Fails       uint32
	AvgDuration time.Duration
}

// Processor reserves messages from the queue, processes them,
// and then either releases or deletes messages from the queue.
type Processor struct {
	q   Queuer
	opt *msgqueue.Options

	handler         msgqueue.Handler
	fallbackHandler msgqueue.Handler

	wg sync.WaitGroup
	ch chan *msgqueue.Message

	workersWG   sync.WaitGroup
	workerLocks []*lock.Lock

	workersStarted uint32
	workersStop    chan struct{}

	messageFetcherStarted uint32
	messageFetcherStop    uint32

	delBatch *internal.Batcher

	errCount   uint32
	delayCount uint32
	delaySec   uint32

	inFlight    uint32
	deleting    uint32
	processed   uint32
	fails       uint32
	retries     uint32
	avgDuration uint32
}

// New creates new Processor for the queue using provided processing options.
func New(q Queuer, opt *msgqueue.Options) *Processor {
	opt.Init()
	p := &Processor{
		q:   q,
		opt: opt,

		ch: make(chan *msgqueue.Message, opt.BufferSize),
	}

	if opt.WorkerLimit > 0 {
		p.workerLocks = make([]*lock.Lock, opt.WorkerLimit)
		for i := 0; i < opt.WorkerLimit; i++ {
			key := fmt.Sprintf("%s:worker-lock:%d", p.q.Name(), i)
			p.workerLocks[i] = lock.NewLock(opt.Redis, key, &lock.LockOptions{
				LockTimeout: opt.ReservationTimeout,
			})
		}
	}

	p.setHandler(opt.Handler)
	if opt.FallbackHandler != nil {
		p.setFallbackHandler(opt.FallbackHandler)
	}

	p.delBatch = internal.NewBatcher(p.opt.WorkerNumber, p.deleteBatch)

	return p
}

// Starts creates new Processor and starts it.
func Start(q Queuer, opt *msgqueue.Options) *Processor {<commit_msg>processor: prefetch n-1 messages<commit_after>package processor

import (
	"errors"
	"fmt"
	"log"
	"sync"
	"sync/atomic"
	"time"

	"github.com/bsm/redis-lock"

	"github.com/go-msgqueue/msgqueue"
	"github.com/go-msgqueue/msgqueue/internal"
)

const consumerBackoff = time.Second
const maxBackoff = 12 * time.Hour
const stopTimeout = 30 * time.Second

type Delayer interface {
	Delay() time.Duration
}

type Stats struct {
	InFlight    uint32
	Deleting    uint32
	Processed   uint32
	Retries     uint32
	Fails       uint32
	AvgDuration time.Duration
}

// Processor reserves messages from the queue, processes them,
// and then either releases or deletes messages from the queue.
type Processor struct {
	q   Queuer
	opt *msgqueue.Options

	handler         msgqueue.Handler
	fallbackHandler msgqueue.Handler

	wg sync.WaitGroup
	ch chan *msgqueue.Message

	workersWG   sync.WaitGroup
	workerLocks []*lock.Lock

	workersStarted uint32
	workersStop    chan struct{}

	messageFetcherStarted uint32
	messageFetcherStop    uint32

	delBatch *internal.Batcher

	errCount   uint32
	delayCount uint32
	delaySec   uint32

	inFlight    uint32
	deleting    uint32
	processed   uint32
	fails       uint32
	retries     uint32
	avgDuration uint32
}

// New creates new Processor for the queue using provided processing options.
func New(q Queuer, opt *msgqueue.Options) *Processor {
	opt.Init()
	p := &Processor{
		q:   q,
		opt: opt,

		ch: make(chan *msgqueue.Message, opt.BufferSize-1),
	}

	if opt.WorkerLimit > 0 {
		p.workerLocks = make([]*lock.Lock, opt.WorkerLimit)
		for i := 0; i < opt.WorkerLimit; i++ {
			key := fmt.Sprintf("%s:worker-lock:%d", p.q.Name(), i)
			p.workerLocks[i] = lock.NewLock(opt.Redis, key, &lock.LockOptions{
				LockTimeout: opt.ReservationTimeout,
			})
		}
	}

	p.setHandler(opt.Handler)
	if opt.FallbackHandler != nil {
		p.setFallbackHandler(opt.FallbackHandler)
	}

	p.delBatch = internal.NewBatcher(p.opt.WorkerNumber, p.deleteBatch)

	return p
}

// Starts creates new Processor and starts it.
func Start(q Queuer, opt *msgqueue.Options) *Processor {
{'fraction_non_alphanumeric': 0.09755507141128056, 'fraction_numerical': 0.019849915274751877, 'mean_word_length': 3.632286995515695, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6307123', 'n_tokens_mistral': 1540, 'n_tokens_neox': 1425, 'n_words': 369}