<commit_before>package connector

import (
	log "github.com/Sirupsen/logrus"
	"sync"
)

// Queue is an interface modeling a task-queue (it is started and more Requests can be pushed to it, an finally it is stopped).
type Queue interface {
	ResponseHandleSetter

	Start() error
	Push(request Request) error
	Stop() error
}

type queue struct {
	sender          Sender
	responseHandler ResponseHandler
	requestsC       chan Request
	nWorkers        int
	wg              sync.WaitGroup
}

// NewQueue returns a new Queue (not started).
func NewQueue(sender Sender, nWorkers int) Queue {
	q := &queue{
		sender:    sender,
		requestsC: make(chan Request),
		nWorkers:  nWorkers,
	}
	return q
}

func (q *queue) SetResponseHandler(rh ResponseHandler) {
	q.responseHandler = rh
}

func (q *queue) ResponseHandler() ResponseHandler {
	return q.responseHandler
}

func (q *queue) Start() error {
	if q.responseHandler == nil {
		logger.Warning("No handler set for connector queue")
	}
	for i := 1; i <= q.nWorkers; i++ {
		go q.worker()
	}
	return nil
}

func (q *queue) worker() {
	for request := range q.requestsC {
		q.wg.Add(1)
		response, err := q.sender.Send(request)
		if q.responseHandler != nil {
			err = q.responseHandler.HandleResponse(request, response, err)
			if err != nil {
				logger.WithFields(log.Fields{
					"error":      err.Error(),
					"subscriber": request.Subscriber(),
					"message":    request.Message(),
				}).Error("Error handling connector response")
			}
		}
		q.wg.Done()
	}
}

func (q *queue) Push(request Request) error {<commit_msg>apns queue worker logging<commit_after>package connector

import (
	log "github.com/Sirupsen/logrus"
	"sync"
)

// Queue is an interface modeling a task-queue (it is started and more Requests can be pushed to it, an finally it is stopped).
type Queue interface {
	ResponseHandleSetter

	Start() error
	Push(request Request) error
	Stop() error
}

type queue struct {
	sender          Sender
	responseHandler ResponseHandler
	requestsC       chan Request
	nWorkers        int
	wg              sync.WaitGroup
}

// NewQueue returns a new Queue (not started).
func NewQueue(sender Sender, nWorkers int) Queue {
	q := &queue{
		sender:    sender,
		requestsC: make(chan Request),
		nWorkers:  nWorkers,
	}
	return q
}

func (q *queue) SetResponseHandler(rh ResponseHandler) {
	q.responseHandler = rh
}

func (q *queue) ResponseHandler() ResponseHandler {
	return q.responseHandler
}

func (q *queue) Start() error {
	if q.responseHandler == nil {
		logger.Warning("No handler set for connector queue")
	}
	for i := 1; i <= q.nWorkers; i++ {
		go q.worker(i)
	}
	return nil
}

func (q *queue) worker(i int) {
	logger.WithField("worker", i).Debug("starting queue worker")
	for request := range q.requestsC {
		q.wg.Add(1)
		response, err := q.sender.Send(request)
		if q.responseHandler != nil {
			err = q.responseHandler.HandleResponse(request, response, err)
			if err != nil {
				logger.WithFields(log.Fields{
					"error":      err.Error(),
					"subscriber": request.Subscriber(),
					"message":    request.Message(),
				}).Error("Error handling connector response")
			}
		}
		q.wg.Done()
	}
}

func (q *queue) Push(request Request) error {
{'fraction_non_alphanumeric': 0.11689526184538654, 'fraction_numerical': 0.0012468827930174563, 'mean_word_length': 3.383879781420765, 'pattern_counts': {'":': 6, '<': 5, '<?xml version=': 0, '>': 3, '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': '4791472', 'n_tokens_mistral': 1130, 'n_tokens_neox': 1036, 'n_words': 320}