GoCollaborate ver:0.3.2

Author: HastingsYoung

CHANGELOG.md

@@ -4,6 +4,8 @@
 ### 0.3.x
 #### 0.3.2
 - Add versioning for coordinator
+- Support more load-balancing algorithms
+- Optimize error responses
 #### 0.3.1
 - Add "methods" field to service response
 - Change rate limiting response header: 422 -> 429 Too Many Requests

artifacts/card/card.go

@@ -14,6 +14,13 @@ type Card struct {
 	Seed  bool   `json:"seed,omitempty"`
 }
 
+func (c *Card) IsInitialized() bool {
+	if len(c.IP) > 0 {
+		return true
+	}
+	return false
+}
+
 func (c *Card) GetFullIP() string {
 	return c.IP + ":" + strconv.Itoa(c.Port)
 }

artifacts/restful/error.go

@@ -1,34 +1,73 @@
 package restful
 
 type ErrorPayload struct {
-	JSONAPI string  `json:"jsonapi, omitempty"`
-	Errors  []Error `json:"errors"`
+	Errors []Error `json:"errors"`
 }
 
 type Error struct {
-	Code   string      `json:"code"`
-	Source ErrorSource `json:"source"`
-	Title  string      `json:"title"`
-	Detail string      `json:"detail"`
+	Status int    `json:"status"`
+	Title  string `json:"title"`
+	Detail string `json:"detail"`
+	Source string `json:"source"`
 }
 
-type ErrorSource struct {
-	Pointer string `json:"pointer"`
+// create error types
+func Error401Unauthorized() ErrorPayload {
+	return ErrorPayload{[]Error{
+		Error{401, "Unauthorized", "Although the HTTP standard specifies 'unauthorized', semantically this response means 'unauthenticated'.", ""}},
+	}
 }
 
-// create specific Error Types Below
-func Error415UnsupportedMediaType() ErrorPayload {
-	return ErrorPayload{"add_your_custom_api_route_here", []Error{Error{"403", ErrorSource{"add_your_custom_error_source_here"}, "add_your_custom_error_title_here", "add_your_custom_error_detail_here"}}}
+func Error403Forbidden() ErrorPayload {
+	return ErrorPayload{[]Error{
+		Error{403, "Forbidden", "The client does not have access rights to the content.", ""}},
+	}
+}
+
+func Error404NotFound() ErrorPayload {
+	return ErrorPayload{[]Error{
+		Error{404, "Not Found", "The server can not find requested resource.", ""}},
+	}
+}
+
+func Error405MethodNotAllowed() ErrorPayload {
+	return ErrorPayload{[]Error{
+		Error{404, "Method Not Allowed", "The request method is known by the server but has been disabled and cannot be used.", ""}},
+	}
+}
+
+func Error408RequestTimeout() ErrorPayload {
+	return ErrorPayload{[]Error{
+		Error{408, "Method Not Allowed", "This response is sent on an idle connection by some servers, even without any previous request by the client.", ""}},
+	}
 }
 
 func Error409Conflict() ErrorPayload {
-	return ErrorPayload{"add_your_custom_api_route_here", []Error{Error{"409", ErrorSource{"add_your_custom_error_source_here"}, "add_your_custom_error_title_here", "add_your_custom_error_detail_here"}}}
+	return ErrorPayload{[]Error{
+		Error{409, "Conflict", "This response is sent when a request conflicts with the current state of the server.", ""}},
+	}
 }
 
-func Error404NotFound() ErrorPayload {
-	return ErrorPayload{"add_your_custom_api_route_here", []Error{Error{"404", ErrorSource{"add_your_custom_error_source_here"}, "add_your_custom_error_title_here", "add_your_custom_error_detail_here"}}}
+func Error415UnsupportedMediaType() ErrorPayload {
+	return ErrorPayload{[]Error{
+		Error{415, "Bad Request", "This response means that server could not understand the request due to invalid syntax.", ""}},
+	}
 }
 
-func Error403Forbidden() ErrorPayload {
-	return ErrorPayload{"add_your_custom_api_route_here", []Error{Error{"403", ErrorSource{"add_your_custom_error_source_here"}, "add_your_custom_error_title_here", "add_your_custom_error_detail_here"}}}
+func Error500InternalServerError() ErrorPayload {
+	return ErrorPayload{[]Error{
+		Error{500, "Internal Server Error", "The server has encountered a situation it doesn't know how to handle.", ""}},
+	}
+}
+
+func Error502BadGateway() ErrorPayload {
+	return ErrorPayload{[]Error{
+		Error{502, "Bad Gateway", "This error response means that the server, while working as a gateway to get a response needed to handle the request, got an invalid response.", ""}},
+	}
+}
+
+func Error503ServiceUnavailable() ErrorPayload {
+	return ErrorPayload{[]Error{
+		Error{503, "Service Unavailable", "The server is not ready to handle the request.", ""}},
+	}
 }

artifacts/service/service.go

@@ -4,6 +4,7 @@ import (
 	"github.com/GoCollaborate/src/artifacts/card"
 	"github.com/GoCollaborate/src/artifacts/parameter"
 	"github.com/GoCollaborate/src/constants"
+	"hash/fnv"
 	"math/rand"
 	"sync"
 	"time"
@@ -16,45 +17,52 @@ type Service struct {
 	Parameters  []parameter.Parameter `json:"parameters"`
 	RegList     []card.Card           `json:"registers"`
 	// a map of last heartbeat timestamps, eact key corresponds to one particular card full endpoint
-	Heartbeats       map[string]int64 `json:"heartbeats,omitempty"`
-	SbscrbList       []string         `json:"subscribers"`  // subscriber tokens
-	Dependencies     []string         `json:"dependencies"` // dependent ServiceIDs
-	Mode             Mode             `json:"mode,omitempty"`
-	LoadBalanceMode  Mode             `json:"load_balance_mode,omitemtpy"`
-	Version          string           `json:"version"`
-	PlatformVersion  string           `json:"platform_version"`
-	LastAssignedTo   card.Card        `json:"last_assigned_to,omitempty"`
-	LastAssignedTime int64            `json:"last_assigned_time,omitempty"`
-	serviceLock      sync.Mutex       `json:"-"`
+	Heartbeats        map[string]int64 `json:"heartbeats,omitempty"`
+	SbscrbList        []string         `json:"subscribers"`  // subscriber tokens
+	Dependencies      []string         `json:"dependencies"` // dependent ServiceIDs
+	Mode              Mode             `json:"mode,omitempty"`
+	LoadBalanceMode   Mode             `json:"load_balance_mode,omitemtpy"`
+	Version           string           `json:"version"`
+	PlatformVersion   string           `json:"platform_version"`
+	LastAssignedTo    card.Card        `json:"last_assigned_to,omitempty"`
+	LastAssignedTime  int64            `json:"last_assigned_time,omitempty"`
+	LastAssignedIndex uint32           `json:"-"`
+	serviceLock       sync.Mutex       `json:"-"`
 }
 
 func NewService() *Service {
 	return &Service{
-		Description:     "",
-		Methods:         []string{},
-		Parameters:      []parameter.Parameter{},
-		RegList:         []card.Card{},
-		Heartbeats:      map[string]int64{},
-		SbscrbList:      []string{},
-		Dependencies:    []string{},
-		Version:         "1.0",
-		PlatformVersion: "golang1.8.1",
-		serviceLock:     sync.Mutex{},
+		Description:      "",
+		Methods:          []string{},
+		Parameters:       []parameter.Parameter{},
+		RegList:          []card.Card{},
+		Heartbeats:       map[string]int64{},
+		SbscrbList:       []string{},
+		Dependencies:     []string{},
+		Mode:             ClbtModeNormal,
+		LoadBalanceMode:  LBModeRandom,
+		Version:          "1.0",
+		PlatformVersion:  "golang1.8.1",
+		LastAssignedTime: int64(0),
+		serviceLock:      sync.Mutex{},
 	}
 }
 
 func NewServiceFrom(from *Service) *Service {
 	return &Service{
-		Description:     from.Description,
-		Methods:         from.Methods,
-		Parameters:      from.Parameters,
-		RegList:         from.RegList,
-		Heartbeats:      map[string]int64{},
-		SbscrbList:      from.SbscrbList,
-		Dependencies:    from.Dependencies,
-		Version:         from.Version,
-		PlatformVersion: from.PlatformVersion,
-		serviceLock:     sync.Mutex{},
+		Description:      from.Description,
+		Methods:          from.Methods,
+		Parameters:       from.Parameters,
+		RegList:          from.RegList,
+		Heartbeats:       map[string]int64{},
+		SbscrbList:       from.SbscrbList,
+		Dependencies:     from.Dependencies,
+		Mode:             from.Mode,
+		LoadBalanceMode:  from.LoadBalanceMode,
+		Version:          from.Version,
+		PlatformVersion:  from.PlatformVersion,
+		LastAssignedTime: int64(0),
+		serviceLock:      sync.Mutex{},
 	}
 }
 
@@ -143,19 +151,41 @@ func (s *Service) UnSubscribe(token string) error {
 }
 
 func (s *Service) Query(token string) (*card.Card, error) {
+	if len(s.RegList) < 1 {
+		return nil, constants.ErrNoRegister
+	}
+
 	for _, x := range s.SbscrbList {
 		if x == token {
-			return s.loadBalanceRoll(), nil
+			return s.loadBalanceRoll(token)
 		}
 	}
-	return nil, constants.ErrNoRegister
+
+	return nil, constants.ErrNoSubscriber
 }
 
-func (s *Service) loadBalanceRoll() *card.Card {
+func (s *Service) loadBalanceRoll(token string) (*card.Card, error) {
+	var last_assigned_to card.Card
+	var last_assigned_index = s.LastAssignedIndex
+	var length = len(s.RegList)
+
 	switch s.LoadBalanceMode {
-	default:
-		return &s.RegList[rand.Intn(len(s.RegList))]
+	case LBModeRoundRobin:
+		last_assigned_index = (last_assigned_index + 1) % uint32(length)
+	case LBModeTokenHash:
+		h := fnv.New32a()
+		h.Write([]byte(token))
+		last_assigned_index = h.Sum32() % uint32(length)
 	}
+
+	if !s.LastAssignedTo.IsInitialized() {
+		last_assigned_index = rand.Uint32() % uint32(length)
+	}
+
+	last_assigned_to = s.RegList[last_assigned_index]
+	s.LastAssignedTo = last_assigned_to
+	s.LastAssignedIndex = last_assigned_index
+	return &last_assigned_to, nil
 }
 
 func (s *Service) UnSubscribeAll() error {

artifacts/service/serviceMode.go

@@ -8,7 +8,7 @@ const (
 	LBModeRandom               // assign tasks as per weighted probability
 	LBModeLeastActive          // assign tasks to active responders
 	LBModeRoundRobin           // assign tasks sequentially based on the order of collaborator
-	LBModeIPHash               // assign tasks based on the hash value of subscriber IP
+	LBModeTokenHash            // assign tasks based on the hash value of subscriber token
 )
 
 type Mode int

coordinator/coordinator.go

@@ -114,7 +114,6 @@ func (cdnt *Coordinator) CreateService(w http.ResponseWriter, bytes []byte) erro
 			id := utils.RandStringBytesMaskImprSrc(ServiceLength)
 			svr := service.NewServiceFrom(&dat.Attributes)
 			svr.ServiceID = id
-			svr.LastAssignedTime = int64(0)
 			cdnt.Services[id] = svr
 			dat.ID = id
 			dat.Attributes = *svr

wrappers/serviceHelper/serviceHelper.go

@@ -12,19 +12,19 @@ import (
 	"net/http"
 )
 
-func UnmarshalMode(original interface{}) service.Mode {
+func ModeInterpret(original interface{}) service.Mode {
 	if original == nil {
 		return service.ClbtModeNormal
 	}
 	var m service.Mode
 	omode := original.(string)
 	switch omode {
-	case "LBModeIPHash":
-		m = service.LBModeIPHash
 	case "ClbtModeOnlyRegister":
 		m = service.ClbtModeOnlyRegister
 	case "ClbtModeOnlySubscribe":
 		m = service.ClbtModeOnlySubscribe
+	case "LBModeTokenHash":
+		m = service.LBModeTokenHash
 	case "LBModeRandom":
 		m = service.LBModeRandom
 	case "LBModeLeastActive":