source

Documentation

Index

• Variables
• func NewPollingSource(ctx context.Context, client s3iface.S3API, logger log.Logger, ...) (*pollingSource, error)
• func NewSQSSource(ctx context.Context, logger log.Logger, reader reader.Reader, ...) (*sqsSource, error)
• func NewSource(src *s3.S3Source, transferID string, logger log.Logger, ...) (abstract.Source, error)
• type Object
• type ObjectFetcher
  • func NewObjectFetcher(ctx context.Context, client s3iface.S3API, logger log.Logger, ...) (ObjectFetcher, error)

Variables

var (
	ReadProgressKey = "ReadProgressKey"
	CreationEvent   = "ObjectCreated:"
	TestEvent       = "s3:TestEvent"
)

Functions

func NewPollingSource

func NewPollingSource(ctx context.Context, client s3iface.S3API, logger log.Logger,
	cp coordinator.Coordinator, transferID string, reader reader.Reader, sourceConfig *s3.S3Source,
) (*pollingSource, error)

func NewSQSSource

func NewSQSSource(ctx context.Context, logger log.Logger, reader reader.Reader,
	sess *session.Session, sourceConfig *s3.S3Source,
) (*sqsSource, error)

func NewSource

func NewSource(src *s3.S3Source, transferID string, logger log.Logger, registry metrics.Registry, cp coordinator.Coordinator) (abstract.Source, error)

Types

type Object

type Object struct {
	Name         string    `json:"name"`
	LastModified time.Time `json:"last_modified"`
}

type ObjectFetcher

type ObjectFetcher interface {
	// FetchObjects derives a list of new objects that need replication from a configured source.
	// This can be a creation event messages from an SQS, SNS, Pub/Sub queue or directly by reading the full object list from the s3 bucket itself.
	FetchObjects() ([]Object, error)

	// Commit persist the processed object to some state.
	// For SQS it deletes the processed messages, for SNS/PubSub it Ack the processed messages
	// and for normal S3 bucket polling it stores the latest object that was read to the transfer state.
	Commit(Object) error
}

func NewObjectFetcher

func NewObjectFetcher(ctx context.Context, client s3iface.S3API, logger log.Logger,
	cp coordinator.Coordinator, transferID string, reader reader.Reader,
	sess *session.Session, sourceConfig *s3.S3Source,
) (ObjectFetcher, error)