将连接模块与客户端，服务端进一步解耦

Author: Cai-ki

ciface/iserver.go

@@ -12,6 +12,6 @@ type IServer interface {
 
 	SetOnConnStart(func(IConn))
 	SetOnConnStop(func(IConn))
-	CallOnConnStart(conn IConn)
-	CallOnConnStop(conn IConn)
+	GetOnConnStart() func(IConn)
+	GetOnConnStop() func(IConn)
 }

cnet/client.go

@@ -7,7 +7,6 @@ import (
 	"github.com/Cai-ki/cinx/ciface"
 )
 
-// 客户端接口实现
 type Client struct {
 	Name      string
 	IPVersion string
@@ -31,7 +30,9 @@ func NewClient(name, ipVersion, ip string, port int) ciface.IClient {
 		IP:        ip,
 		Port:      port,
 
-		msgHandler: NewMsgHandle(),
+		msgHandler:  NewMsgHandle(),
+		onConnStart: func(conn ciface.IConn) {},
+		onConnStop:  func(conn ciface.IConn) {},
 	}
 	return c
 }
@@ -70,9 +71,8 @@ func (c *Client) Start() {
 }
 func (c *Client) Stop() {
 	con := c.Conn()
-	if con != nil {
-		con.Stop()
-	}
+	con.Stop()
+	c.exitChan <- struct{}{}
 }
 func (c *Client) Conn() ciface.IConn {
 	return c.conn

cnet/conn.go

@@ -13,20 +13,16 @@ import (
 )
 
 type Connection struct {
-	//当前Conn属于哪个Server
-	TcpServer ciface.IServer //当前conn属于哪个server，在conn初始化的时候添加即可
-	//当前连接的socket TCP套接字
+	TcpServer ciface.IServer
+
 	Conn *net.TCPConn
-	//当前连接的ID 也可以称作为SessionID，ID全局唯一
+
 	ConnID uint32
 
-	// 正常退出
 	IsClosed atomic.Bool
 
-	// 异常退出
 	IsAborted atomic.Bool
 
-	//消息管理MsgId和对应处理方法的消息管理模块
 	MsgHandler ciface.IMsgHandle
 
 	ExitBuffChan chan struct{}
@@ -36,43 +32,47 @@ type Connection struct {
 
 	property     map[string]interface{}
 	propertyLock sync.RWMutex
+
+	onConnStart func(conn ciface.IConn)
+	onConnStop  func(conn ciface.IConn)
 }
 
-// 创建连接的方法
 func NewConntion(server ciface.IServer, conn *net.TCPConn, connID uint32, msgHandler ciface.IMsgHandle) *Connection {
 	c := &Connection{
-		TcpServer:    server, //将隶属的server传递进来
+		TcpServer:    server,
 		Conn:         conn,
 		ConnID:       connID,
 		MsgHandler:   msgHandler,
 		ExitBuffChan: make(chan struct{}, 1),
-		msgChan:      make(chan []byte), //msgChan初始化
+		msgChan:      make(chan []byte),
 		msgBuffChan:  make(chan []byte, cutils.GlobalObject.MaxMsgChanLen),
-		property:     make(map[string]interface{}), //对链接属性map初始化
+		property:     make(map[string]interface{}),
+		onConnStart:  server.GetOnConnStart(),
+		onConnStop:   server.GetOnConnStop(),
 	}
 	c.IsClosed.Store(false)
 	c.IsAborted.Store(false)
 
-	c.TcpServer.GetConnMgr().Add(c)
+	server.GetConnMgr().Add(c)
 	return c
 }
 
 func NewClientConn(client ciface.IClient, conn *net.TCPConn) ciface.IConn {
 	c := &Connection{
-		TcpServer:    NewServer(), // TODO: 临时创建一个server，后续需要修改
+		TcpServer:    NewServer(),
 		Conn:         conn,
-		ConnID:       0, // ignore
+		ConnID:       0,
 		MsgHandler:   client.GetMsgHandler(),
 		ExitBuffChan: make(chan struct{}, 1),
-		msgChan:      make(chan []byte), //msgChan初始化
+		msgChan:      make(chan []byte),
 		msgBuffChan:  make(chan []byte, cutils.GlobalObject.MaxMsgChanLen),
-		property:     make(map[string]interface{}), //对链接属性map初始化
+		property:     make(map[string]interface{}),
+		onConnStart:  client.GetOnConnStart(),
+		onConnStop:   client.GetOnConnStop(),
 	}
 	c.IsClosed.Store(false)
 	c.IsAborted.Store(false)
 
-	c.TcpServer.SetOnConnStart(client.GetOnConnStart())
-	c.TcpServer.SetOnConnStop(client.GetOnConnStop())
 	return c
 }
 
@@ -82,26 +82,23 @@ func (c *Connection) StartReader() {
 	defer c.Stop()
 
 	for {
-		// 创建拆包解包的对象
+
 		dp := NewDataPack()
 
-		//读取客户端的Msg head
 		headData := make([]byte, dp.GetHeadLen())
 		if _, err := io.ReadFull(c.GetTCPConn(), headData); err != nil {
 			fmt.Println("read msg head error ", err)
 			c.IsAborted.Store(true)
 			break
 		}
 
-		//拆包，得到msgid 和 datalen 放在msg中
 		msg, err := dp.Unpack(headData)
 		if err != nil {
 			fmt.Println("unpack error ", err)
 			c.IsAborted.Store(true)
 			break
 		}
 
-		//根据 dataLen 读取 data，放在msg.Data中
 		var data []byte
 		if msg.GetDataLen() > 0 {
 			data = make([]byte, msg.GetDataLen())
@@ -138,7 +135,7 @@ func (c *Connection) StartWriter() {
 		select {
 		case data, ok := <-c.msgChan:
 			if ok {
-				//有数据要写给客户端
+
 				if _, err := c.Conn.Write(data); err != nil {
 					fmt.Println("Send Data error:, ", err, " Conn Writer exit")
 					return
@@ -147,10 +144,10 @@ func (c *Connection) StartWriter() {
 				fmt.Println("msgChan is Closed")
 				return
 			}
-			//针对有缓冲channel需要些的数据处理
+
 		case data, ok := <-c.msgBuffChan:
 			if ok {
-				//有数据要写给客户端
+
 				if _, err := c.Conn.Write(data); err != nil {
 					fmt.Println("Send Buff Data error:, ", err, " Conn Writer exit")
 					return
@@ -163,79 +160,66 @@ func (c *Connection) StartWriter() {
 	}
 }
 
-// 启动连接，让当前连接开始工作
 func (c *Connection) Start() {
-
-	//1 开启用户从客户端读取数据流程的Goroutine
 	go c.StartReader()
-	//2 开启用于写回客户端数据流程的Goroutine
 	go c.StartWriter()
 
-	//按照用户传递进来的创建连接时需要处理的业务，执行钩子方法
-	c.TcpServer.CallOnConnStart(c)
+	c.onConnStart(c)
 
 	for {
 		select {
 		case <-c.ExitBuffChan:
-			// 得到退出消息，不再阻塞
+
 			return
 		}
 	}
 }
 
-// 停止连接，结束当前连接状态
 func (c *Connection) Stop() {
 	if c.IsClosed.Load() {
 		return
 	}
 	c.IsClosed.Store(true)
 
-	c.TcpServer.CallOnConnStop(c)
+	c.onConnStop(c)
 
-	// 关闭socket链接
 	c.Conn.Close()
 
 	c.ExitBuffChan <- struct{}{}
 
-	// 将链接从连接管理器中删除
 	c.TcpServer.GetConnMgr().Remove(c)
 
 	close(c.ExitBuffChan)
 	close(c.msgBuffChan)
 	close(c.msgChan)
 }
 
-// 从当前连接获取原始的socket TCPConn
 func (c *Connection) GetTCPConn() *net.TCPConn {
 	return c.Conn
 }
 
-// 获取当前连接ID
 func (c *Connection) GetConnID() uint32 {
 	return c.ConnID
 }
 
-// 获取远程客户端地址信息
 func (c *Connection) RemoteAddr() net.Addr {
 	return c.Conn.RemoteAddr()
 }
 
-// 直接将Message数据发送数据给远程的TCP客户端
 func (c *Connection) SendMsg(msgId uint32, data []byte) error {
 	IsClosed := c.IsClosed.Load()
 	if IsClosed {
 		return errors.New("Connection closed when send msg")
 	}
-	//将data封包，并且发送
+
 	dp := NewDataPack()
 	msg, err := dp.Pack(NewMsgPackage(msgId, data))
 	if err != nil {
 		fmt.Println("Pack error msg id = ", msgId)
 		return errors.New("Pack error msg ")
 	}
 
-	//写回客户端
-	c.msgChan <- msg //将之前直接回写给conn.Write的方法 改为 发送给Channel 供Writer读取
+	c.msgChan <- msg
 
 	return nil
 }
@@ -245,29 +229,26 @@ func (c *Connection) SendBuffMsg(msgId uint32, data []byte) error {
 	if IsClosed {
 		return errors.New("Connection closed when send buff msg")
 	}
-	//将data封包，并且发送
+
 	dp := NewDataPack()
 	msg, err := dp.Pack(NewMsgPackage(msgId, data))
 	if err != nil {
 		fmt.Println("Pack error msg id = ", msgId)
 		return errors.New("Pack error msg ")
 	}
 
-	//写回客户端
 	c.msgBuffChan <- msg
 
 	return nil
 }
 
-// 设置链接属性
 func (c *Connection) SetProperty(key string, value interface{}) {
 	c.propertyLock.Lock()
 	defer c.propertyLock.Unlock()
 
 	c.property[key] = value
 }
 
-// 获取链接属性
 func (c *Connection) GetProperty(key string) (interface{}, bool) {
 	c.propertyLock.RLock()
 	defer c.propertyLock.RUnlock()
@@ -279,7 +260,6 @@ func (c *Connection) GetProperty(key string) (interface{}, bool) {
 	}
 }
 
-// 移除链接属性
 func (c *Connection) RemoveProperty(key string) {
 	c.propertyLock.Lock()
 	defer c.propertyLock.Unlock()