gki

Apr 8, 2024

Changes in this version

type DefaultAdminClient

+ func (d DefaultAdminClient) Close()

Jan 16, 2023

Changes in this version

type SharedProducerFactory

+ func (s *SharedProducerFactory) CloseSharedProducer()

Jan 10, 2023

Changes in this version

+ func IsNil(v any) bool

type DefaultProducerFactory

+ func (d DefaultProducerFactory) CloseProducer(p Producer)

type ProducerFactory

+ CloseProducer func(p Producer)

type SharedProducerFactory

+ func (s *SharedProducerFactory) CloseProducer(p Producer)

Jan 9, 2023

Changes in this version

+ type SharedProducerFactory struct

+ func NewSharedProducerFactory() *SharedProducerFactory

+ func (s *SharedProducerFactory) NewAdminClientFromProducer(p Producer) (AdminClient, error)

+ func (s *SharedProducerFactory) NewProducer(conf *kafka.ConfigMap) (Producer, error)

Jan 8, 2023

Changes in this version

type DLQConfig

+ ProducerConfig map[string]any

Nov 14, 2022

Changes in this version

type Config

+ func (c *Config) SetDLQConfig(dlqConfig DLQConfig)

+ type DLQConfig struct

+ StreamIDEnrichmentPath string

+ Topic *entity.TopicSpecification

+ func NewDLQConfig(enrichPath string, topic *entity.TopicSpecification) (DLQConfig, error)

type Extractor

+ func (e *Extractor) DLQConfig() DLQConfig

Oct 5, 2022

Changes in this version

type Config

+ func (c *Config) SetLoaderProps(props ConfigMap)

Sep 27, 2022

Changes in this version

type Config

+ func (c *Config) SetSendToSource(value bool)

Aug 21, 2022

Changes in this version

+ const DefaultPollTimeoutMs

+ const PropAutoCommit

+ const PropAutoOffsetStore

type Config

+ func (c *Config) SetCreateTopics(value bool)

+ type Extractor struct

+ func (e *Extractor) Extract(ctx context.Context, query entity.ExtractorQuery, result any) (error, bool)

+ func (e *Extractor) ExtractFromSink(ctx context.Context, query entity.ExtractorQuery, ...) (error, bool)

+ func (e *Extractor) KafkaConfig() (pollTimeoutMs int, cfgMap map[string]any)

+ func (e *Extractor) SendToSource(ctx context.Context, eventData any) (string, error)

+ func (e *Extractor) SetConsumerFactory(cf ConsumerFactory)

+ func (e *Extractor) SetProducerFactory(pf ProducerFactory)

+ func (e *Extractor) StreamExtract(ctx context.Context, reportEvent entity.ProcessEventFunc, err *error, ...)

type Loader

+ func (l *Loader) KafkaConfig() map[string]any

Jul 28, 2022

Changes in this version

+ func NewExtractor(config *Config, id string) (*extractor, error)

+ type AdminClient interface

+ CreateTopics func(ctx context.Context, topics []kafka.TopicSpecification, ...) ([]kafka.TopicResult, error)

+ GetMetadata func(topic *string, allTopics bool, timeoutMs int) (*kafka.Metadata, error)

+ type Config struct

+ func NewExtractorConfig(spec *entity.Spec, topics []string, topicCreationMutex *sync.Mutex) *Config

+ func NewLoaderConfig(spec *entity.Spec, topic *entity.TopicSpecification, ...) *Config

+ func (c *Config) SetKafkaProperty(prop string, value any)

+ func (c *Config) SetPollTimout(timeout int)

+ func (c *Config) SetProps(props ConfigMap)

+ func (c *Config) String() string

+ type ConfigMap map[string]any

+ type Consumer interface

+ Close func() error

+ Commit func() ([]kafka.TopicPartition, error)

+ CommitMessage func(m *kafka.Message) ([]kafka.TopicPartition, error)

+ Poll func(timeoutMs int) (event kafka.Event)

+ StoreOffsets func(offsets []kafka.TopicPartition) ([]kafka.TopicPartition, error)

+ SubscribeTopics func(topics []string, rebalanceCb kafka.RebalanceCb) error

+ type ConsumerFactory interface

+ NewAdminClientFromConsumer func(c Consumer) (AdminClient, error)

+ NewConsumer func(conf *kafka.ConfigMap) (Consumer, error)

+ type DefaultAdminClient struct

+ func (d DefaultAdminClient) CreateTopics(ctx context.Context, topics []kafka.TopicSpecification, ...) ([]kafka.TopicResult, error)

+ func (d DefaultAdminClient) GetMetadata(topic *string, allTopics bool, timeoutMs int) (*kafka.Metadata, error)

+ type DefaultConsumerFactory struct

+ func (d DefaultConsumerFactory) NewAdminClientFromConsumer(c Consumer) (AdminClient, error)

+ func (d DefaultConsumerFactory) NewConsumer(conf *kafka.ConfigMap) (Consumer, error)

+ type DefaultProducerFactory struct

+ func (d DefaultProducerFactory) NewAdminClientFromProducer(p Producer) (AdminClient, error)

+ func (d DefaultProducerFactory) NewProducer(conf *kafka.ConfigMap) (Producer, error)

+ type Loader struct

+ func NewLoader(ctx context.Context, config *Config, id string, pf ProducerFactory) (*Loader, error)

+ func (l *Loader) Shutdown()

+ func (l *Loader) StreamLoad(ctx context.Context, data []*entity.Transformed) (string, error, bool)

+ type Producer interface

+ Close func()

+ Events func() chan kafka.Event

+ Flush func(timeoutMs int) int

+ Produce func(msg *kafka.Message, deliveryChan chan kafka.Event) error

+ type ProducerFactory interface

+ NewAdminClientFromProducer func(p Producer) (AdminClient, error)

+ NewProducer func(conf *kafka.ConfigMap) (Producer, error)