func (e *GenericElem) AddUnique(timestamp time.Time, values []float64, sourceID uint32) error {
	alignedStart := timestamp.Truncate(e.sp.Resolution().Window).UnixNano()
	lockedAgg, err := e.findOrCreate(alignedStart, createAggregationOptions{initSourceSet: true})
	if err != nil {
		return err
	}
	lockedAgg.Lock()
	if lockedAgg.closed {
		lockedAgg.Unlock()
		return errAggregationClosed
	}
	source := uint(sourceID)
	if lockedAgg.sourcesSeen.Test(source) {
		lockedAgg.Unlock()
		return errDuplicateForwardingSource
	}
	lockedAgg.sourcesSeen.Set(source)
	for _, v := range values {
		lockedAgg.aggregation.Add(v)
	}
	lockedAgg.Unlock()
	return nil
}
AddUnique adds a metric value from a given source at a given timestamp.
If previous values from the same source have already been added to the
same aggregation, the incoming value is discarded.
package aggregator
import (
	"fmt"
	"math"
	"sync"
	"time"
	raggregation "github.com/m3db/m3aggregator/aggregation"
	maggregation "github.com/m3db/m3metrics/aggregation"
	"github.com/m3db/m3metrics/metric"
	"github.com/m3db/m3metrics/metric/id"
	"github.com/m3db/m3metrics/metric/unaggregated"
	"github.com/m3db/m3metrics/pipeline/applied"
	"github.com/m3db/m3metrics/policy"
	"github.com/m3db/m3metrics/transformation"
	"github.com/mauricelam/genny/generic"
	"github.com/willf/bitset"
)
type typeSpecificAggregation interface {
	generic.Type
	Add(value float64)
	AddUnion(mu unaggregated.MetricUnion)
	ValueOf(aggType maggregation.Type) float64
	Close()
}
type genericElemPool interface {
	generic.Type
	Put(value *GenericElem)
}
type typeSpecificElemBase interface {
	generic.Type
	Type() metric.Type
	FullPrefix(opts Options) []byte
	DefaultAggregationTypes(aggTypesOpts maggregation.TypesOptions) maggregation.Types
	TypeStringFor(aggTypesOpts maggregation.TypesOptions, aggType maggregation.Type) []byte
	ElemPool(opts Options) genericElemPool
	NewAggregation(opts Options, aggOpts raggregation.Options) typeSpecificAggregation
	ResetSetData(
		aggTypesOpts maggregation.TypesOptions,
		aggTypes maggregation.Types,
		useDefaultAggregation bool,
	) error
	Close()
}
type lockedAggregation struct {
	sync.Mutex
	closed      bool
	sourcesSeen *bitset.BitSet
	aggregation typeSpecificAggregation
}
type timedAggregation struct {
	startAtNanos int64  
	lockedAgg    *lockedAggregation
}
func (ta *timedAggregation) Reset() {
	ta.startAtNanos = 0
	ta.lockedAgg = nil
}
type GenericElem struct {
	elemBase
	typeSpecificElemBase
	values              []timedAggregation  
	toConsume           []timedAggregation  
	lastConsumedAtNanos int64               
	lastConsumedValues  []float64           
}
func NewGenericElem(
	id id.RawID,
	sp policy.StoragePolicy,
	aggTypes maggregation.Types,
	pipeline applied.Pipeline,
	numForwardedTimes int,
	idPrefixSuffixType IDPrefixSuffixType,
	opts Options,
) (*GenericElem, error) {
	e := &GenericElem{
		elemBase: newElemBase(opts),
		values:   make([]timedAggregation, 0, defaultNumAggregations),  
	}
	if err := e.ResetSetData(id, sp, aggTypes, pipeline, numForwardedTimes, idPrefixSuffixType); err != nil {
		return nil, err
	}
	return e, nil
}
func MustNewGenericElem(
	id id.RawID,
	sp policy.StoragePolicy,
	aggTypes maggregation.Types,
	pipeline applied.Pipeline,
	numForwardedTimes int,
	idPrefixSuffixType IDPrefixSuffixType,
	opts Options,
) *GenericElem {
	elem, err := NewGenericElem(id, sp, aggTypes, pipeline, numForwardedTimes, idPrefixSuffixType, opts)
	if err != nil {
		panic(fmt.Errorf("unable to create element: %v", err))
	}
	return elem
}
func (e *GenericElem) ResetSetData(
	id id.RawID,
	sp policy.StoragePolicy,
	aggTypes maggregation.Types,
	pipeline applied.Pipeline,
	numForwardedTimes int,
	idPrefixSuffixType IDPrefixSuffixType,
) error {
	useDefaultAggregation := aggTypes.IsDefault()
	if useDefaultAggregation {
		aggTypes = e.DefaultAggregationTypes(e.aggTypesOpts)
	}
	if err := e.elemBase.resetSetData(id, sp, aggTypes, useDefaultAggregation, pipeline, numForwardedTimes, idPrefixSuffixType); err != nil {
		return err
	}
	if err := e.typeSpecificElemBase.ResetSetData(e.aggTypesOpts, aggTypes, useDefaultAggregation); err != nil {
		return err
	}
	if !e.parsedPipeline.HasDerivativeTransform {
		return nil
	}
	numAggTypes := len(e.aggTypes)
	if cap(e.lastConsumedValues) < numAggTypes {
		e.lastConsumedValues = make([]float64, numAggTypes)
	}
	e.lastConsumedValues = e.lastConsumedValues[:numAggTypes]
	for i := 0; i < len(e.lastConsumedValues); i++ {
		e.lastConsumedValues[i] = nan
	}
	return nil
}
func (e *GenericElem) AddUnion(timestamp time.Time, mu unaggregated.MetricUnion) error {
	alignedStart := timestamp.Truncate(e.sp.Resolution().Window).UnixNano()
	lockedAgg, err := e.findOrCreate(alignedStart, createAggregationOptions{})
	if err != nil {
		return err
	}
	lockedAgg.Lock()
	if lockedAgg.closed {
		lockedAgg.Unlock()
		return errAggregationClosed
	}
	lockedAgg.aggregation.AddUnion(mu)
	lockedAgg.Unlock()
	return nil
}
func (e *GenericElem) AddValue(timestamp time.Time, value float64) error {
	alignedStart := timestamp.Truncate(e.sp.Resolution().Window).UnixNano()
	lockedAgg, err := e.findOrCreate(alignedStart, createAggregationOptions{})
	if err != nil {
		return err
	}
	lockedAgg.Lock()
	if lockedAgg.closed {
		lockedAgg.Unlock()
		return errAggregationClosed
	}
	lockedAgg.aggregation.Add(value)
	lockedAgg.Unlock()
	return nil
}