kafka_datasource

Oct 10, 2022

Changes in this version

+ const BalanceStrategyRange

+ const BalanceStrategyRoundRobin

+ const BalanceStrategySticky

+ const DefaultBalanceStrategy

+ const DefaultIsolationLevel

+ const IsolationLevelReadCommitted

+ const IsolationLevelReadUncommitted

+ var DefaultKafkaVersion = "V1_0_0_0"

+ var SaramaSupportedKafkaVersions = map[string]sarama.KafkaVersion

+ func ConfigJSON(config Configuration) json.RawMessage

+ type Configuration struct

+ Subscription SubscriptionConfiguration

+ type Factory struct

+ func (f *Factory) Planner(ctx context.Context) plan.DataSourcePlanner

+ type GraphQLSubscriptionClient interface

+ Subscribe func(ctx context.Context, options GraphQLSubscriptionOptions, next chan<- []byte) error

+ type GraphQLSubscriptionOptions struct

+ BalanceStrategy string

+ BrokerAddresses []string

+ ClientID string

+ GroupID string

+ IsolationLevel string

+ KafkaVersion string

+ SASL SASL

+ StartConsumingLatest bool

+ Topics []string

+ func (g *GraphQLSubscriptionOptions) Sanitize()

+ func (g *GraphQLSubscriptionOptions) Validate() error

+ type KafkaConsumerGroup struct

+ func NewKafkaConsumerGroup(log log.Logger, saramaConfig *sarama.Config, ...) (*KafkaConsumerGroup, error)

+ func (k *KafkaConsumerGroup) Close() error

+ func (k *KafkaConsumerGroup) StartConsuming(messages chan *sarama.ConsumerMessage)

+ func (k *KafkaConsumerGroup) WaitUntilConsumerStop()

+ type KafkaConsumerGroupBridge struct

+ func NewKafkaConsumerGroupBridge(ctx context.Context, logger log.Logger) *KafkaConsumerGroupBridge

+ func (c *KafkaConsumerGroupBridge) Subscribe(ctx context.Context, options GraphQLSubscriptionOptions, next chan<- []byte) error

+ type Planner struct

+ func (p *Planner) ConfigureFetch() plan.FetchConfiguration

+ func (p *Planner) ConfigureSubscription() plan.SubscriptionConfiguration

+ func (p *Planner) DataSourcePlanningBehavior() plan.DataSourcePlanningBehavior

+ func (p *Planner) DownstreamResponseFieldAlias(_ int) (alias string, exists bool)

+ func (p *Planner) Register(_ *plan.Visitor, configuration plan.DataSourceConfiguration, _ bool) error

+ type SASL struct

+ Enable bool

+ Password string

+ User string

+ type SubscriptionConfiguration struct

+ BalanceStrategy string

+ BrokerAddresses []string

+ ClientID string

+ GroupID string

+ IsolationLevel string

+ KafkaVersion string

+ SASL SASL

+ StartConsumingLatest bool

+ Topics []string

+ type SubscriptionSource struct

+ func (s *SubscriptionSource) Start(ctx context.Context, input []byte, next chan<- []byte) error