dispatcher

Documentation

Index

• type Args
• type NatssDispatcher
  • func NewDispatcher(args Args) (NatssDispatcher, error)
• type SubscriptionChannelMapping
• type SubscriptionsSupervisor
  • func (s *SubscriptionsSupervisor) Connect(ctx context.Context)
  • func (s *SubscriptionsSupervisor) ProcessChannels(ctx context.Context, chanList []messagingv1beta1.Channel) error
  • func (s *SubscriptionsSupervisor) Start(ctx context.Context) error
  • func (s *SubscriptionsSupervisor) UpdateSubscriptions(ctx context.Context, channel *messagingv1beta1.Channel, isFinalizer bool) (map[eventingduck.SubscriberSpec]error, error)

Types

type Args

type Args struct {
	NatssURL  string
	ClusterID string
	ClientID  string
	Cargs     kncloudevents.ConnectionArgs
	Logger    *zap.Logger
}

type NatssDispatcher

type NatssDispatcher interface {
	Start(ctx context.Context) error
	UpdateSubscriptions(ctx context.Context, channel *messagingv1beta1.Channel, isFinalizer bool) (map[eventingduck.SubscriberSpec]error, error)
	ProcessChannels(ctx context.Context, chanList []messagingv1beta1.Channel) error
}

func NewDispatcher

func NewDispatcher(args Args) (NatssDispatcher, error)

NewDispatcher returns a new NatssDispatcher.

type SubscriptionChannelMapping

type SubscriptionChannelMapping map[eventingchannels.ChannelReference]map[types.UID]*stan.Subscription

type SubscriptionsSupervisor

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

SubscriptionsSupervisor manages the state of NATS Streaming subscriptions

func (*SubscriptionsSupervisor) Connect

func (s *SubscriptionsSupervisor) Connect(ctx context.Context)

Connect is called for initial connection as well as after every disconnect

func (*SubscriptionsSupervisor) ProcessChannels

func (s *SubscriptionsSupervisor) ProcessChannels(ctx context.Context, chanList []messagingv1beta1.Channel) error

ProcessChannels will be called from the controller that watches natss channels. It will update internal hostToChannelMap which is used to resolve the hostHeader of the incoming request to the correct ChannelReference in the receiver function.

func (*SubscriptionsSupervisor) Start

func (s *SubscriptionsSupervisor) Start(ctx context.Context) error

func (*SubscriptionsSupervisor) UpdateSubscriptions

func (s *SubscriptionsSupervisor) UpdateSubscriptions(ctx context.Context, channel *messagingv1beta1.Channel, isFinalizer bool) (map[eventingduck.SubscriberSpec]error, error)

UpdateSubscriptions creates/deletes the natss subscriptions based on channel.Spec.Subscribable.Subscribers Return type:map[eventingduck.SubscriberSpec]error --> Returns a map of subscriberSpec that failed with the value=error encountered. Ignore the value in case error != nil