cluster

Documentation

Index

• func NewDelayQueue(parts *persistence.Partitions, ring *Ring) *clusterDelayQueue
• func NewFlowDao(parts *persistence.Partitions, ring *Ring) *clusterFlowDao
• func NewHasher() *hasher
• func NewQueue(parts *persistence.Partitions, ring *Ring) *clusterQueue
• func ValidateStateHandler(st string) error
• type Config
• type DelayQueue
• type FlowDao
• type Handler
• type Membership
  • func New(handler Handler, config Config) (*Membership, error)
  • func (m *Membership) Leave() error
  • func (m *Membership) Members() []serf.Member
  • func (m *Membership) RefreshCluster()
• type Node
  • func (n Node) String() string
• type Queue
• type Ring
  • func NewRing(c RingConfig) *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) RefreshCluster()
• type RingConfig
• type StateHandlerContainer
  • func NewStateHandlerContainer(flowDao FlowDao) *StateHandlerContainer
  • func (s *StateHandlerContainer) GetHandler(st Statehandler) func(wfName string, wfId string) error
  • func (s *StateHandlerContainer) Init()
• type Statehandler

Functions

func NewDelayQueue

func NewDelayQueue(parts *persistence.Partitions, ring *Ring) *clusterDelayQueue

func NewFlowDao

func NewFlowDao(parts *persistence.Partitions, ring *Ring) *clusterFlowDao

func NewHasher

func NewHasher() *hasher

func NewQueue

func NewQueue(parts *persistence.Partitions, ring *Ring) *clusterQueue

func ValidateStateHandler

func ValidateStateHandler(st string) error

Types

type Config

type Config struct {
	NodeName       string
	BindAddr       string
	Tags           map[string]string
	StartJoinAddrs []string
}

type DelayQueue

type DelayQueue interface {
	Push(queueName string, flowId string, mesage []byte) error
	Pop(queueName string) ([]string, error)
	PushWithDelay(queueName string, flowId string, delay time.Duration, message []byte) error
}

type FlowDao

type FlowDao interface {
	SaveFlowContext(wfName string, flowId string, flowCtx *model.FlowContext) error
	CreateAndSaveFlowContext(wFname string, flowId string, action int, dataMap map[string]any) (*model.FlowContext, error)
	AddActionOutputToFlowContext(wFname string, flowId string, action int, dataMap map[string]any) (*model.FlowContext, error)
	GetFlowContext(wfName string, flowId string) (*model.FlowContext, error)
	DeleteFlowContext(wfName string, flowId string) error
}

type Handler

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

type Membership

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

func New

func New(handler Handler, config Config) (*Membership, error)

func (*Membership) Leave

func (m *Membership) Leave() error

func (*Membership) Members

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

func (*Membership) RefreshCluster

func (m *Membership) RefreshCluster()

type Node

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

func (Node) String

func (n Node) String() string

type Queue

type Queue interface {
	Push(queueName string, flowId string, mesage []byte) error
	Pop(queuName string, batchSize int) ([]string, error)
}

type Ring

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

func NewRing

func NewRing(c RingConfig) *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) RefreshCluster

func (r *Ring) RefreshCluster()

type RingConfig

type RingConfig struct {
	PartitionCount int
}

type StateHandlerContainer

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

func NewStateHandlerContainer

func NewStateHandlerContainer(flowDao FlowDao) *StateHandlerContainer

func (*StateHandlerContainer) GetHandler

func (s *StateHandlerContainer) GetHandler(st Statehandler) func(wfName string, wfId string) error

func (*StateHandlerContainer) Init

func (s *StateHandlerContainer) Init()

type Statehandler

type Statehandler string

const DELETE Statehandler = "DELETE"

const NOOP Statehandler = "NOOP"