actorcomm

Documentation

Index

• Constants
• func AtLeastOnceDeliveryMsgReceiver() chan interface{}
• func ContainerRouterMsgReceiver() chan interface{}
• func GetActorSystem() *actor.ActorSystem
• func GetContainerRouterPid(workerIdAddr string) *actor.PID
• func GetHeartbeatActorPid(workerIdAddr string) *actor.PID
• func GetMapMasterPid(workerIdAddr string) *actor.PID
• func GetRealWorkerAddr(workerIdAddr string) string
• func HeartbeatMsgReceiver() chan interface{}
• func IsSchedulerxServer(pid *actor.PID) bool
• func SchedulerxServerPid(ctx context.Context) *actor.PID
• func SxMsgReceiver() chan interface{}
• func TaskMasterMsgReceiver() chan interface{}
• func WorkerBatchCreateTasksRespMsgSender() chan *schedulerx.WorkerBatchCreateTasksResponse
• func WorkerBatchUpdateTaskStatusRespMsgSender() chan *schedulerx.WorkerBatchUpdateTaskStatusResponse
• func WorkerClearTasksRespMsgSender() chan *schedulerx.WorkerClearTasksResponse
• func WorkerMapTaskRespMsgSender() chan *schedulerx.WorkerMapTaskResponse
• func WorkerPullTasksRespMsgSender() chan *schedulerx.WorkerPullTasksResponse
• func WorkerQueryJobInstanceStatusRespMsgSender() chan *schedulerx.WorkerQueryJobInstanceStatusResponse
• func WorkerReportTaskListStatusRespMsgSender() chan *schedulerx.WorkerReportTaskListStatusResponse
• type SchedulerWrappedMsg
  • func WrapSchedulerxMsg(ctx context.Context, msg interface{}, senderPath string) *SchedulerWrappedMsg

Constants

const (
	ContainerRouterPidId     = "user_container_routing"
	JobInstancePidId         = "job_instance_routing"
	MapMasterPidId           = "map_master_router"
	AtLeastOnceDeliveryPidId = "at_least_once_delivery_routing"
	HeartbeatPidId           = "heartbeat_routing"
)

Functions

func AtLeastOnceDeliveryMsgReceiver

func AtLeastOnceDeliveryMsgReceiver() chan interface{}

func ContainerRouterMsgReceiver

func ContainerRouterMsgReceiver() chan interface{}

func GetActorSystem

func GetActorSystem() *actor.ActorSystem

GetActorSystem must be executed before InitActorSystem, otherwise it returns nil

func GetContainerRouterPid

func GetContainerRouterPid(workerIdAddr string) *actor.PID

GetContainerRouterPid get remote PID of container router

func GetHeartbeatActorPid

func GetHeartbeatActorPid(workerIdAddr string) *actor.PID

GetHeartbeatActorPid get remote PID of heartbeat actor

func GetMapMasterPid

func GetMapMasterPid(workerIdAddr string) *actor.PID

GetMapMasterPid get remote PID of mapMaster router

func GetRealWorkerAddr

func GetRealWorkerAddr(workerIdAddr string) string

GetRealWorkerAddr get the real workerAddr, which is the address of the remote worker's ActorSystem The workerAddr issued by the server is the address reported by the heartbeat. It is the connection address obtained from the connection pool, not the ActorSystem address, so it needs to be converted.

func HeartbeatMsgReceiver

func HeartbeatMsgReceiver() chan interface{}

func IsSchedulerxServer

func IsSchedulerxServer(pid *actor.PID) bool

func SchedulerxServerPid

func SchedulerxServerPid(ctx context.Context) *actor.PID

func SxMsgReceiver

func SxMsgReceiver() chan interface{}

func TaskMasterMsgReceiver

func TaskMasterMsgReceiver() chan interface{}

func WorkerBatchCreateTasksRespMsgSender

func WorkerBatchCreateTasksRespMsgSender() chan *schedulerx.WorkerBatchCreateTasksResponse

func WorkerBatchUpdateTaskStatusRespMsgSender

func WorkerBatchUpdateTaskStatusRespMsgSender() chan *schedulerx.WorkerBatchUpdateTaskStatusResponse

func WorkerClearTasksRespMsgSender

func WorkerClearTasksRespMsgSender() chan *schedulerx.WorkerClearTasksResponse

func WorkerMapTaskRespMsgSender

func WorkerMapTaskRespMsgSender() chan *schedulerx.WorkerMapTaskResponse

func WorkerPullTasksRespMsgSender

func WorkerPullTasksRespMsgSender() chan *schedulerx.WorkerPullTasksResponse

func WorkerQueryJobInstanceStatusRespMsgSender

func WorkerQueryJobInstanceStatusRespMsgSender() chan *schedulerx.WorkerQueryJobInstanceStatusResponse

func WorkerReportTaskListStatusRespMsgSender

func WorkerReportTaskListStatusRespMsgSender() chan *schedulerx.WorkerReportTaskListStatusResponse

Types

type SchedulerWrappedMsg

type SchedulerWrappedMsg struct {
	Ctx        context.Context
	Msg        interface{}
	SenderPath string
}

func WrapSchedulerxMsg

func WrapSchedulerxMsg(ctx context.Context, msg interface{}, senderPath string) *SchedulerWrappedMsg