Method ReadFromStream

pkg/acquisition/modules/kinesis/run.go:389–431

(ctx context.Context, out chan pipeline.Event, t *tomb.Tomb)

}

func (s *Source) ReadFromStream(ctx context.Context, out chan pipeline.Event, t *tomb.Tomb) error {
	s.logger = s.logger.WithField("stream", s.Config.StreamName)
	s.logger.Info("starting kinesis acquisition from shards")

	for {
		shards, err := s.kClient.ListShards(ctx, &kinesis.ListShardsInput{
				StreamName: aws.String(s.Config.StreamName),
			})
		if err != nil {
			return fmt.Errorf("cannot list shards: %w", err)
		}

		s.shardReaderTomb = &tomb.Tomb{}

		for _, shard := range shards.Shards {
			shardID := *shard.ShardId

			s.shardReaderTomb.Go(func() error {
				defer trace.ReportPanic()
				return s.ReadFromShard(ctx, out, shardID)
			})
		}

		select {
		case <-t.Dying():
			s.logger.Info("kinesis source is dying")
			s.shardReaderTomb.Kill(nil)
			_ = s.shardReaderTomb.Wait() // we don't care about the error as we kill the tomb ourselves

			return nil
		case <-s.shardReaderTomb.Dying():
			reason := s.shardReaderTomb.Err()
			if reason != nil {
				s.logger.Errorf("Unexpected error from shard reader : %s", reason)
				return reason
			}

			s.logger.Infof("All shards have been closed, probably a resharding event, restarting acquisition")

			continue
		}
	}
}

func (s *Source) StreamingAcquisition(ctx context.Context, out chan pipeline.Event, t *tomb.Tomb) error {
	t.Go(func() error {