func (c *MaxProcessor) FinalizeBucket(bucket dagger.Timestamp) *dagger.Record {
	log.Println("[Max] finalizing", bucket)
	max := math.Inf(-1)
	for _, v := range c.state.Values[bucket] {
		if v > max {
			max = v
		}
	}
	new := &dagger.Record{
		Data:      max,
		Timestamp: bucket,
		ID:        uuid.NewV4().String(),
	}
	delete(c.state.Values, bucket)
	log.Println("[Max] finalization finished", bucket)
	return new
}
FinalizeBucket produces a new record from the bucket and deletes it
package main
import (
	"encoding/json"
	"fmt"
	log "github.com/Sirupsen/logrus"
	"math"
	"os"
	"os/signal"
	"strconv"
	"github.com/nsaje/dagger/computations"
	"github.com/nsaje/dagger/dagger"
	"github.com/twinj/uuid"
)
type MaxProcessor struct {
	state MaxCompState
}
type MaxCompState struct {
	Values map[dagger.Timestamp][]float64
}
type MaxCompStateJSON struct {
	Buckets []dagger.Timestamp
	Values  [][]float64
}
func (c *MaxProcessor) GetState() ([]byte, error) {
	stateJSON := NewMaxProcessorStateJSON()
	for bucket := range c.state.Values {
		stateJSON.Buckets = append(stateJSON.Buckets, bucket)
		stateJSON.Values = append(stateJSON.Values, c.state.Values[bucket])
	}
	log.Println("returning stateJSON:", stateJSON)
	return json.Marshal(stateJSON)
}
func (c *MaxProcessor) SetState(state []byte) error {
	newState := NewMaxProcessorState()
	newStateJSON := NewMaxProcessorStateJSON()
	err := json.Unmarshal(state, &newStateJSON)
	if err != nil {
		return fmt.Errorf("Error setting plugin state: %s", err)
	}
	for i, bucket := range newStateJSON.Buckets {
		newState.Values[bucket] = newStateJSON.Values[i]
	}
	c.state = newState
	log.Println("[Max] new state:", c.state)
	return nil
}
func (c *MaxProcessor) ProcessBucket(bucket dagger.Timestamp, t *dagger.Record) error {
	log.Println("[Max] processing", t)
	value, _ := strconv.ParseFloat(t.Data.(string), 64)
	c.state.Values[bucket] = append(c.state.Values[bucket], value)
	return nil
}