上传基础的集群相关代码

Author: iwind

teacluster/action.go

@@ -0,0 +1,27 @@
+package teacluster
+
+import (
+	"time"
+)
+
+type Action struct {
+	Id          uint64
+	RequestTime time.Time
+	RequestId   uint64
+}
+
+func (this *Action) Execute() error {
+	return nil
+}
+
+func (this *Action) OnSuccess(success *SuccessAction) error {
+	return nil
+}
+
+func (this *Action) OnFail(fail *FailAction) error {
+	return nil
+}
+
+func (this *Action) BaseAction() *Action {
+	return this
+}

teacluster/action_fail.go

@@ -0,0 +1,15 @@
+package teacluster
+
+type FailAction struct {
+	Action
+
+	Message string
+}
+
+func (this *FailAction) Name() string {
+	return "fail"
+}
+
+func (this *FailAction) TypeId() int8 {
+	return 2
+}

teacluster/action_interface.go

@@ -0,0 +1,10 @@
+package teacluster
+
+type ActionInterface interface {
+	Name() string
+	Execute() error
+	OnSuccess(success *SuccessAction) error
+	OnFail(fail *FailAction) error
+	TypeId() int8
+	BaseAction() *Action
+}

teacluster/action_notify.go

@@ -0,0 +1,19 @@
+package teacluster
+
+// cluster -> slave node
+type NotifyAction struct {
+	Action
+}
+
+func (this *NotifyAction) Name() string {
+	return "notify"
+}
+
+func (this *NotifyAction) TypeId() int8 {
+	return 4
+}
+
+func (this *NotifyAction) Execute() error {
+	ClusterManager.Write(&SumAction{})
+	return nil
+}

teacluster/action_pull.go

@@ -0,0 +1,24 @@
+package teacluster
+
+import (
+	"github.com/TeaWeb/code/teacluster/configs"
+)
+
+// node <- cluster
+type PullAction struct {
+	Action
+
+	LocalItems []*configs.Item // items without data
+}
+
+func (this *PullAction) Name() string {
+	return "pull"
+}
+
+func (this *PullAction) Execute() error {
+	return nil
+}
+
+func (this *PullAction) TypeId() int8 {
+	return 6
+}

teacluster/action_push.go

@@ -0,0 +1,41 @@
+package teacluster
+
+import (
+	"github.com/TeaWeb/code/teacluster/configs"
+	"github.com/iwind/TeaGo/logs"
+)
+
+// master -> cluster
+type PushAction struct {
+	Action
+
+	Items []*configs.Item
+}
+
+func (this *PushAction) Name() string {
+	return "push"
+}
+
+func (this *PushAction) Execute() error {
+	return nil
+}
+
+func (this *PushAction) AddItem(item *configs.Item) {
+	this.Items = append(this.Items, item)
+}
+
+func (this *PushAction) OnSuccess(success *SuccessAction) error {
+	return nil
+}
+
+func (this *PushAction) OnFail(fail *FailAction) error {
+	logs.Println("[push]fail:", fail.Message)
+
+	// TODO retry later
+
+	return nil
+}
+
+func (this *PushAction) TypeId() int8 {
+	return 5
+}

teacluster/action_register.go

@@ -0,0 +1,43 @@
+package teacluster
+
+import (
+	"github.com/TeaWeb/code/teaconfigs"
+	"github.com/iwind/TeaGo/logs"
+)
+
+type RegisterAction struct {
+	Action
+
+	ClusterId     string
+	ClusterSecret string
+	NodeId        string
+	NodeName      string
+	NodeRole      string
+}
+
+func (this *RegisterAction) Name() string {
+	return "register"
+}
+
+func (this *RegisterAction) Execute() error {
+	return nil
+}
+
+func (this *RegisterAction) OnSuccess(success *SuccessAction) error {
+	if this.NodeRole == teaconfigs.NodeRoleMaster {
+		logs.Println("[cluster]register master ok")
+		ClusterManager.Write(&SumAction{})
+	} else {
+		logs.Println("[cluster]register node ok")
+	}
+	return nil
+}
+
+func (this *RegisterAction) OnFail(fail *FailAction) error {
+	logs.Println("[cluster]fail to register node:", fail.Message)
+	return nil
+}
+
+func (this *RegisterAction) TypeId() int8 {
+	return 3
+}

teacluster/action_success.go

@@ -0,0 +1,18 @@
+package teacluster
+
+import "github.com/iwind/TeaGo/maps"
+
+type SuccessAction struct {
+	Action
+
+	Message string
+	Data    maps.Map
+}
+
+func (this *SuccessAction) Name() string {
+	return "success"
+}
+
+func (this *SuccessAction) TypeId() int8 {
+	return 1
+}

teacluster/action_sum.go

@@ -0,0 +1,33 @@
+package teacluster
+
+import (
+	"github.com/iwind/TeaGo/logs"
+)
+
+// cluster -> master|node
+type SumAction struct {
+	Action
+}
+
+func (this *SumAction) Name() string {
+	return "sum"
+}
+
+func (this *SumAction) OnSuccess(success *SuccessAction) error {
+	logs.Println("sum:", success.Data)
+
+	// write to local file
+	//file := files.NewFile(Tea.ConfigFile("cluster.sum"))
+	//file.WriteString()
+
+	return nil
+}
+
+func (this *SumAction) OnFail(fail *FailAction) error {
+	// TODO retry later
+	return nil
+}
+
+func (this *SumAction) TypeId() int8 {
+	return 9
+}

teacluster/action_utils.go

@@ -0,0 +1,39 @@
+package teacluster
+
+import (
+	"errors"
+	"fmt"
+	"github.com/iwind/TeaGo/logs"
+	"github.com/vmihailenco/msgpack"
+	"reflect"
+	"sync/atomic"
+)
+
+var actionMap = map[int8]reflect.Type{}
+var actionId uint64
+
+func RegisterActionType(actions ...ActionInterface) {
+	for _, action := range actions {
+		typeId := action.TypeId()
+		_, ok := actionMap[typeId]
+		if ok {
+			logs.Error(errors.New("action type '" + fmt.Sprintf("%d", typeId) + "' already exist"))
+			continue
+		}
+
+		msgpack.RegisterExt(typeId, action)
+		actionMap[typeId] = reflect.TypeOf(action).Elem()
+	}
+}
+
+func FindActionInstance(typeId int8) ActionInterface {
+	t, ok := actionMap[typeId]
+	if !ok {
+		return nil
+	}
+	return reflect.New(t).Interface().(ActionInterface)
+}
+
+func GenerateActionId() uint64 {
+	return atomic.AddUint64(&actionId, 1)
+}