Update coordinator, now get connected to collaborator, refactor HTTP status code

Author: HastingsYoung

artifacts/service/queryResource.go

@@ -5,7 +5,7 @@ import (
 )
 
 type QueryResource struct {
-	Resource   *restful.Resource
+	*restful.Resource
 	Attributes Query `json:"attributes"`
 }
 

artifacts/service/registryResource.go

@@ -5,7 +5,7 @@ import (
 )
 
 type RegistryResource struct {
-	Resource   *restful.Resource
+	*restful.Resource
 	Attributes Registry `json:"attributes"`
 }
 

artifacts/service/service.go

@@ -5,10 +5,11 @@ import (
 	"github.com/GoCollaborate/artifacts/parameter"
 	"github.com/GoCollaborate/constants"
 	"math/rand"
+	"time"
 )
 
 type Service struct {
-	ServiceID   string                `json:"serviceid"`
+	ServiceID   string                `json:"serviceid,omitempty"`
 	Description string                `json:"description"`
 	Parameters  []parameter.Parameter `json:"parameters"`
 	RegList     []card.Card           `json:"registers"`
@@ -24,6 +25,19 @@ type Service struct {
 	LastAssignedTime int64            `json:"last_assigned_time,omitempty"`
 }
 
+func NewService() *Service {
+	return &Service{
+		Description:     "",
+		Parameters:      []parameter.Parameter{},
+		RegList:         []card.Card{},
+		Heartbeats:      map[string]int64{},
+		SbscrbList:      []string{},
+		Dependencies:    []string{},
+		Version:         "1.0",
+		PlatformVersion: "golang1.8.1",
+	}
+}
+
 func (s *Service) SetMode(m *Mode) Mode {
 	s.Mode = *m
 	return *m
@@ -119,3 +133,13 @@ func (s *Service) UnSubscribeAll() error {
 	s.SbscrbList = y
 	return nil
 }
+
+func (s *Service) Heartbeat(agt *card.Card) {
+	y := s.RegList
+	for _, x := range y {
+		if agt.IsEqualTo(&x) {
+			s.Heartbeats[x.GetFullEndPoint()] = time.Now().Unix()
+			return
+		}
+	}
+}

artifacts/service/serviceResource.go

@@ -5,7 +5,7 @@ import (
 )
 
 type ServiceResource struct {
-	Resource   *restful.Resource
+	*restful.Resource
 	Attributes Service `json:"attributes"`
 }
 

artifacts/service/subscriptionResource.go

@@ -5,7 +5,7 @@ import (
 )
 
 type SubscriptionResource struct {
-	Resource   *restful.Resource
+	*restful.Resource
 	Attributes Subscription `json:"attributes"`
 }
 

collaborator/collaborator.go

@@ -7,6 +7,7 @@ import (
 	"github.com/GoCollaborate/artifacts/iremote"
 	"github.com/GoCollaborate/artifacts/iworkable"
 	"github.com/GoCollaborate/artifacts/message"
+	"github.com/GoCollaborate/artifacts/service"
 	"github.com/GoCollaborate/artifacts/stats"
 	"github.com/GoCollaborate/artifacts/task"
 	"github.com/GoCollaborate/cmd"
@@ -17,6 +18,7 @@ import (
 	"github.com/GoCollaborate/web"
 	"github.com/GoCollaborate/wrappers/cardHelper"
 	"github.com/GoCollaborate/wrappers/messageHelper"
+	"github.com/GoCollaborate/wrappers/serviceHelper"
 	"github.com/gorilla/mux"
 	"net"
 	"net/http"
@@ -93,13 +95,23 @@ func (clbt *Collaborator) Join(wk iworkable.Workable) {
 	go func() {
 		for {
 			clbt.Catchup()
-			<-time.After(constants.DefaultSynInterval)
+			<-time.After(constants.DefaultSyncInterval)
 			// clean collaborators that are no longer alive
 			clbt.Clean()
 			// dump data to local store
 			clbt.CardCase.writeStream()
 		}
 	}()
+
+	go func() {
+		// service registry
+		clbt.RegisterService()
+		// todo: service mornitoring
+		// for {
+		// 	<-time.After(constants.DefaultHeartbeatInterval)
+		// 	clbt.HeartBeat()
+		// }
+	}()
 }
 
 // Catchup with peer servers.
@@ -170,6 +182,69 @@ func (clbt *Collaborator) Clean() {
 	cardHelper.RangePrint(cards)
 }
 
+func (clbt *Collaborator) RegisterService() {
+	var (
+		cdntIP = clbt.CardCase.Reserved.Coordinator.GetFullIP()
+		clbtIP = clbt.CardCase.Local.GetFullIP()
+	)
+
+	resp, err := http.Get("http://" + cdntIP + "/cluster/" + clbt.CardCase.CaseID + "/services")
+	if err != nil {
+		return
+	}
+
+	// if cluster already get created, return to call general service registry api on coordinator
+	if resp.StatusCode >= 200 && resp.StatusCode <= 299 {
+		return
+	}
+
+	var (
+		router   = store.GetRouter()
+		services = map[string]*service.Service{}
+	)
+
+	router.Walk(func(route *mux.Route, router *mux.Router, ancestors []*mux.Route) error {
+		t, err := route.GetPathTemplate()
+		if err != nil {
+			return err
+		}
+		svr := service.NewService()
+		svr.Description = route.GetName()
+		svr.RegList = append(svr.RegList, card.Card{
+			IP:    clbt.CardCase.Local.IP,
+			Port:  clbt.CardCase.Local.Port,
+			API:   t,
+			Alive: true,
+		})
+		services[clbtIP+"/"+t] = svr
+		return nil
+	})
+
+	reader, err := serviceHelper.MarshalServiceResourceToByteStreamReader(services)
+	if err != nil {
+		panic(constants.ErrCoordinatorNotFound)
+		return
+	}
+
+	// walk through services and get them created on the coordinator
+	resp2, err := http.Post("http://"+cdntIP+"/services", "application/json", reader)
+	if err != nil {
+		panic(err)
+	}
+
+	_, err = http.Post("http://"+cdntIP+"/cluster/"+clbt.CardCase.CaseID+"/services", "application/json", resp2.Body)
+	if err != nil {
+		panic(err)
+	}
+	return
+}
+
+func (clbt *Collaborator) HeartBeat() {
+	// cdntIP := clbt.CardCase.Reserved.Coordinator.GetFullIP()
+	// // todo
+	// http.Get(cdntIP + "")
+}
+
 // Start handling server routes.
 func (clbt *Collaborator) Handle(router *mux.Router) *mux.Router {
 

constants/constants.go

@@ -43,6 +43,7 @@ const (
 	DefaultHashLength = 12
 )
 
+// time consts
 var (
 	DefaultReadTimeout              = 15 * time.Second
 	DefaultPeriodShort              = 500 * time.Millisecond
@@ -54,7 +55,8 @@ var (
 	DefaultTaskExpireTime           = 30 * time.Second
 	DefaultGCInterval               = 30 * time.Second
 	DefaultMaxMappingTime           = 600 * time.Second
-	DefaultSynInterval              = 3 * time.Minute
+	DefaultSyncInterval             = 3 * time.Minute
+	DefaultHeartbeatInterval        = 1 * time.Minute
 	DefaultJobRequestRefillInterval = 1 * time.Millisecond
 	DefaultStatFlushInterval        = 20 * time.Millisecond
 	DefaultStatAbstractInterval     = 3 * time.Second
@@ -130,27 +132,32 @@ var (
 	ErrMessageChannelDirty             = errors.New("GoCollaborate: message channel has unconsumed message error")
 	ErrTaskChannelDirty                = errors.New("GoCollaborate: task channel has unconsumed task error")
 	ErrStatTypeNotFound                = errors.New("GoCollaborate: stat type not found error")
+	ErrCoordinatorNotFound             = errors.New("GoCollaborate: coordinator not found error")
 )
 
 type Header struct {
 	Key   string `json:"key"`
 	Value string `json:"value"`
 }
 
+// HTTP Status
+// var (
+// Header200OK                   = Header{"200", "OK"}
+// Header201Created              = Header{"201", "Created"}
+// Header202Accepted             = Header{"202", "Accepted"}
+// Header204NoContent            = Header{"204", "NoContent"}
+// Header403Forbidden            = Header{"403", "Forbidden"}
+// Header404NotFound             = Header{"404", "NotFound"}
+// Header409Conflict             = Header{"409", "Conflict"}
+// Header415UnsupportedMediaType = Header{"415", "UnsupportedMediaType"}
+// Header422ExceedLimit          = Header{"422", "ExceedLimit"}
+// )
+
 // HTTP headers
 var (
-	Header200OK                   = Header{"200", "OK"}
-	Header201Created              = Header{"201", "Created"}
-	Header202Accepted             = Header{"202", "Accepted"}
-	Header204NoContent            = Header{"204", "NoContent"}
-	Header403Forbidden            = Header{"403", "Forbidden"}
-	Header404NotFound             = Header{"404", "NotFound"}
-	Header409Conflict             = Header{"409", "Conflict"}
-	Header415UnsupportedMediaType = Header{"415", "UnsupportedMediaType"}
-	Header422ExceedLimit          = Header{"422", "ExceedLimit"}
-	HeaderContentTypeJSON         = Header{"Content-Type", "application/json"}
-	HeaderContentTypeText         = Header{"Content-Type", "text/html"}
-	HeaderCORSEnableAllOrigin     = Header{"Access-Control-Allow-Origin", "*"}
+	HeaderContentTypeJSON     = Header{"Content-Type", "application/json"}
+	HeaderContentTypeText     = Header{"Content-Type", "text/html"}
+	HeaderCORSEnableAllOrigin = Header{"Access-Control-Allow-Origin", "*"}
 )
 
 // Gossip Protocol headers