adjust code

Author: Geapefurit

api/stream/connection.go

@@ -1,13 +1,15 @@
 package stream
 
 import (
+	"github.com/NpoolPlatform/go-service-framework/pkg/logger"
 	"github.com/NpoolPlatform/message/npool/sphinxproxy"
 	"github.com/NpoolPlatform/sphinx-proxy/pkg/connection"
 )
 
 func (s *Server) ProxyConnnection(stream sphinxproxy.SphinxProxyStream_ProxyConnnectionServer) error {
 	conn, err := connection.RegisterConnection(stream)
 	if err != nil {
+		logger.Sugar().Error(err)
 		return err
 	}
 	defer conn.Close()

pkg/connection/connection.go pkg/connection/connectionmgr.gopkg/connection/connection.go renamed to pkg/connection/connectionmgr.go

@@ -119,17 +119,17 @@ type ConnectionMGR struct {
 	connections []*Connection
 }
 
-var mgr *ConnectionMGR
+var cmgr *ConnectionMGR
 
 func GetConnectionMGR() *ConnectionMGR {
-	if mgr == nil {
-		mgr = &ConnectionMGR{
+	if cmgr == nil {
+		cmgr = &ConnectionMGR{
 			siderInfos:  make(map[string]*sphinxproxy.SiderInfo),
 			connInfos:   make(map[string]map[sphinxproxy.SiderType][]*Connection),
 			recvChannel: sync.Map{},
 		}
 	}
-	return mgr
+	return cmgr
 }
 
 func (mgr *ConnectionMGR) AddConnection(conn *Connection) {
@@ -321,8 +321,15 @@ func (mgr *ConnectionMGR) DealDataElement(data *sphinxproxy.DataElement) {
 		}:
 		}
 	}
-	logger.Sugar().Warn(data)
 
-	err := mgr.ReplyMsg(data, &MsgInfo{Payload: []byte(fmt.Sprintf("proxy reply for :%v", string(data.Payload)))})
-	logger.Sugar().Error(err)
+	handler, err := GetDEHandlerMGR().GetDEHandler(data.MsgType)
+	if err != nil {
+		logger.Sugar().Error(err)
+		return
+	}
+
+	err = handler(data)
+	if err != nil {
+		logger.Sugar().Error(err)
+	}
 }

pkg/handler/handlermgr.go pkg/connection/handlermgr.gopkg/handler/handlermgr.go renamed to pkg/connection/handlermgr.go

@@ -1,12 +1,11 @@
-package handler
+package connection
 
 import (
 	"context"
 	"encoding/json"
 	"fmt"
 
 	"github.com/NpoolPlatform/message/npool/sphinxproxy"
-	"github.com/NpoolPlatform/sphinx-proxy/pkg/connection"
 	"github.com/NpoolPlatform/sphinx-proxy/pkg/utils"
 )
 
@@ -21,15 +20,15 @@ type DEHandlerMGR struct {
 	DEHandlers map[sphinxproxy.MsgType]*DEHandler
 }
 
-var mgr *DEHandlerMGR
+var hmgr *DEHandlerMGR
 
 func GetDEHandlerMGR() *DEHandlerMGR {
-	if mgr == nil {
-		mgr = &DEHandlerMGR{
+	if hmgr == nil {
+		hmgr = &DEHandlerMGR{
 			DEHandlers: make(map[sphinxproxy.MsgType]*DEHandler),
 		}
 	}
-	return mgr
+	return hmgr
 }
 
 func (mgr *DEHandlerMGR) RegisterDEHandler(
@@ -67,9 +66,9 @@ func (mgr *DEHandlerMGR) RegisterDEHandler(
 				statusMsg = err.Error()
 			}
 
-			return connection.GetConnectionMGR().ReplyMsg(
+			return GetConnectionMGR().ReplyMsg(
 				data,
-				&connection.MsgInfo{
+				&MsgInfo{
 					Payload:    outPayload,
 					StatusCode: statusCode,
 					StatusMsg:  &statusMsg,

pkg/handler/handlermgr_test.go pkg/connection/handlermgr_test.gopkg/handler/handlermgr_test.go renamed to pkg/connection/handlermgr_test.go

@@ -1,4 +1,4 @@
-package handler
+package connection
 
 import (
 	"context"

pkg/handler/handler.go pkg/connection/register.gopkg/handler/handler.go renamed to pkg/connection/register.go

@@ -1,4 +1,4 @@
-package handler
+package connection
 
 import (
 	"context"