func (g *AllGroup) findNewShards(ctx context.Context, shardc chan types.Shard) {
	g.shardMu.Lock()
	defer g.shardMu.Unlock()
	g.logger.Log("[GROUP]", "fetching shards")
	shards, err := listShards(ctx, g.ksis, g.streamName)
	if err != nil {
		g.logger.Log("[GROUP] error:", err)
		return
	}
	for _, shard := range shards {
		if _, ok := g.shards[*shard.ShardId]; ok {
			continue
		}
		g.shards[*shard.ShardId] = shard
		shardc <- shard
	}
}
findNewShards pulls the list of shards from the Kinesis API
and uses a local cache to determine if we are already processing
a particular shard.
package consumer
import (
	"context"
	"sync"
	"time"
	"github.com/aws/aws-sdk-go-v2/service/kinesis/types"
)
func NewAllGroup(ksis kinesisClient, store Store, streamName string, logger Logger) *AllGroup {
	return &AllGroup{
		ksis:       ksis,
		shards:     make(map[string]types.Shard),
		streamName: streamName,
		logger:     logger,
		Store:      store,
	}
}
type AllGroup struct {
	ksis       kinesisClient
	streamName string
	logger     Logger
	Store
	shardMu sync.Mutex
	shards  map[string]types.Shard
}
func (g *AllGroup) Start(ctx context.Context, shardc chan types.Shard) {
	var ticker = time.NewTicker(30 * time.Second)
	for {
		g.findNewShards(ctx, shardc)
		select {
		case <-ctx.Done():
			ticker.Stop()
			return
		case <-ticker.C:
		}
	}
}