Merge branch 'develop' of https://gitee.com/fusenpack/fusenapi into develop

server/auth/internal/logic/useremailconfirmationlogic.go

@@ -2,6 +2,7 @@ package logic
 
 import (
 	"fmt"
+	"fusenapi/model/gmodel"
 	"fusenapi/utils/auth"
 	"fusenapi/utils/basic"
 	"fusenapi/utils/wevent"
@@ -34,6 +35,56 @@ func NewUserEmailConfirmationLogic(ctx context.Context, svcCtx *svc.ServiceConte
 // func (l *UserEmailConfirmationLogic) BeforeLogic(w http.ResponseWriter, r *http.Request) {
 // }
 
+func FinishRegister(svcCtx *svc.ServiceContext, user *gmodel.FsUser, token *auth.RegisterToken) error {
+	// 创建签证
+	jwtToken, err := auth.GenerateJwtTokenUint64(
+		auth.StringToHash(*user.PasswordHash),
+		svcCtx.Config.Auth.AccessExpire,
+		time.Now().UTC().Unix(),
+		user.Id,
+		0,
+	)
+
+	if err != nil {
+
+		return err
+	}
+
+	event := wevent.NewWebsocketEventSuccess(wevent.UserEmailRegister, token.TraceId)
+	event.Data = wevent.DataEmailRegister{
+		JwtToken: jwtToken,
+	}
+	err = CommonNotify(svcCtx.Config.MainAddress, token.Wid, event)
+	if err != nil {
+		// logx.Error(err, token.TraceId)
+		return err
+	}
+
+	return nil
+}
+
+func CommonNotify(MainAddress, wid string, event *wevent.WebsocketEvent) error {
+	tp := requests.Post(fmt.Sprintf("%s/api/websocket/common_notify", MainAddress))
+	tp.SetBodyJson(requests.M{
+		"wid":  wid,
+		"data": event,
+	})
+
+	wresp, err := tp.Execute()
+	if err != nil {
+		// logx.Error(err, token.TraceId)
+		return err
+	}
+
+	result := wresp.Json()
+	if result.Get("code").Int() != 200 {
+		// logx.Error(result.Get("message"))
+		return fmt.Errorf("%s", result.Get("message").Str)
+	}
+
+	return nil
+}
+
 func (l *UserEmailConfirmationLogic) UserEmailConfirmation(req *types.RequestEmailConfirmation, userinfo *auth.UserInfo) (resp *basic.Response) {
 	// 返回值必须调用Set重新返回, resp可以空指针调用 resp.SetStatus(basic.CodeOK, data)
 	// userinfo 传入值时, 一定不为null
@@ -59,43 +110,8 @@ func (l *UserEmailConfirmationLogic) UserEmailConfirmation(req *types.RequestEma
 				return resp.SetStatus(basic.CodeDbSqlErr)
 			}
 
-			// 创建签证
+			FinishRegister(l.svcCtx, user, token)
-			jwtToken, err := auth.GenerateJwtTokenUint64(
+			logx.Info("success", token.TraceId)
-				auth.StringToHash(*user.PasswordHash),
-				l.svcCtx.Config.Auth.AccessExpire,
-				time.Now().Unix(),
-				user.Id,
-				0,
-			)
-
-			if err != nil {
-				logx.Error(err, token.TraceId)
-				return
-			}
-
-			event := wevent.NewWebsocketEventSuccess(wevent.UserEmailRegister, token.TraceId)
-			event.Data = wevent.DataEmailRegister{
-				JwtToken: jwtToken,
-			}
-
-			tp := requests.Post(fmt.Sprintf("%s/api/websocket/common_notify", l.svcCtx.Config.MainAddress))
-			tp.SetBodyJson(requests.M{
-				"wid":  token.Wid,
-				"data": event,
-			})
-
-			wresp, err := tp.Execute()
-			if err != nil {
-				logx.Error(err, token.TraceId)
-				return
-			}
-
-			result := wresp.Json()
-			if result.Get("code").Int() != 200 {
-				logx.Error(result.Get("message"))
-				return
-			}
-			logx.Info("success", token.TraceId, jwtToken)
 
 		case "facebook":
 		case "fusen":
@@ -106,38 +122,8 @@ func (l *UserEmailConfirmationLogic) UserEmailConfirmation(req *types.RequestEma
 				return resp.SetStatus(basic.CodeDbSqlErr)
 			}
 
-			// 创建签证
+			FinishRegister(l.svcCtx, user, token)
-			jwtToken, err := auth.GenerateJwtTokenUint64(
+			logx.Info("success", token.TraceId)
-				auth.StringToHash(*user.PasswordHash),
-				l.svcCtx.Config.Auth.AccessExpire,
-				time.Now().Unix(),
-				user.Id,
-				0,
-			)
-
-			if err != nil {
-				logx.Error(err, token.TraceId)
-				return
-			}
-
-			event := wevent.NewWebsocketEventSuccess(wevent.UserEmailRegister, token.TraceId)
-			event.Data = wevent.DataEmailRegister{
-				JwtToken: jwtToken,
-			}
-			tp := requests.Post(fmt.Sprintf("%s/api/websocket/common_notify", l.svcCtx.Config.MainAddress))
-			tp.SetBodyJson(requests.M{
-				"wid":  token.Wid,
-				"data": event,
-			})
-			wresp, err := tp.Execute()
-			if err != nil {
-				logx.Error(err, token.TraceId)
-			}
-			result := wresp.Json()
-			if result.Get("code").Int() != 200 {
-				logx.Error(result.Get("message"))
-			}
-			logx.Info("success", token.TraceId, jwtToken)
 		}
 
 	default:

server/auth/internal/types/types.go

@@ -16,7 +16,6 @@ type DataUserLogin struct {
 
 type RequestUserRegister struct {
 	Wid         string `json:"wid"`         // websocket的id
-	GuestId     int64  `json:"guest_id"`    // 游客id
 	FirstName   string `json:"first_name"`  // 首名
 	LastName    string `json:"last_name"`   // 名
 	Resetaurant string `json:"resetaurant"` // 餐厅类型

server/websocket/internal/logic/commonnotifylogic.go

@@ -37,7 +37,7 @@ type commonConnectionNotFoundDataCacheChanItem struct {
 }
 
 // 放入缓冲队列
-func (l *CommonNotifyLogic) pushCommonCache(data commonConnectionNotFoundDataCacheChanItem) {
+func (l *CommonNotifyLogic) pushCommonNotifyCache(data commonConnectionNotFoundDataCacheChanItem) {
 	select {
 	case commonConnectionNotFoundDataCacheChan <- data:
 		return
@@ -47,45 +47,41 @@ func (l *CommonNotifyLogic) pushCommonCache(data commonConnectionNotFoundDataCac
 }
 
 // 取出元素
-func (l *CommonNotifyLogic) popCommonCache() (data commonConnectionNotFoundDataCacheChanItem) {
+func (l *CommonNotifyLogic) popCommonNotifyCache() (data commonConnectionNotFoundDataCacheChanItem) {
 	return <-commonConnectionNotFoundDataCacheChan
 }
 
 // 保证处理消息就一个循环在执行
 var consumeCommonCacheData sync.Once
 
-// 消费公共通知未处理的消息
+// 消费公共通知未处理的消息(目前是轮巡方式，待优化)
 func (l *CommonNotifyLogic) consumeCommonCacheData() {
 	//单例
 	consumeCommonCacheData.Do(func() {
-		tick := time.Tick(time.Millisecond * 200)
 		for {
-			select {
+			time.Sleep(time.Millisecond * 200)
-			case <-tick: //200毫秒触发一次
+			info := l.popCommonNotifyCache()
-				info := l.popCommonCache()
+			//查询websocket连接
-				//查询websocket连接
+			value, ok := mapConnPool.Load(info.data.Wid)
-				value, ok := mapConnPool.Load(info.data.Wid)
+			//没有连接
-				//没有连接
+			if !ok {
-				if !ok {
+				info.retryTimes--
-					info.retryTimes--
+				//大于0，则放回队列
-					//大于0，则放回队列
+				if info.retryTimes > 0 {
-					if info.retryTimes > 0 {
+					l.pushCommonNotifyCache(info)
-						l.pushCommonCache(info)
-						continue
-					}
-					//否则直接丢弃消息
 					continue
 				}
-				//断言连接
+				//否则直接丢弃消息
-				ws, ok := value.(wsConnectItem)
+				continue
-				if !ok {
-					logx.Error("渲染回调断言websocket连接失败")
-					continue
-				}
-				//发送
-				ws.sendToOutChan(ws.respondDataFormat(constants.WEBSOCKET_COMMON_NOTIFY, info.data.Data))
 			}
-
+			//断言连接
+			ws, ok := value.(wsConnectItem)
+			if !ok {
+				logx.Error("渲染回调断言websocket连接失败")
+				continue
+			}
+			//发送
+			ws.sendToOutChan(ws.respondDataFormat(constants.WEBSOCKET_COMMON_NOTIFY, info.data.Data))
 		}
 	})
 }
@@ -105,7 +101,7 @@ func (l *CommonNotifyLogic) CommonNotify(req *types.CommonNotifyReq, userinfo *a
 	value, ok := mapConnPool.Load(req.Wid)
 	if !ok {
 		//没找到连接就放到公共缓冲队列
-		go l.pushCommonCache(commonConnectionNotFoundDataCacheChanItem{
+		go l.pushCommonNotifyCache(commonConnectionNotFoundDataCacheChanItem{
 			retryTimes: 20, //重试20次
 			data: types.CommonNotifyReq{
 				Wid:  req.Wid,

server/websocket/internal/logic/ws_reuse_last_connect.go

@@ -45,6 +45,8 @@ func (w *wsConnectItem) reuseLastConnect(data []byte) {
 		}
 		//是当前自己占用(无需处理)
 		if obj.uniqueId == w.uniqueId {
+			rsp := w.respondDataFormat(constants.WEBSOCKET_CONNECT_SUCCESS, wid)
+			w.sendToOutChan(rsp)
 			return
 		} else {
 			w.reuseLastConnErrResponse("the wid is used by other people")

server_api/auth.api

@@ -52,8 +52,7 @@ type (
 type (
 	// UserAddAddressHandler 用户登录请求结构
 	RequestUserRegister {
-		Wid     string `json:"wid"`      // websocket的id
+		Wid string `json:"wid"` // websocket的id
-		GuestId int64  `json:"guest_id"` // 游客id
 
 		FirstName   string `json:"first_name"`  // 首名
 		LastName    string `json:"last_name"`   // 名

utils/curl/client_resty.go

@@ -9,6 +9,10 @@ import (
 	"github.com/go-resty/resty/v2"
 	"github.com/zeromicro/go-zero/core/logc"
 	"github.com/zeromicro/go-zero/core/logx"
+	"github.com/zeromicro/go-zero/core/trace"
+	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/propagation"
+	oteltrace "go.opentelemetry.io/otel/trace"
 )
 
 func NewClient(ctx context.Context, c *Config) Client {
@@ -18,6 +22,21 @@ func NewClient(ctx context.Context, c *Config) Client {
 	// 设置超时时间为 5 分钟
 	client.SetTimeout(5 * time.Minute)
 
+	/* 传输链路 */
+	tracer := otel.GetTracerProvider().Tracer(trace.TraceName)
+	spanCtx, span := tracer.Start(
+		ctx,
+		"client_resty_send",
+		oteltrace.WithSpanKind(oteltrace.SpanKindClient),
+	)
+	carrier := &propagation.HeaderCarrier{}
+	otel.GetTextMapPropagator().Inject(spanCtx, carrier)
+	for _, cacarrierKey := range carrier.Keys() {
+		client.SetHeader(cacarrierKey, carrier.Get(cacarrierKey))
+	}
+	defer span.End()
+	/* 传输链路 */
+
 	if c.HeaderData != nil {
 		for k, v := range c.HeaderData {
 			client = client.SetHeader(k, v)