async

Documentation

Index

• Constants
• func BuildHostAddress(node *memberlist.Node) string
• type ClusterDelegate
  • func (d *ClusterDelegate) GetBroadcasts(overhead, limit int) [][]byte
  • func (d *ClusterDelegate) LocalState(join bool) []byte
  • func (d *ClusterDelegate) MergeRemoteState(buf []byte, join bool)
  • func (d *ClusterDelegate) NodeMeta(limit int) []byte
  • func (d *ClusterDelegate) NotifyMsg(msg []byte)
• type ClusterDelegateMetaData
  • func ParseClusterDelegateMetaData(data []byte) (ClusterDelegateMetaData, error)
  • func (m ClusterDelegateMetaData) Bytes() []byte
• type ClusterEventDelegate
  • func (d *ClusterEventDelegate) GetAsyncServerAddressFor(shardId int32) string
  • func (d *ClusterEventDelegate) NotifyJoin(node *memberlist.Node)
  • func (d *ClusterEventDelegate) NotifyLeave(node *memberlist.Node)
  • func (d *ClusterEventDelegate) NotifyUpdate(node *memberlist.Node)
• type Membership
  • func NewMembershipImpl(cfg config.Config, logger log.Logger, asyncService *Service, serverType string) Membership
• type Server
  • func NewDefaultAsyncServerWithGin(rootCtx context.Context, cfg config.Config, ...) Server
• type Service
  • func NewAsyncServiceImpl(rootCtx context.Context, processStore persistence.ProcessStore, ...) Service

Constants

const (
	ServerTypeApi   = "api"
	ServerTypeAsync = "async"
)

const PathNotifyImmediateTasks = "/internal/api/v1/xcherry/notify-immediate-tasks"

const PathNotifyTimerTasks = "/internal/api/v1/xcherry/notify-timer-tasks"

const PathWaitForProcessCompletion = "/internal/api/v1/xcherry/wait-for-process-completion"

Functions

func BuildHostAddress

func BuildHostAddress(node *memberlist.Node) string

Types

type ClusterDelegate

type ClusterDelegate struct {
	Meta ClusterDelegateMetaData
}

func (*ClusterDelegate) GetBroadcasts

func (d *ClusterDelegate) GetBroadcasts(overhead, limit int) [][]byte

func (*ClusterDelegate) LocalState

func (d *ClusterDelegate) LocalState(join bool) []byte

func (*ClusterDelegate) MergeRemoteState

func (d *ClusterDelegate) MergeRemoteState(buf []byte, join bool)

func (*ClusterDelegate) NodeMeta

func (d *ClusterDelegate) NodeMeta(limit int) []byte

func (*ClusterDelegate) NotifyMsg

func (d *ClusterDelegate) NotifyMsg(msg []byte)

type ClusterDelegateMetaData

type ClusterDelegateMetaData struct {
	ServerType    string
	ServerAddress string
}

func ParseClusterDelegateMetaData

func ParseClusterDelegateMetaData(data []byte) (ClusterDelegateMetaData, error)

func (ClusterDelegateMetaData) Bytes

func (m ClusterDelegateMetaData) Bytes() []byte

type ClusterEventDelegate

type ClusterEventDelegate struct {
	Logger        log.Logger
	Shard         int
	ServerAddress string
	AsyncService  *Service
	// contains filtered or unexported fields
}

func (*ClusterEventDelegate) GetAsyncServerAddressFor

func (d *ClusterEventDelegate) GetAsyncServerAddressFor(shardId int32) string

func (*ClusterEventDelegate) NotifyJoin

func (d *ClusterEventDelegate) NotifyJoin(node *memberlist.Node)

func (*ClusterEventDelegate) NotifyLeave

func (d *ClusterEventDelegate) NotifyLeave(node *memberlist.Node)

func (*ClusterEventDelegate) NotifyUpdate

func (d *ClusterEventDelegate) NotifyUpdate(node *memberlist.Node)

type Membership

type Membership interface {
	GetServerAddress() string
	GetAsyncServerAddressForShard(shardId int32) string
}

func NewMembershipImpl

func NewMembershipImpl(cfg config.Config, logger log.Logger, asyncService *Service, serverType string) Membership

type Server

type Server interface {
	// Start will start running on the background
	Start() error
	Stop(ctx context.Context) error
}

func NewDefaultAsyncServerWithGin

func NewDefaultAsyncServerWithGin(
	rootCtx context.Context,
	cfg config.Config,
	processStore persistence.ProcessStore,
	visibilityStore persistence.VisibilityStore,
	logger log.Logger,
) Server

type Service

type Service interface {
	Start() error
	NotifyPollingImmediateTask(req xcapi.NotifyImmediateTasksRequest) error
	NotifyPollingTimerTask(req xcapi.NotifyTimerTasksRequest) error
	NotifyRemoteImmediateTaskAsyncInCluster(req xcapi.NotifyImmediateTasksRequest, serverAddress string)
	NotifyRemoteTimerTaskAsyncInCluster(req xcapi.NotifyTimerTasksRequest, serverAddress string)
	AskRemoteToWaitForProcessCompletionInCluster(ctx context.Context, req xcapi.WaitForProcessCompletionRequest,
		serverAddress string) (*xcapi.WaitForProcessCompletionResponse, error)
	Stop(ctx context.Context) error
	ReBalance(assignedShardIds []int32)
	WaitForProcessCompletion(ctx context.Context, req xcapi.WaitForProcessCompletionRequest) (*xcapi.WaitForProcessCompletionResponse, error)
}

func NewAsyncServiceImpl

func NewAsyncServiceImpl(
	rootCtx context.Context, processStore persistence.ProcessStore,
	visibilityStore persistence.VisibilityStore,
	cfg config.Config, logger log.Logger,
) Service