func (ag *Aggregator) Run(ctx context.Context, results <-chan *AnalyzerResult) error {
	var sentiments = make(map[string]*Sentiment)
	for res := range results {
		topic := res.SearchTerm.Topic
		if sentiments[topic] == nil {
			sentiments[topic] = &Sentiment{}
		}
		sentiments[topic] = ag.updateAggregate(
			res.Score,
			res.Magnitude,
			res.TweetID,
			sentiments[topic],
		)
		sentiments[topic].populateWithSearch(res.SearchTerm)
	}
	for topic, sentiment := range sentiments {
		sentiment.finalize()
		id, err := ag.db.SaveSentiment(ctx, *sentiment)
		if err != nil {
			ag.logger.Log(
				"err", errors.Wrap(err, "failed to save topic"),
				"topic", topic,
			)
			continue
		}
		ag.logger.Log(
			"state", "saved",
			"topic", sentiment.Topic,
			"slug", sentiment.Slug,
			"id", id,
			"score", sentiment.Score,
			"count", sentiment.Count,
			"stddev", sentiment.StdDev,
			"variance", sentiment.Variance,
		)
	}
	return nil
}
Run an aggregatation on the provided results.
package centiment
import (
	"context"
	"github.com/go-kit/kit/log"
	"github.com/pkg/errors"
)
type Aggregator struct {
	logger log.Logger
	db     DB
}
func NewAggregator(logger log.Logger, db DB) (*Aggregator, error) {
	agg := &Aggregator{
		db:     db,
		logger: logger,
	}
	return agg, nil
}