sink

Documentation

Index

• func SinkFactory(job *model.Job) model.MessageProcessor
• type KafkaMessageSink
  • func NewKafkaMessageSink(context context.Context, config aws.Config, topicName string, brokers []string) *KafkaMessageSink
  • func (k *KafkaMessageSink) Process(message kafka.Message) error
• type MessagePrinter
  • func (p *MessagePrinter) Process(message kafka.Message) error
• type S3MessageSink
  • func NewS3MessageSink(ctx context.Context, config aws.Config, bucketName string) *S3MessageSink
  • func (s *S3MessageSink) Process(message kafka.Message) error
  • func (s *S3MessageSink) Upload(message kafka.Message) error
• type S3ObjectPutter

Functions

func SinkFactory

func SinkFactory(job *model.Job) model.MessageProcessor

Types

type KafkaMessageSink

type KafkaMessageSink struct {
	// contains filtered or unexported fields
}

func NewKafkaMessageSink

func NewKafkaMessageSink(context context.Context, config aws.Config, topicName string, brokers []string) *KafkaMessageSink

func (*KafkaMessageSink) Process

func (k *KafkaMessageSink) Process(message kafka.Message) error

type MessagePrinter

type MessagePrinter struct{}

func (*MessagePrinter) Process

func (p *MessagePrinter) Process(message kafka.Message) error

type S3MessageSink

type S3MessageSink struct {
	Context    context.Context
	S3Client   S3ObjectPutter
	BucketName string
}

func NewS3MessageSink

func NewS3MessageSink(ctx context.Context, config aws.Config, bucketName string) *S3MessageSink

func (*S3MessageSink) Process

func (s *S3MessageSink) Process(message kafka.Message) error

func (*S3MessageSink) Upload

func (s *S3MessageSink) Upload(message kafka.Message) error

type S3ObjectPutter

type S3ObjectPutter interface {
	PutObject(ctx context.Context, params *s3.PutObjectInput, optFns ...func(*s3.Options)) (*s3.PutObjectOutput, error)
}