micro/registry/discovery.go

package registry

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"git.nspix.com/golang/micro/helper/httpclient"
	"git.nspix.com/golang/micro/helper/utils"
	"io"
	"io/ioutil"
	"net/http"
	"time"
)

const (
	DefaultDiscoveryUrl = "https://discovery.nspix.com"
)

type (
	Discovery struct {
		baseUrl string
		Timeout time.Duration
	}

	discoveryResponse struct {
		Code    int             `json:"errno"`
		Message string          `json:"errmsg"`
		Data    json.RawMessage `json:"result"`
	}
)

func (r *Discovery) sendRequest(ctx context.Context, method, path string, body io.Reader, data interface{}) (err error) {
	var (
		req *http.Request
		res *http.Response
	)
	if req, err = http.NewRequest(method, r.baseUrl+path, body); err != nil {
		return
	}
	if body != nil {
		req.Header.Set("Content-Type", "application/json")
	}
	if res, err = httpclient.Do(ctx, req); err != nil {
		return
	}
	defer func() {
		_ = res.Body.Close()
	}()
	rv := &discoveryResponse{}
	if err = json.NewDecoder(res.Body).Decode(rv); err == nil {
		if rv.Code == 0 {
			if data != nil {
				err = json.Unmarshal(rv.Data, data)
			}
		} else {
			err = errors.New(rv.Message)
		}
	}
	return
}

func (r *Discovery) Register(ctx context.Context, instance *ServiceNode) (err error) {
	var (
		buf []byte
	)
	if buf, err = json.Marshal(instance); err != nil {
		return
	}
	if ctx == nil{
		ctx = context.Background()
	}
	cc, cancelFunc := context.WithTimeout(ctx, r.Timeout)
	defer func() {
		cancelFunc()
	}()
	return r.sendRequest(cc, "POST", "/register", bytes.NewReader(buf), nil)
}

func (r *Discovery) Deregister(ctx context.Context, instance *ServiceNode) (err error) {
	cc, cancelFunc := context.WithTimeout(ctx, r.Timeout)
	if ctx == nil{
		ctx = context.Background()
	}
	defer func() {
		cancelFunc()
	}()
	return r.sendRequest(cc, "DELETE", "/deregister/"+instance.ID, nil, nil)
}

func (r *Discovery) Get(ctx context.Context, name string) (instances []*ServiceNode, err error) {
	instances = make([]*ServiceNode, 0)
	if ctx == nil{
		ctx = context.Background()
	}
	cc, cancelFunc := context.WithTimeout(ctx, r.Timeout)
	defer func() {
		cancelFunc()
	}()
	err = r.sendRequest(cc, "GET", "/service?name="+name, nil, &instances)
	return
}

func (r *Discovery) Fetch(ctx context.Context) (instances []*ServiceNode, err error) {
	instances = make([]*ServiceNode, 0)
	if ctx == nil{
		ctx = context.Background()
	}
	cc, cancelFunc := context.WithTimeout(ctx, r.Timeout)
	defer func() {
		cancelFunc()
	}()
	err = r.sendRequest(cc, "GET", "/services", nil, &instances)
	return
}

func NewDiscovery(uri string) *Discovery {
	if uri == "" {
		//兼容k8s的情况
		namespaceFile := "/var/run/secrets/kubernetes.io/serviceaccount/namespace"
		if utils.FileExists(namespaceFile) {
			if buf, err := ioutil.ReadFile(namespaceFile); err == nil {
				uri = fmt.Sprintf("http://discovery.%s.svc.cluster.local", string(buf))
			}
		}
	}
	if uri == "" {
		uri = DefaultDiscoveryUrl
	}
	return &Discovery{
		baseUrl: uri,
		Timeout: time.Second * 10,
	}
}