<commit_before>package storage

import (
	"fmt"
	"time"

	"github.com/pkg/errors"

	"github.com/yuuki/diamondb/lib/series"
	"github.com/yuuki/diamondb/lib/storage/dynamo"
	"github.com/yuuki/diamondb/lib/storage/redis"
)

// Fetcher defines the interface for data store reader.
type Fetcher interface {
	Fetch(string, time.Time, time.Time) (series.SeriesSlice, error)
	Ping() error
}

// Store provides each data store client.
type Store struct {
	Redis    *redis.Redis
	DynamoDB *dynamo.DynamoDB
	// s3 client
}

// NewStore create a new Store wrapped by Fetcher.
func NewStore() Fetcher {
	return &Store{
		Redis:    redis.NewRedis(),
		DynamoDB: dynamo.NewDynamoDB(),
	}
}

// Ping pings each storage.
func (s *Store) Ping() error {
	rerr := s.Redis.Ping()
	derr := s.DynamoDB.Ping()
	if rerr != nil || derr != nil {
		var errMsg string
		if rerr != nil {
			errMsg += fmt.Sprintf("Redis connection error: %s \n", rerr)
		}
		if derr != nil {
			errMsg += fmt.Sprintf("DynamoDB connection error: %s ", derr)
		}
		return errors.New(errMsg)
	}
	return nil
}

// Fetch fetches series from Redis, DynamoDB and S3.
func (s *Store) Fetch(name string, start, end time.Time) (series.SeriesSlice, error) {
	type item struct {
		result series.SeriesMap
		err    error
	}

	redisCh := make(chan item, 1)
	dynamodbCh := make(chan item, 1)

	// Redis task
	go func(name string, start, end time.Time) {
		sm, err := s.Redis.Fetch(name, start, end)
		redisCh <- item{result: sm, err: err}
	}(name, start, end)

	// DynamoDB task
	go func(name string, start, end time.Time) {
		sm, err := s.DynamoDB.Fetch(name, start, end)
		dynamodbCh <- item{result: sm, err: err}
	}(name, start, end)

	var (
		smR, smD     series.SeriesMap
		rdone, ddone bool
	)
	for {
		select {
		case rit := <-redisCh:
			if rit.err != nil {
				return nil, errors.Wrapf(rit.err, "redis.Fetch(%s,%d,%d)",
					name, start.Unix(), end.Unix(),
				)
			}
			smR = rit.result
			rdone = true
		case dit := <-dynamodbCh:
			if dit.err != nil {
				return nil, errors.Wrapf(dit.err, "dynamodb.Fetch(%s,%d,%d)",
					name, start.Unix(), end.Unix(),
				)
			}
			smD = dit.result
			ddone = true
			// TODO timeout
		}
		if rdone && ddone {
			break
		}
	}
	// TODO S3

	sm := smR.MergePointsToSlice(smD)
	return sm, nil
}
<commit_msg>Refactor the part of concurrent fetching from each storage by using future<commit_after>package storage

import (
	"fmt"
	"time"

	"github.com/pkg/errors"

	"github.com/yuuki/diamondb/lib/series"
	"github.com/yuuki/diamondb/lib/storage/dynamo"
	"github.com/yuuki/diamondb/lib/storage/redis"
)

// Fetcher defines the interface for data store reader.
type Fetcher interface {
	Fetch(string, time.Time, time.Time) (series.SeriesSlice, error)
	Ping() error
}

// Store provides each data store client.
type Store struct {
	Redis    *redis.Redis
	DynamoDB *dynamo.DynamoDB
	// s3 client
}

// NewStore create a new Store wrapped by Fetcher.
func NewStore() Fetcher {
	return &Store{
		Redis:    redis.NewRedis(),
		DynamoDB: dynamo.NewDynamoDB(),
	}
}

// Ping pings each storage.
func (s *Store) Ping() error {
	rerr := s.Redis.Ping()
	derr := s.DynamoDB.Ping()
	if rerr != nil || derr != nil {
		var errMsg string
		if rerr != nil {
			errMsg += fmt.Sprintf("Redis connection error: %s \n", rerr)
		}
		if derr != nil {
			errMsg += fmt.Sprintf("DynamoDB connection error: %s ", derr)
		}
		return errors.New(errMsg)
	}
	return nil
}

type futureSeriesMap struct {
	result series.SeriesMap
	err    error
	done   chan struct{}
}

func (f *futureSeriesMap) Get() (series.SeriesMap, error) {
	<-f.done
	return f.result, f.err
}

// Fetch fetches series from Redis, DynamoDB and S3.
// TODO S3
func (s *Store) Fetch(name string, start, end time.Time) (series.SeriesSlice, error) {
	fredis := futureSeriesMap{
		done: make(chan struct{}, 1),
	}
	fdynamodb := futureSeriesMap{
		done: make(chan struct{}, 1),
	}

	// Redis task
	go func(name string, start, end time.Time) {
		fredis.result, fredis.err = s.Redis.Fetch(name, start, end)
		fredis.done <- struct{}{}
	}(name, start, end)

	// DynamoDB task
	go func(name string, start, end time.Time) {
		fdynamodb.result, fdynamodb.err = s.DynamoDB.Fetch(name, start, end)
		fdynamodb.done <- struct{}{}
	}(name, start, end)

	smR, err := fredis.Get()
	if err != nil {
		return nil, errors.Wrapf(err, "redis.Fetch(%s,%d,%d)",
			name, start.Unix(), end.Unix(),
		)
	}
	smD, err := fdynamodb.Get()
	if err != nil {
		return nil, errors.Wrapf(err, "dynamodb.Fetch(%s,%d,%d)",
			name, start.Unix(), end.Unix(),
		)
	}

	sm := smR.MergePointsToSlice(smD)
	return sm, nil
}

{'fraction_non_alphanumeric': 0.1486633340578135, 'fraction_numerical': 0.0021734405564007822, 'mean_word_length': 3.3211267605633803, 'pattern_counts': {'":': 0, '<': 10, '<?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': '1167057', 'n_tokens_mistral': 1941, 'n_tokens_neox': 1780, 'n_words': 467}