policy

Documentation

Index

• type DefaultPolicyExecutor
  • func NewDefaultPolicyExecutor(bootstrapper *bootstrapping.BootstrapService) *DefaultPolicyExecutor
  • func (e *DefaultPolicyExecutor) Flush() error
  • func (e *DefaultPolicyExecutor) GetSubscriptionMappings(topicName string) (topic.SubscriptionInfo, error)
  • func (e *DefaultPolicyExecutor) GetTopicFragmentMappings(topicName string) (topic.FragMappingInfo, error)
  • func (e *DefaultPolicyExecutor) OnPublisherAdded(id string, topicName string, host string) error
  • func (e *DefaultPolicyExecutor) OnPublisherRemoved(id string, topicName string) error
  • func (e *DefaultPolicyExecutor) OnSubscriberAdded(id string, topicName string) error
  • func (e *DefaultPolicyExecutor) OnSubscriberRemoved(id string, topicName string) error
  • func (e *DefaultPolicyExecutor) UpdateSubscriptionMappings(topicName string, subscriptionMappings topic.SubscriptionInfo)
  • func (e *DefaultPolicyExecutor) UpdateTopicFragments(topicName string, fragMappings topic.FragMappingInfo)
• type DistributionPolicyExecutor
  • func NewDistributionPolicyExecutor(bootstrapper *bootstrapping.BootstrapService) *DistributionPolicyExecutor
  • func (e *DistributionPolicyExecutor) Flush() error
  • func (e *DistributionPolicyExecutor) GetSubscriptionMappings(topicName string) (topic.SubscriptionInfo, error)
  • func (e *DistributionPolicyExecutor) GetTopicFragmentMappings(topicName string) (topic.FragMappingInfo, error)
  • func (d *DistributionPolicyExecutor) OnPublisherAdded(id string, topicName string, host string) error
  • func (d *DistributionPolicyExecutor) OnPublisherRemoved(id string, topicName string) error
  • func (d *DistributionPolicyExecutor) OnSubscriberAdded(id string, topicName string) error
  • func (d *DistributionPolicyExecutor) OnSubscriberRemoved(id string, topicName string) error
  • func (e *DistributionPolicyExecutor) UpdateSubscriptionMappings(topicName string, subscriptionMappings topic.SubscriptionInfo)
  • func (e *DistributionPolicyExecutor) UpdateTopicFragments(topicName string, fragMappings topic.FragMappingInfo)
• type Executor
• type Flushable
• type FlushableExecutor

Types

type DefaultPolicyExecutor

type DefaultPolicyExecutor struct {
	Executor
	// contains filtered or unexported fields
}

func NewDefaultPolicyExecutor

func NewDefaultPolicyExecutor(bootstrapper *bootstrapping.BootstrapService) *DefaultPolicyExecutor

func (*DefaultPolicyExecutor) Flush

func (e *DefaultPolicyExecutor) Flush() error

Flush : flush subscription info and fragment mappings into zk this is safe because only master broker do rebalance process

func (*DefaultPolicyExecutor) GetSubscriptionMappings

func (e *DefaultPolicyExecutor) GetSubscriptionMappings(topicName string) (topic.SubscriptionInfo, error)

func (*DefaultPolicyExecutor) GetTopicFragmentMappings

func (e *DefaultPolicyExecutor) GetTopicFragmentMappings(topicName string) (topic.FragMappingInfo, error)

func (*DefaultPolicyExecutor) OnPublisherAdded

func (e *DefaultPolicyExecutor) OnPublisherAdded(id string, topicName string, host string) error

OnPublisherAdded : when a publisher connects, it either sets the fragment active or adds a new one then add newly active fragments to each subscription equally

func (*DefaultPolicyExecutor) OnPublisherRemoved

func (e *DefaultPolicyExecutor) OnPublisherRemoved(id string, topicName string) error

OnPublisherRemoved : when a publisher is disconnected, set fragment as inactive and remove fragments from subscriptions

func (*DefaultPolicyExecutor) OnSubscriberAdded

func (e *DefaultPolicyExecutor) OnSubscriberAdded(id string, topicName string) error

OnSubscriberAdded : when a subscriber connected, create new subscription for it and assign all fragments of topic.

func (*DefaultPolicyExecutor) OnSubscriberRemoved

func (e *DefaultPolicyExecutor) OnSubscriberRemoved(id string, topicName string) error

OnSubscriberRemoved : when a subscriber disconnected, just delete subscription of it.

func (*DefaultPolicyExecutor) UpdateSubscriptionMappings

func (e *DefaultPolicyExecutor) UpdateSubscriptionMappings(topicName string, subscriptionMappings topic.SubscriptionInfo)

func (*DefaultPolicyExecutor) UpdateTopicFragments

func (e *DefaultPolicyExecutor) UpdateTopicFragments(topicName string, fragMappings topic.FragMappingInfo)

type DistributionPolicyExecutor

type DistributionPolicyExecutor struct {
	Executor
	// contains filtered or unexported fields
}

func NewDistributionPolicyExecutor

func NewDistributionPolicyExecutor(bootstrapper *bootstrapping.BootstrapService) *DistributionPolicyExecutor

func (*DistributionPolicyExecutor) Flush

func (e *DistributionPolicyExecutor) Flush() error

Flush : flush subscription info and fragment mappings into zk this is safe because only master broker do rebalance process

func (*DistributionPolicyExecutor) GetSubscriptionMappings

func (e *DistributionPolicyExecutor) GetSubscriptionMappings(topicName string) (topic.SubscriptionInfo, error)

func (*DistributionPolicyExecutor) GetTopicFragmentMappings

func (e *DistributionPolicyExecutor) GetTopicFragmentMappings(topicName string) (topic.FragMappingInfo, error)

func (*DistributionPolicyExecutor) OnPublisherAdded

func (d *DistributionPolicyExecutor) OnPublisherAdded(id string, topicName string, host string) error

OnPublisherAdded : when a publisher connects, it either sets the fragment active or adds a new one until it completes num_subscribers. then assign newly active fragments to each subscription one by one (round-robin)

func (*DistributionPolicyExecutor) OnPublisherRemoved

func (d *DistributionPolicyExecutor) OnPublisherRemoved(id string, topicName string) error

OnPublisherRemoved : when a publisher is disconnected, set fragment as inactive and remove fragments from subscriptions

func (*DistributionPolicyExecutor) OnSubscriberAdded

func (d *DistributionPolicyExecutor) OnSubscriberAdded(id string, topicName string) error

OnSubscriberAdded : when a subscriber connected, create new subscription for it and assign new 1*num_publisher fragments.

func (*DistributionPolicyExecutor) OnSubscriberRemoved

func (d *DistributionPolicyExecutor) OnSubscriberRemoved(id string, topicName string) error

OnSubscriberRemoved : when a subscriber disconnected, delete subscription of it and set subscribing fragments as stale

func (*DistributionPolicyExecutor) UpdateSubscriptionMappings

func (e *DistributionPolicyExecutor) UpdateSubscriptionMappings(topicName string, subscriptionMappings topic.SubscriptionInfo)

func (*DistributionPolicyExecutor) UpdateTopicFragments

func (e *DistributionPolicyExecutor) UpdateTopicFragments(topicName string, fragMappings topic.FragMappingInfo)

type Executor

type Executor interface {
	OnPublisherAdded(id string, topicName string, host string) error
	OnPublisherRemoved(id string, topicName string) error
	OnSubscriberAdded(id string, topicName string) error
	OnSubscriberRemoved(id string, topicName string) error
}

type Flushable

type Flushable interface {
	Flush() error
}

type FlushableExecutor

type FlushableExecutor interface {
	Flushable
	Executor
}