dispatcher

Documentation

Overview

Copyright 2021 The Knative Authors

Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

Index

• type ChannelConfig
  • func (cc ChannelConfig) SubscriptionsUIDs() []string
• type KafkaDispatcher
  • func NewDispatcher(ctx context.Context, args *KafkaDispatcherArgs, ...) (*KafkaDispatcher, error)
  • func (d *KafkaDispatcher) CleanupChannel(name, namespace, hostname string) error
  • func (d *KafkaDispatcher) ReconcileConsumers(ctx context.Context, config *ChannelConfig) error
  • func (d *KafkaDispatcher) RegisterChannelHost(channelConfig *ChannelConfig) error
  • func (d *KafkaDispatcher) Start(ctx context.Context) error
• type KafkaDispatcherArgs
• type KafkaSubscription
  • func NewKafkaSubscription(logger *zap.SugaredLogger) *KafkaSubscription
  • func (ks *KafkaSubscription) SetReady(subID types.UID, partition int32, ready bool)
• type Subscription
  • func (sub Subscription) String() string
• type TopicFunc
• type UpdateError
  • func (k UpdateError) Error() string

Types

type ChannelConfig

type ChannelConfig struct {
	Namespace     string
	Name          string
	HostName      string
	Subscriptions []Subscription
}

func (ChannelConfig) SubscriptionsUIDs

func (cc ChannelConfig) SubscriptionsUIDs() []string

type KafkaDispatcher

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

func NewDispatcher

func NewDispatcher(ctx context.Context, args *KafkaDispatcherArgs, enqueue func(ref types.NamespacedName)) (*KafkaDispatcher, error)

NewDispatcher creates a new dispatcher struct. enqueue argument is a function that is used to requeue a KafkaChannel instance via the reconciler which is used when creating the consumer.

func (*KafkaDispatcher) CleanupChannel

func (d *KafkaDispatcher) CleanupChannel(name, namespace, hostname string) error

func (*KafkaDispatcher) ReconcileConsumers

func (d *KafkaDispatcher) ReconcileConsumers(ctx context.Context, config *ChannelConfig) error

ReconcileConsumers will be called by new CRD based kafka channel dispatcher controller.

func (*KafkaDispatcher) RegisterChannelHost

func (d *KafkaDispatcher) RegisterChannelHost(channelConfig *ChannelConfig) error

RegisterChannelHost adds a new channel to the host-channel mapping.

func (*KafkaDispatcher) Start

func (d *KafkaDispatcher) Start(ctx context.Context) error

Start starts the kafka dispatcher's message processing.

type KafkaDispatcherArgs

type KafkaDispatcherArgs struct {
	Brokers   []string
	Config    *config.EventingKafkaConfig
	TopicFunc TopicFunc
}

type KafkaSubscription

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

func NewKafkaSubscription

func NewKafkaSubscription(logger *zap.SugaredLogger) *KafkaSubscription

func (*KafkaSubscription) SetReady

func (ks *KafkaSubscription) SetReady(subID types.UID, partition int32, ready bool)

SetReady will mark the subid in the KafkaSubscription and call any registered callbacks

type Subscription

type Subscription struct {
	UID types.UID
	fanout.Subscription
}

func (Subscription) String

func (sub Subscription) String() string

type TopicFunc

type TopicFunc func(separator, namespace, name string) string

type UpdateError

type UpdateError map[types.UID]error

UpdateError is the error returned from the ReconcileConsumers method, with the details of which subscriptions failed to subscribe to.

func (UpdateError) Error

func (k UpdateError) Error() string