pkg/log/hooks/client.go

package hooks

import (
	"context"
	"encoding/json"
	"errors"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/sirupsen/logrus"
	"github.com/wpajqz/brpc/export"
)

const (
	timeout       = 5 * time.Second
	retryInterval = 100 * time.Microsecond
)

const (
	notifyPath = "/v1/notify"
)

var (
	defaultClient *Client
	once          = &sync.Once{}
)

type Client struct{ tcpClient *export.Client }

func (c *Client) Notify(req *NotifyRequest) (*NotifyResponse, error) {
	if err := req.Check(); err != nil {
		return nil, err
	}
	param, err := json.Marshal(req)
	if err != nil {
		return nil, err
	}

	var resp NotifyResponse

	err = c.tcpClient.SyncSend(notifyPath, param, &RequestStatusCallback{
		Success: func(header, body []byte) {
			err = json.Unmarshal(body, &resp)
		},
		Error: func(code int, message string) {
			err = errors.New(message)
		},
	})
	if err != nil {
		return nil, err
	}

	return &resp, err
}

func NewClient(server string, port int) *Client {
	once.Do(func() {
		address := strings.Join([]string{server, strconv.Itoa(port)}, ":")
		c, err := export.NewClient(address, &ReadyStateCallback{})
		if err != nil {
			panic("connect link faild")
		}

		defaultClient = &Client{tcpClient: c}

		ctx, cancel := context.WithTimeout(context.Background(), timeout)
		defer cancel()

		for {
			select {
			case <-time.After(retryInterval):
				if defaultClient.tcpClient.GetReadyState() == export.OPEN {
					return
				}
			case <-ctx.Done():
				panic("connect link timeout ...")
			}
		}
	})

	return defaultClient
}

func GetClient() *Client {
	return defaultClient
}

type (
	NotifyRequest struct {
		Project     string   `json:"project"`
		Maintainers []string `json:"maintainers"`
		Message     string   `json:"message"`
		Time        string   `json:"time"`
	}
	NotifyResponse struct {
	}
)

func (ar *NotifyRequest) Check() error {
	return nil
}

type (
	ReadyStateCallback    struct{}
	RequestStatusCallback struct {
		Start   func()
		End     func()
		Success func(header, body []byte)
		Error   func(code int, message string)
	}
)

func (readyStateCallback *ReadyStateCallback) OnOpen() {
	logrus.Info("open link socket connection")
}

func (readyStateCallback *ReadyStateCallback) OnClose() {
	logrus.Info("close link socket connection")
}

func (readyStateCallback *ReadyStateCallback) OnError(err string) {
	logrus.Infof("error:%s", err)
}

func (r RequestStatusCallback) OnStart() {
	if r.Start != nil {
		r.Start()
	}
}

func (r RequestStatusCallback) OnSuccess(header, body []byte) {
	if r.Success != nil {
		r.Success(header, body)
	}
}

func (r RequestStatusCallback) OnError(code int, message string) {
	if r.Error != nil {
		r.Error(code, message)
	}
}

func (r RequestStatusCallback) OnEnd() {
	if r.End != nil {
		r.End()
	}
}