123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135 |
- 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,
- }
- }
|