addjust code

Author: Geapefurit

pkg/connection/connectionmgr.go

@@ -1,117 +1,15 @@
 package connection
 
 import (
-	"context"
-	"encoding/json"
 	"fmt"
 	"sync"
 	"time"
 
 	"github.com/NpoolPlatform/go-service-framework/pkg/logger"
-	"github.com/NpoolPlatform/go-service-framework/pkg/wlog"
 	"github.com/NpoolPlatform/message/npool/sphinxproxy"
 	"github.com/google/uuid"
 )
 
-type Connection struct {
-	sphinxproxy.SphinxProxyStream_ProxyConnnectionServer
-	*sphinxproxy.ConnectionInfo
-	ctx          context.Context
-	cancel       context.CancelFunc
-	onCloseFuncs []func(conn *Connection)
-	recvHandlers []func(data *sphinxproxy.DataElement)
-	closeOnce    sync.Once
-}
-
-func RegisterConnection(stream sphinxproxy.SphinxProxyStream_ProxyConnnectionServer) (*Connection, error) {
-	select {
-	case <-time.NewTicker(time.Second * 3).C:
-		return nil, wlog.Errorf("timeout for register connection")
-	default:
-		data, err := stream.Recv()
-		if err != nil {
-			return nil, wlog.WrapError(err)
-		}
-
-		statusCode := sphinxproxy.StatusCode_StatusCodeSuccess
-		statusMsg := ""
-
-		connInfo := &sphinxproxy.ConnectionInfo{}
-		err = json.Unmarshal(data.Payload, connInfo)
-		if err != nil {
-			statusCode = sphinxproxy.StatusCode_StatusCodeFailed
-			statusMsg = err.Error()
-		}
-
-		if statusCode == sphinxproxy.StatusCode_StatusCodeSuccess && data.ConnectID != connInfo.ID {
-			statusCode = sphinxproxy.StatusCode_StatusCodeFailed
-			statusMsg = err.Error()
-		}
-
-		err = stream.Send(&sphinxproxy.DataElement{
-			ConnectID:  data.ConnectID,
-			MsgID:      data.MsgID,
-			StatusCode: statusCode,
-			StatusMsg:  &statusMsg,
-		})
-		if err != nil {
-			return nil, wlog.WrapError(err)
-		}
-
-		if statusCode != sphinxproxy.StatusCode_StatusCodeSuccess {
-			return nil, wlog.Errorf(statusMsg)
-		}
-
-		ctx, cancel := context.WithCancel(stream.Context())
-		return &Connection{
-			SphinxProxyStream_ProxyConnnectionServer: stream,
-			ConnectionInfo:                           connInfo,
-			ctx:                                      ctx,
-			cancel:                                   cancel,
-		}, nil
-	}
-}
-
-func (conn *Connection) WatchClose(onClose func(conn *Connection)) {
-	conn.onCloseFuncs = append(conn.onCloseFuncs, onClose)
-}
-
-func (conn *Connection) WatchRecv(onRecv func(data *sphinxproxy.DataElement)) {
-	conn.recvHandlers = append(conn.recvHandlers, onRecv)
-}
-
-func (conn *Connection) Close() {
-	conn.closeOnce.Do(func() {
-		for _, onClose := range conn.onCloseFuncs {
-			onClose(conn)
-		}
-		conn.cancel()
-		logger.Sugar().Warnf(
-			"connection is closed, siderType: %v, ID: %v, Position: %v",
-			conn.SiderType,
-			conn.Position,
-			conn.Position)
-	})
-}
-
-func (conn *Connection) OnRecv() {
-	go func() {
-		defer conn.Close()
-		for {
-			data, err := conn.Recv()
-			if err != nil {
-				logger.Sugar().Error(err)
-				return
-			}
-			for _, recvHandler := range conn.recvHandlers {
-				go recvHandler(data)
-			}
-		}
-	}()
-
-	<-conn.ctx.Done()
-}
-
 type ConnectionMGR struct {
 	siderInfos  map[string]*sphinxproxy.SiderInfo
 	connInfos   map[string]map[sphinxproxy.SiderType][]*Connection