<commit_before>package main

import (
	"errors"
	"github.com/Comcast/webpa-common/logging"
	"github.com/Comcast/webpa-common/webhook"
	"github.com/Comcast/webpa-common/wrp"
	"net/http"
	"net/url"
	"strings"
	"sync"
	"time"
)

// SenderWrapperFactory configures the CaduceusSenderWrapper for creation
type SenderWrapperFactory struct {
	// The number of workers to assign to each OutboundSender created.
	NumWorkersPerSender int

	// The queue size to assign to each OutboundSender created.
	QueueSizePerSender int

	// The cut off time to assign to each OutboundSender created.
	CutOffPeriod time.Duration

	// The amount of time to let expired OutboundSenders linger before
	// shutting them down and cleaning up the resources associated with them.
	Linger time.Duration

	// The factory that we'll use to make new ServerProfilers on a per
	// outboundSender basis
	ProfilerFactory ServerProfilerFactory

	// The logger implementation to share with OutboundSenders.
	Logger logging.Logger

	// The http client to share with OutboundSenders.
	Client *http.Client
}

type SenderWrapper interface {
	Update([]webhook.W)
	Queue(CaduceusRequest)
	Shutdown(bool)
}

// CaduceusSenderWrapper contains no external parameters.
type CaduceusSenderWrapper struct {
	client              *http.Client
	numWorkersPerSender int
	queueSizePerSender  int
	cutOffPeriod        time.Duration
	linger              time.Duration
	logger              logging.Logger
	mutex               sync.RWMutex
	senders             map[string]OutboundSender
	profilerFactory     ServerProfilerFactory
	wg                  sync.WaitGroup
	shutdown            chan struct{}
}

// New produces a new SenderWrapper implemented by CaduceusSenderWrapper
// based on the factory configuration.
func (swf SenderWrapperFactory) New() (sw SenderWrapper, err error) {
	caduceusSenderWrapper := &CaduceusSenderWrapper{
		client:              swf.Client,
		numWorkersPerSender: swf.NumWorkersPerSender,
		queueSizePerSender:  swf.QueueSizePerSender,
		cutOffPeriod:        swf.CutOffPeriod,
		linger:              swf.Linger,
		logger:              swf.Logger,
		profilerFactory:     swf.ProfilerFactory,
	}

	if swf.Linger <= 0 {
		err = errors.New("Linger must be positive.")
		sw = nil
		return
	}

	caduceusSenderWrapper.senders = make(map[string]OutboundSender)
	caduceusSenderWrapper.shutdown = make(chan struct{})

	caduceusSenderWrapper.wg.Add(1)
	go undertaker(caduceusSenderWrapper)

	sw = caduceusSenderWrapper
	return
}

// Update is called when we get changes to our webhook listeners with either
// additions, or updates.  This code takes care of building new OutboundSenders
// and maintaining the existing OutboundSenders.
func (sw *CaduceusSenderWrapper) Update(list []webhook.W) {
	// We'll like need this, so let's get one ready
	osf := OutboundSenderFactory{
		Client:          sw.client,
		CutOffPeriod:    sw.cutOffPeriod,
		NumWorkers:      sw.numWorkersPerSender,
		QueueSize:       sw.queueSizePerSender,
		ProfilerFactory: sw.profilerFactory,
		Logger:          sw.logger,
	}

	ids := make([]struct {
		Listener webhook.W
		ID       string
	}, len(list))

	for i, v := range list {
		v.DurationValidator()
		v.Until = time.Now().Add(v.Duration)

		ids[i].Listener = v
		ids[i].ID = v.ID()
	}

	sw.mutex.Lock()
	for _, inValue := range ids {
		sender, ok := sw.senders[inValue.ID]
		if true == ok {
			sender.Extend(inValue.Listener.Until)
		} else {
			osf.Listener = inValue.Listener
			obs, err := osf.New()
			if nil == err {
				sw.senders[inValue.ID] = obs
			}
		}
	}
	sw.mutex.Unlock()
}

// Queue is used to send all the possible outbound senders a request.  This
// function performs the fan-out and filtering to multiple possible endpoints.
func (sw *CaduceusSenderWrapper) Queue(req CaduceusRequest) {
	switch req.ContentType {

	case "application/json":
		if url, err := url.Parse(req.TargetURL); nil == err {
			elements := strings.Split(url.Path, "/")
			if 6 < len(elements) &&
				"" == elements[0] &&
				"api" == elements[1] &&<commit_msg>removing durationValidator and until set.<commit_after>package main

import (
	"errors"
	"github.com/Comcast/webpa-common/logging"
	"github.com/Comcast/webpa-common/webhook"
	"github.com/Comcast/webpa-common/wrp"
	"net/http"
	"net/url"
	"strings"
	"sync"
	"time"
)

// SenderWrapperFactory configures the CaduceusSenderWrapper for creation
type SenderWrapperFactory struct {
	// The number of workers to assign to each OutboundSender created.
	NumWorkersPerSender int

	// The queue size to assign to each OutboundSender created.
	QueueSizePerSender int

	// The cut off time to assign to each OutboundSender created.
	CutOffPeriod time.Duration

	// The amount of time to let expired OutboundSenders linger before
	// shutting them down and cleaning up the resources associated with them.
	Linger time.Duration

	// The factory that we'll use to make new ServerProfilers on a per
	// outboundSender basis
	ProfilerFactory ServerProfilerFactory

	// The logger implementation to share with OutboundSenders.
	Logger logging.Logger

	// The http client to share with OutboundSenders.
	Client *http.Client
}

type SenderWrapper interface {
	Update([]webhook.W)
	Queue(CaduceusRequest)
	Shutdown(bool)
}

// CaduceusSenderWrapper contains no external parameters.
type CaduceusSenderWrapper struct {
	client              *http.Client
	numWorkersPerSender int
	queueSizePerSender  int
	cutOffPeriod        time.Duration
	linger              time.Duration
	logger              logging.Logger
	mutex               sync.RWMutex
	senders             map[string]OutboundSender
	profilerFactory     ServerProfilerFactory
	wg                  sync.WaitGroup
	shutdown            chan struct{}
}

// New produces a new SenderWrapper implemented by CaduceusSenderWrapper
// based on the factory configuration.
func (swf SenderWrapperFactory) New() (sw SenderWrapper, err error) {
	caduceusSenderWrapper := &CaduceusSenderWrapper{
		client:              swf.Client,
		numWorkersPerSender: swf.NumWorkersPerSender,
		queueSizePerSender:  swf.QueueSizePerSender,
		cutOffPeriod:        swf.CutOffPeriod,
		linger:              swf.Linger,
		logger:              swf.Logger,
		profilerFactory:     swf.ProfilerFactory,
	}

	if swf.Linger <= 0 {
		err = errors.New("Linger must be positive.")
		sw = nil
		return
	}

	caduceusSenderWrapper.senders = make(map[string]OutboundSender)
	caduceusSenderWrapper.shutdown = make(chan struct{})

	caduceusSenderWrapper.wg.Add(1)
	go undertaker(caduceusSenderWrapper)

	sw = caduceusSenderWrapper
	return
}

// Update is called when we get changes to our webhook listeners with either
// additions, or updates.  This code takes care of building new OutboundSenders
// and maintaining the existing OutboundSenders.
func (sw *CaduceusSenderWrapper) Update(list []webhook.W) {
	// We'll like need this, so let's get one ready
	osf := OutboundSenderFactory{
		Client:          sw.client,
		CutOffPeriod:    sw.cutOffPeriod,
		NumWorkers:      sw.numWorkersPerSender,
		QueueSize:       sw.queueSizePerSender,
		ProfilerFactory: sw.profilerFactory,
		Logger:          sw.logger,
	}

	ids := make([]struct {
		Listener webhook.W
		ID       string
	}, len(list))

	for i, v := range list {
		ids[i].Listener = v
		ids[i].ID = v.ID()
	}

	sw.mutex.Lock()
	for _, inValue := range ids {
		sender, ok := sw.senders[inValue.ID]
		if true == ok {
			sender.Extend(inValue.Listener.Until)
		} else {
			osf.Listener = inValue.Listener
			obs, err := osf.New()
			if nil == err {
				sw.senders[inValue.ID] = obs
			}
		}
	}
	sw.mutex.Unlock()
}

// Queue is used to send all the possible outbound senders a request.  This
// function performs the fan-out and filtering to multiple possible endpoints.
func (sw *CaduceusSenderWrapper) Queue(req CaduceusRequest) {
	switch req.ContentType {

	case "application/json":
		if url, err := url.Parse(req.TargetURL); nil == err {
			elements := strings.Split(url.Path, "/")
			if 6 < len(elements) &&
				"" == elements[0] &&
				"api" == elements[1] &&
{'fraction_non_alphanumeric': 0.0904228855721393, 'fraction_numerical': 0.0012437810945273632, 'mean_word_length': 3.327771797631862, 'pattern_counts': {'":': 2, '<': 7, '<?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': '641032', 'n_tokens_mistral': 2655, 'n_tokens_neox': 2503, 'n_words': 711}