Decompress for types

policy/aggregation_type.go

@@ -155,9 +155,19 @@ type AggregationTypeCompressor interface {
 	Compress(aggTypes AggregationTypes) (CompressedAggregationTypes, error)
 }
 
-// AggregationTypeDecompressor can decompress AggregationTypes.
+// AggregationTypeDecompressor can decompress CompressedAggregationTypes for different types of metrics.
 type AggregationTypeDecompressor interface {
-	Decompress(code CompressedAggregationTypes) (AggregationTypes, error)
+	// Decompress decompresses aggregation types, return error if any invalid aggregation type is found.
+	Decompress(compressed CompressedAggregationTypes) (AggregationTypes, error)
+	// DecompressForCounter decompresses aggregation types,
+	// returns error if any invalid aggregation type for Counter is found.
+	DecompressForCounter(compressed CompressedAggregationTypes) (AggregationTypes, error)
+	// DecompressForTimer decompresses aggregation types,
+	// returns error if any invalid aggregation type for Timer is found.
+	DecompressForTimer(compressed CompressedAggregationTypes) (AggregationTypes, error)
+	// DecompressForTimer decompresses aggregation types,
+	// returns error if any invalid aggregation type for Gauge is found.
+	DecompressGauge(compressed CompressedAggregationTypes) (AggregationTypes, error)
 }
 
 type aggregationCompresser struct {
@@ -197,6 +207,45 @@ func NewAggregationTypeDecompressor() AggregationTypeDecompressor {
 }
 
 func (c *aggregationDecompresser) Decompress(compressed CompressedAggregationTypes) (AggregationTypes, error) {
+	return c.decompress(compressed, validateForAll, errGen)
+}
+
+func (c *aggregationDecompresser) DecompressGauge(compressed CompressedAggregationTypes) (AggregationTypes, error) {
+	return c.decompress(compressed, validateForGauge, gaugeErrGen)
+}
+
+func (c *aggregationDecompresser) DecompressForCounter(compressed CompressedAggregationTypes) (AggregationTypes, error) {
+	return c.decompress(compressed, validateForGauge, counterErrGen)
+}
+
+func (c *aggregationDecompresser) DecompressForTimer(compressed CompressedAggregationTypes) (AggregationTypes, error) {
+	return c.decompress(compressed, validateForTimer, timerErrGen)
+}
+
+type validateFn func(aggType AggregationType) bool
+type errorGen func(aggType AggregationType) error
+
+var (
+	validateForGauge   = validateFn(func(aggType AggregationType) bool { return aggType.IsValidForGauge() })
+	validateForCounter = validateFn(func(aggType AggregationType) bool { return aggType.IsValidForCounter() })
+	validateForTimer   = validateFn(func(aggType AggregationType) bool { return aggType.IsValidForTimer() })
+	validateForAll     = validateFn(func(aggType AggregationType) bool { return aggType.IsValid() })
+
+	gaugeErrGen = func(aggType AggregationType) error {
+		return fmt.Errorf("invalid AggregationType %s for Gauge", aggType.String())
+	}
+	counterErrGen = func(aggType AggregationType) error {
+		return fmt.Errorf("invalid AggregationType %s for Counter", aggType.String())
+	}
+	timerErrGen = func(aggType AggregationType) error {
+		return fmt.Errorf("invalid AggregationType %s for Timer", aggType.String())
+	}
+	errGen = func(aggType AggregationType) error {
+		return fmt.Errorf("invalid AggregationType %s", aggType.String())
+	}
+)
+
+func (c *aggregationDecompresser) decompress(compressed CompressedAggregationTypes, validate validateFn, err errorGen) (AggregationTypes, error) {
 	codes := make([]uint64, CompressedSize)
 	for i := range compressed {
 		codes[i] = uint64(compressed[i])
@@ -206,8 +255,8 @@ func (c *aggregationDecompresser) Decompress(compressed CompressedAggregationTyp
 	c.res = c.res[:0]
 	for i, e := bs.NextSet(0); e; i, e = bs.NextSet(i + 1) {
 		aggType := AggregationType(i)
-		if !aggType.IsValid() {
-			return EmptyAggregationTypes, fmt.Errorf("could not decompress invalid AggregationType %v", i)
+		if !validate(aggType) {
+			return EmptyAggregationTypes, err(aggType)
 		}
 		c.res = append(c.res, aggType)
 	}