change code struct

README.md

@@ -30,12 +30,12 @@
 ## 问题与优化
 
 - 优化命令匹配的算法，当有数量级的连接时需要考虑；如使用固定 ID，自动生成匹配缓存等方式
-- 建立内存池来分配内存，减少内存碎片
 - 同地址多连接共存，使用不同的连接发送消息，减少延时，提高消息送达可靠性
 - 并发相同channel连接时出现竞争的情况
 
 ## 已经解决的问题
 
+- 建立内存池来分配内存，减少内存碎片
 - 出现断线没有自动重连的问题
 - 随机频道获取
 - 断线时命令在还没有超时的情况下，要等待重连后发送

conn/type.go

@@ -9,30 +9,42 @@ import (
 type MsgType byte
 
 const (
-	PingMsg MsgType = iota
-	RequestMsg
-	ResponseMsg
-	ProxyConnectMsg
-	ProxyResultMsg
+	MsgPing MsgType = iota
+	MsgRequest
+	MsgResponse
+	MsgAuth
+	MsgProxyConnect
+	MsgProxyResult
 )
 
 func (t MsgType) String() string {
 	switch t {
-	case PingMsg:
-		return "PingMsg"
-	case RequestMsg:
-		return "RequestMsg"
-	case ResponseMsg:
-		return "ResponseMsg"
-	case ProxyConnectMsg:
-		return "ProxyConnectMsg"
-	case ProxyResultMsg:
-		return "ProxyResultMsg"
+	case MsgPing:
+		return "Ping"
+	case MsgRequest:
+		return "Request"
+	case MsgResponse:
+		return "Response"
+	case MsgAuth:
+		return "auth"
+	case MsgProxyConnect:
+		return "ProxyConnect"
+	case MsgProxyResult:
+		return "ProxyResult"
 	default:
-		return fmt.Sprintf("Unknown MsgType (%d)", t)
+		return fmt.Sprintf("Unknown (%d)", t)
 	}
 }
 
+// 网络数据包结构
+// type MsgData struct {
+// 	Type  MsgType
+// 	Id    uint16
+// 	Cmd   string
+// 	State uint8
+// 	Data  []byte
+// }
+
 // 连接接口，代表一个原始的连接
 type Connect interface {
 	ReadAuthInfo() (proto string, version uint8, channel string, auth []byte, err error)

conn/util.go

@@ -225,7 +225,7 @@ func PackageDecode(msg []byte, recycle bool, compress bool, cf *config.Config) (
 	if id > config.ID_MAX {
 		switch id {
 		case config.ID_PROXY_CONNECT: // 代理连接请求
-			msgType = ProxyConnectMsg
+			msgType = MsgProxyConnect
 			// 请求代理的地址
 			if recycle {
 				data = slices.Clone(msg[2:])
@@ -234,7 +234,7 @@ func PackageDecode(msg []byte, recycle bool, compress bool, cf *config.Config) (
 			}
 			return
 		case config.ID_PROXY_RESULT: // 代理请求执行结果
-			msgType = ProxyResultMsg
+			msgType = MsgProxyResult
 			// 错误信息
 			if recycle {
 				data = slices.Clone(msg[2:])
@@ -243,7 +243,7 @@ func PackageDecode(msg []byte, recycle bool, compress bool, cf *config.Config) (
 			}
 			// 如果没有错误信息则表示成功
 			if len(data) <= 0 {
-				state = 1
+				state = config.STATE_OK
 			}
 			return
 		}
@@ -253,20 +253,20 @@ func PackageDecode(msg []byte, recycle bool, compress bool, cf *config.Config) (
 
 	// ping信息
 	if len(msg) == 2 {
-		msgType = PingMsg
+		msgType = MsgPing
 		return
 	}
 
 	cmdx := msg[2]
 	if (cmdx & 0x80) == 0 {
 		// 请求包
-		msgType = RequestMsg
+		msgType = MsgRequest
 		cmdLen := int(cmdx)
 		cmd = string(msg[3 : cmdLen+3])
 		data = msg[cmdLen+3:]
 	} else {
 		// 响应数据包
-		msgType = ResponseMsg
+		msgType = MsgResponse
 		state = cmdx & 0x7F
 		data = msg[3:]
 	}

examples/client-tcp2.go

@@ -9,6 +9,7 @@ import (
 
 	"git.me9.top/git/tinymq"
 	"git.me9.top/git/tinymq/config"
+	"git.me9.top/git/tinymq/conn"
 )
 
 func main() {
@@ -16,7 +17,7 @@ func main() {
 	localChannel := "/tinymq/client/tcp2"
 	remoteChannel := "/tinymq/server"
 	remoteFilter := tinymq.StrChannelFilter(remoteChannel)
-	host := &tinymq.HostInfo{
+	host := &conn.HostInfo{
 		Proto:   "tcp",
 		Version: 2,
 		Host:    "127.0.0.1",
@@ -27,7 +28,7 @@ func main() {
 	hub := tinymq.NewHub(
 		cf,
 		localChannel,
-		func(channel string, hostType tinymq.HostType) (hostInfo *tinymq.HostInfo, err error) {
+		func(channel string, hostType tinymq.HostType) (hostInfo *conn.HostInfo, err error) {
 			return host, nil
 		},
 		func(client bool, proto string, version uint8, channel string, remoteAuth []byte) (auth []byte) {

examples/client-ws2.go

@@ -9,6 +9,7 @@ import (
 
 	"git.me9.top/git/tinymq"
 	"git.me9.top/git/tinymq/config"
+	"git.me9.top/git/tinymq/conn"
 )
 
 func main() {
@@ -17,7 +18,7 @@ func main() {
 	remoteChannel := "/tinymq/server"
 	remoteFilter := tinymq.StrChannelFilter(remoteChannel)
 
-	host := &tinymq.HostInfo{
+	host := &conn.HostInfo{
 		Proto:   "ws",
 		Version: 2,
 		Host:    "127.0.0.1",
@@ -30,7 +31,7 @@ func main() {
 	hub := tinymq.NewHub(
 		cf,
 		localChannel,
-		func(channel string, hostType tinymq.HostType) (hostInfo *tinymq.HostInfo, err error) {
+		func(channel string, hostType tinymq.HostType) (hostInfo *conn.HostInfo, err error) {
 			return host, nil
 		},
 		func(client bool, proto string, version uint8, channel string, remoteAuth []byte) (auth []byte) {

examples/server.go

@@ -12,6 +12,7 @@ import (
 
 	"git.me9.top/git/tinymq"
 	"git.me9.top/git/tinymq/config"
+	"git.me9.top/git/tinymq/conn"
 )
 
 func main() {
@@ -52,7 +53,7 @@ func main() {
 	)
 
 	// tcp2协议
-	bindTpv2Info := &tinymq.HostInfo{
+	bindTpv2Info := &conn.HostInfo{
 		Proto:   "tcp",
 		Version: 2,
 		Bind:    "127.0.0.1",
@@ -62,7 +63,7 @@ func main() {
 	hub.BindForServer(bindTpv2Info)
 
 	// ws2协议
-	bindws2Info := &tinymq.HostInfo{
+	bindws2Info := &conn.HostInfo{
 		Proto:   "ws",
 		Version: 2,
 		Bind:    "127.0.0.1",
@@ -73,7 +74,7 @@ func main() {
 	hub.BindForServer(bindws2Info)
 
 	// ws2协议，没有加密算法
-	bindInfo := &tinymq.HostInfo{
+	bindInfo := &conn.HostInfo{
 		Proto:   "ws",
 		Version: 2,
 		// Bind:      "127.0.0.1",

host.go

@@ -1,147 +0,0 @@
-package tinymq
-
-import (
-	"bytes"
-	"errors"
-	"fmt"
-	"regexp"
-	"strconv"
-	"strings"
-	"time"
-)
-
-// 连接服务结构
-type HostInfo struct {
-	Proto    string    `json:"proto" yaml:"proto"`                 // 协议
-	Version  uint8     `json:"version" yaml:"version"`             // 版本
-	Host     string    `json:"host" yaml:"host"`                   // 连接的IP地址或者域名
-	Bind     string    `json:"bind,omitempty" yaml:"bind"`         // 绑定的地址
-	Port     uint16    `json:"port,omitempty" yaml:"port"`         // 连接的端口
-	Path     string    `json:"path,omitempty" yaml:"path"`         // 连接的路径
-	Hash     string    `json:"hash,omitempty" yaml:"hash"`         // 连接验证使用，格式 method:key
-	Proxy    bool      `json:"proxy,omitempty" yaml:"proxy"`       // 是否代理
-	Nat      bool      `json:"nat,omitempty" yaml:"nat"`           // 是否是前端nat的方式处理
-	Priority int16     `json:"priority,omitempty" yaml:"priority"` // 优先级，-1 表示不可用，0 表示最高优先级（为了兼容没有优先级的节点），1-100 表示优先级别，数值越高优先级越高，不过这是由应用端决定的规则
-	Errors   uint16    `json:"errors,omitempty" yaml:"errors"`     // 连接失败计数，如果成功了则重置为0
-	Updated  time.Time `json:"updated" yaml:"updated"`             // 节点信息刷新时间
-}
-
-// 从 url 中解析信息
-// url 格式：ws2://xor:s^7mv7L!Mrn8Y!vn@127.0.0.1:14541/wsv2?proxy=1
-// 仅支持客户端连接使用
-func ParseUrl(url string) (hostInfo *HostInfo, err error) {
-	mx := regexp.MustCompile(`^([a-z]+)([0-9]*)://([^#/\?]+)(/[\w\-/]+)?`).FindStringSubmatch(url)
-	if mx == nil {
-		return nil, errors.New("invalid url")
-	}
-	protocol := mx[1]
-	version, _ := strconv.Atoi(mx[2])
-	host := mx[3]
-	index := strings.LastIndex(host, "@")
-	hash := ""
-	if index >= 0 {
-		hash = host[0:index]
-		host = host[index+1:]
-	}
-	// 检查是否ipv6，解析出ip和端口
-	index = strings.Index(host, "]:")
-	port := 0
-	if index > 0 {
-		// ipv6 地址和端口
-		port, err = strconv.Atoi(host[index+2:])
-		if err != nil {
-			return nil, err
-		}
-		host = host[1:index]
-	} else {
-		hs := strings.Split(host, ":")
-		if len(hs) == 2 {
-			host = hs[0]
-			port, err = strconv.Atoi(hs[1])
-			if err != nil {
-				return nil, err
-			}
-		}
-	}
-	path := ""
-	if len(mx) > 4 {
-		path = mx[4]
-	}
-	hostInfo = &HostInfo{
-		Proto:   protocol,
-		Version: uint8(version),
-		Host:    host,
-		Port:    uint16(port),
-		Hash:    hash,
-		Path:    path,
-	}
-	// 查找是否代理
-	url = url[len(mx[0]):]
-	if regexp.MustCompile(`[\?&]proxy=1`).MatchString(url) {
-		hostInfo.Proxy = true
-	}
-	if regexp.MustCompile(`[\?&]nat=1`).MatchString(url) {
-		hostInfo.Nat = true
-	}
-	priorityM := regexp.MustCompile(`[\?&]priority=(-?\d+)`).FindStringSubmatch(url)
-	if priorityM != nil {
-		priority, err := strconv.Atoi(priorityM[1])
-		if err != nil {
-			return nil, err
-		}
-		hostInfo.Priority = int16(priority)
-	}
-	return hostInfo, nil
-}
-
-// 只输出客户端要连接的信息
-func (h *HostInfo) Url() string {
-	if h == nil {
-		// 避免空出错
-		return ""
-	}
-	var b bytes.Buffer
-	fmt.Fprintf(&b, "%s%d://", h.Proto, h.Version)
-	if h.Hash != "" {
-		b.WriteString(h.Hash)
-		b.WriteString("@")
-	}
-	if strings.Contains(h.Host, ":") {
-		// ipv6
-		b.WriteString("[")
-		b.WriteString(h.Host)
-		b.WriteString("]")
-	} else {
-		b.WriteString(h.Host)
-	}
-	if h.Port > 0 {
-		fmt.Fprintf(&b, ":%d", h.Port)
-	}
-	if h.Path != "" {
-		b.WriteString(h.Path)
-	}
-	param := make([]string, 0)
-	if h.Proxy {
-		param = append(param, "proxy=1")
-	}
-	if h.Nat {
-		param = append(param, "nat=1")
-	}
-	if h.Priority != 0 {
-		param = append(param, fmt.Sprintf("priority=%d", h.Priority))
-	}
-	if len(param) > 0 {
-		b.WriteString("?")
-		b.WriteString(strings.Join(param, "&"))
-	}
-	return b.String()
-}
-
-// 输出代表一个节点的关键信息
-func (h *HostInfo) Key() string {
-	if h == nil {
-		// 避免空出错
-		return ""
-	}
-	return fmt.Sprintf("%s%d://%s:%d%s", h.Proto, h.Version, h.Host, h.Port, h.Path)
-}

hub.go

@@ -608,7 +608,7 @@ func (h *Hub) ConnectDuration(line *Line) time.Duration {
 
 // 绑定端口，建立服务
 // 需要程序运行时调用
-func (h *Hub) BindForServer(info *HostInfo) (err error) {
+func (h *Hub) BindForServer(info *conn.HostInfo) (err error) {
 	doConnectFunc := func(conn conn.Connect) {
 		proto, version, channel, auth, err := conn.ReadAuthInfo()
 		if err != nil {
@@ -663,7 +663,7 @@ func (h *Hub) BindForServer(info *HostInfo) (err error) {
 
 // 新建一个连接，不同的连接协议由底层自己选择
 // channel: 要连接的频道信息，需要能表达频道关键信息的部分
-func (h *Hub) ConnectToServer(channel string, force bool, host *HostInfo, autoReconnect bool) (err error) {
+func (h *Hub) ConnectToServer(channel string, force bool, host *conn.HostInfo, autoReconnect bool) (err error) {
 	h.connectMutex.Lock()
 	defer h.connectMutex.Unlock()
 	// 检查当前channel是否已经存在
@@ -691,8 +691,8 @@ func (h *Hub) ConnectToServer(channel string, force bool, host *HostInfo, autoRe
 	}
 
 	var conn conn.Connect
-	var runProto string
-	addr := net.JoinHostPort(host.Host, strconv.Itoa(int(host.Port)))
+	var rawProto string
+	// addr := net.JoinHostPort(host.Host, strconv.Itoa(int(host.Port)))
 
 	// 添加定时器
 	ctx, cancel := context.WithTimeout(context.Background(), time.Millisecond*time.Duration(h.cf.ConnectTimeout))
@@ -702,21 +702,7 @@ func (h *Hub) ConnectToServer(channel string, force bool, host *HostInfo, autoRe
 
 	// 发送连接请求
 	go func() {
-		if host.Version == ws2.VERSION && (host.Proto == ws2.PROTO || host.Proto == ws2.PROTO_STL) {
-			runProto = ws2.PROTO
-			if h.cf.PrintMsg {
-				log.Println("[ConnectToServer] connecting:", host.Proto, addr, host.Path, host.Hash)
-			}
-			conn, err = ws2.Dial(h.cf, host.Proto, addr, host.Path, host.Hash)
-		} else if host.Version == tcp2.VERSION && host.Proto == tcp2.PROTO {
-			runProto = tcp2.PROTO
-			if h.cf.PrintMsg {
-				log.Println("[ConnectToServer] connecting:", host.Proto, addr, host.Hash)
-			}
-			conn, err = tcp2.Dial(h.cf, addr, host.Hash)
-		} else {
-			err = fmt.Errorf("not correct protocol and version found in: %+v", host)
-		}
+		rawProto, conn, err = Dail(host, h.cf)
 		if done {
 			if err != nil {
 				log.Println("[ConnectToServer] dial failed:", err)
@@ -754,7 +740,7 @@ func (h *Hub) ConnectToServer(channel string, force bool, host *HostInfo, autoRe
 		localChannel = localChannel + "?proxy=" + host.Host
 	}
 	// 发送验证信息
-	if err := conn.WriteAuthInfo(localChannel, h.authFunc(true, runProto, host.Version, channel, nil)); err != nil {
+	if err := conn.WriteAuthInfo(localChannel, h.authFunc(true, rawProto, host.Version, channel, nil)); err != nil {
 		log.Println("[ConnectToServer] WriteAuthInfo failed:", err)
 		conn.Close()
 		host.Errors++
@@ -771,7 +757,7 @@ func (h *Hub) ConnectToServer(channel string, force bool, host *HostInfo, autoRe
 		return err
 	}
 	// 检查版本和协议是否一致
-	if version != host.Version || proto != runProto {
+	if version != host.Version || proto != rawProto {
 		err = fmt.Errorf("[ConnectToServer] version or protocol wrong: %d, %s", version, proto)
 		log.Println(err)
 		conn.Close()
@@ -833,7 +819,7 @@ func (h *Hub) ConnectToServer(channel string, force bool, host *HostInfo, autoRe
 
 // 重试方式连接服务
 // 将会一直阻塞直到连接成功
-func (h *Hub) ConnectToServerX(channel string, force bool, host *HostInfo) {
+func (h *Hub) ConnectToServerX(channel string, force bool, host *conn.HostInfo) {
 	for {
 		if host == nil {
 			if h.connectHostFunc == nil {

line.go

@@ -18,15 +18,15 @@ type Line struct {
 	cf             *config.Config
 	hub            *Hub
 	conn           conn.Connect
-	channel        string       // 连接对端的频道
-	state          ConnectState // 连接状态
-	host           *HostInfo    // 如果有值说明是客户端，就是主动连接端
-	autoReconnect  bool         // 指示是否自动重连
-	pingID         uint16       // 只有客户端使用
-	pingWrongCount uint8        // 记录 ping id 反馈错误次数，超过3次则重新连接
+	channel        string         // 连接对端的频道
+	state          ConnectState   // 连接状态
+	host           *conn.HostInfo // 如果有值说明是客户端，就是主动连接端
+	autoReconnect  bool           // 指示是否自动重连
+	pingID         uint16         // 只有客户端使用
+	pingWrongCount uint8          // 记录 ping id 反馈错误次数，超过3次则重新连接
 
-	proxyConn conn.Connect // 代理转发连接
-	proxyHost *HostInfo    // 代理地址
+	proxyConn conn.Connect   // 代理转发连接
+	proxyHost *conn.HostInfo // 代理地址
 
 	Extra sync.Map // 附加临时信息，由应用端决定具体内容，断线会自动清理
 	Keep  sync.Map // 附加固定信息，由应用端决定具体内容，断线不会自动清理，不过也不能保证一直保持有值，断线后可能会被系统清理
@@ -44,7 +44,7 @@ type Line struct {
 }
 
 // 获取当前连接信息
-func (c *Line) Host() *HostInfo {
+func (c *Line) Host() *conn.HostInfo {
 	return c.host
 }
 
@@ -148,12 +148,12 @@ func (c *Line) readPump() {
 		// 记录最后读到数据的时间
 		c.lastRead = time.Now()
 		switch msgType {
-		case conn.PingMsg:
+		case conn.MsgPing:
 			// ping或pong包
 			c.pingRequest <- &PingData{
 				Id: id,
 			}
-		case conn.RequestMsg:
+		case conn.MsgRequest:
 			// 请求数据包
 			go c.hub.requestFromNet(&RequestData{
 				Id:   id,
@@ -161,7 +161,7 @@ func (c *Line) readPump() {
 				Data: data,
 				conn: c,
 			})
-		case conn.ResponseMsg:
+		case conn.MsgResponse:
 			// 网络回应数据包
 			go c.hub.outResponse(&ResponseData{
 				Id:    id,
@@ -321,7 +321,7 @@ func (c *Line) cleanClose() {
 }
 
 // 连接开始运行
-func (c *Line) Start(channel string, conn conn.Connect, host *HostInfo) {
+func (c *Line) Start(channel string, conn conn.Connect, host *conn.HostInfo) {
 	now := time.Now()
 	c.updated = now
 	c.lastRead = now // 避免默认为0时被清理
@@ -340,7 +340,7 @@ func NewConnect(
 	hub *Hub,
 	channel string,
 	conn conn.Connect,
-	host *HostInfo,
+	host *conn.HostInfo,
 	autoReconnect bool,
 ) *Line {
 	now := time.Now()

type.go

@@ -3,6 +3,8 @@ package tinymq
 import (
 	"fmt"
 	"time"
+
+	"git.me9.top/git/tinymq/conn"
 )
 
 // 中间件函数
@@ -129,7 +131,7 @@ type GetMsg struct {
 }
 
 // 获取对应频道的一个连接地址
-type ConnectHostFunc func(channel string, hostType HostType) (hostInfo *HostInfo, err error)
+type ConnectHostFunc func(channel string, hostType HostType) (hostInfo *conn.HostInfo, err error)
 
 // 获取认证信息
 type AuthFunc func(client bool, proto string, version uint8, channel string, remoteAuth []byte) (auth []byte)

type_test.go

@@ -1,16 +0,0 @@
-package tinymq
-
-import "testing"
-
-func TestParseUrl(t *testing.T) {
-	url := "ws2://xor:s^7mv7L!Mrn8Y!vn@127.0.0.1:141/wsv2?proxy=1&priority=90"
-	hostInfo, err := ParseUrl(url)
-	if err != nil {
-		t.Error(err)
-		return
-	}
-	result := hostInfo.Url()
-	if result != url {
-		t.Errorf("want: %s, get: %s", url, result)
-	}
-}