cluster

Documentation

Index

• func NewHasher() *hasher
• type Cluster
  • func NewCluster(conf config.Config, metadataService metadata.MetadataService, ...) *Cluster
  • func (c *Cluster) ExecuteAction(wfName string, flowId string, event string, actionId int, data map[string]any)
  • func (c *Cluster) ExecuteResume(wfName string, flowId string, event string)
  • func (c *Cluster) ExecuteResumeAfterWait(wfName string, flowId string, event string) error
  • func (c *Cluster) GetServerer() *Ring
  • func (c *Cluster) GetShard(shardId int) *shard.Shard
  • func (c *Cluster) Init(wfName string, input map[string]any) (string, error)
  • func (c *Cluster) MarkPaused(wfName string, flowId string)
  • func (c *Cluster) Poll(queuName string, batchSize int) (*api.Actions, error)
  • func (c *Cluster) Rebalance(partitions []int)
  • func (c *Cluster) RetryAction(wfName string, flowId string, actionName string, actionId int, reason string)
  • func (c *Cluster) Start() error
  • func (c *Cluster) Stop() error
  • func (c *Cluster) Timeout(wfName string, flowId string, actionName string, actionId int, ...) error
• type Handler
• type Membership
  • func NewMemberShip(handler Handler, conf config.ClusterConfig) (*Membership, error)
  • func (m *Membership) Leave() error
  • func (m *Membership) Members() []serf.Member
• type Node
  • func (n Node) String() string
• type Ring
  • func NewRing(paritionCount int) *Ring
  • func (r *Ring) GetPartition(key string) int
  • func (r *Ring) GetPartitions() []int
  • func (r *Ring) GetServers() ([]*api_v1.Server, error)
  • func (r *Ring) Join(name, addr string, isLocal bool) error
  • func (r *Ring) Leave(name string) error
  • func (r *Ring) SetRebalancer(reb func([]int))

Functions

func NewHasher

func NewHasher() *hasher

Types

type Cluster

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

func NewCluster

func NewCluster(conf config.Config, metadataService metadata.MetadataService, wg *sync.WaitGroup) *Cluster

func (*Cluster) ExecuteAction

func (c *Cluster) ExecuteAction(wfName string, flowId string, event string, actionId int, data map[string]any)

func (*Cluster) ExecuteResume

func (c *Cluster) ExecuteResume(wfName string, flowId string, event string)

func (*Cluster) ExecuteResumeAfterWait

func (c *Cluster) ExecuteResumeAfterWait(wfName string, flowId string, event string) error

func (*Cluster) GetServerer

func (c *Cluster) GetServerer() *Ring

func (*Cluster) GetShard

func (c *Cluster) GetShard(shardId int) *shard.Shard

func (*Cluster) Init

func (c *Cluster) Init(wfName string, input map[string]any) (string, error)

func (*Cluster) MarkPaused

func (c *Cluster) MarkPaused(wfName string, flowId string)

func (*Cluster) Poll

func (c *Cluster) Poll(queuName string, batchSize int) (*api.Actions, error)

func (*Cluster) Rebalance

func (c *Cluster) Rebalance(partitions []int)

func (*Cluster) RetryAction

func (c *Cluster) RetryAction(wfName string, flowId string, actionName string, actionId int, reason string)

func (*Cluster) Start

func (c *Cluster) Start() error

func (*Cluster) Stop

func (c *Cluster) Stop() error

func (*Cluster) Timeout

func (c *Cluster) Timeout(wfName string, flowId string, actionName string, actionId int, delay time.Duration) error

type Handler

type Handler interface {
	Join(name, addr string, isLocal bool) error
	Leave(name string) error
}

type Membership

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

func NewMemberShip

func NewMemberShip(handler Handler, conf config.ClusterConfig) (*Membership, error)

func (*Membership) Leave

func (m *Membership) Leave() error

func (*Membership) Members

func (m *Membership) Members() []serf.Member

type Node

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

func (Node) String

func (n Node) String() string

type Ring

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

func NewRing

func NewRing(paritionCount int) *Ring

func (*Ring) GetPartition

func (r *Ring) GetPartition(key string) int

func (*Ring) GetPartitions

func (r *Ring) GetPartitions() []int

func (*Ring) GetServers

func (r *Ring) GetServers() ([]*api_v1.Server, error)

func (*Ring) Join

func (r *Ring) Join(name, addr string, isLocal bool) error

func (*Ring) Leave

func (r *Ring) Leave(name string) error

func (*Ring) SetRebalancer

func (r *Ring) SetRebalancer(reb func([]int))