fix

README.md

@@ -18,6 +18,7 @@
 - 自动重连
 - 多协议绑定
 - 连接验证
+- 自动使用代理来完成连接
 
 ## 优化连接方式
 

config/const.go

@@ -1,5 +1,8 @@
 package config
 
+const STATE_OK = 1
+const STATE_FAILED = 0
+
 const (
 	// 系统错误号定义，最低号为110，最高127
 	MIN_SYSTEM_ERROR_CODE = 110 // 系统信息最小值

conn/tcp2/tcp2.go

@@ -401,7 +401,7 @@ func (c *Tcp2) WriteRequest(id uint16, cmd string, data []byte) error {
 	}
 	dlen := len(data)
 	if c.compress && dlen > 0 {
-		compressedData, ok := util.CompressData(data)
+		compressedData, ok, _ := util.CompressData(data)
 		ddlen := 2 + 1 + cmdLen + 1 + len(compressedData)
 		buf, start := c.writeDataLen(ddlen)
 		index := start
@@ -447,7 +447,7 @@ func (c *Tcp2) WriteRequest(id uint16, cmd string, data []byte) error {
 func (c *Tcp2) WriteResponse(id uint16, state uint8, data []byte) error {
 	dlen := len(data)
 	if c.compress && dlen > 0 {
-		compressedData, ok := util.CompressData(data)
+		compressedData, ok, _ := util.CompressData(data)
 		ddlen := 2 + 1 + 1 + len(compressedData)
 		buf, start := c.writeDataLen(ddlen)
 		index := start
@@ -528,8 +528,9 @@ func (c *Tcp2) ReadMessage(deadline int) (msgType conn.MsgType, id uint16, cmd s
 	}
 	dataLen := len(data)
 	if c.compress && dataLen > 1 {
-		data, _ = util.UncompressData(data)
-		if c.cf.PrintMsg {
+		var ok bool
+		data, ok, err = util.UncompressData(data)
+		if ok && c.cf.PrintMsg {
 			log.Println("[UncompressData]", dataLen, "->", len(data))
 		}
 	}

conn/util/bridge.go

@@ -0,0 +1,26 @@
+package util
+
+import "io"
+
+// 桥接两个 IO 连接
+func Bridge(conn1, conn2 io.ReadWriteCloser) error {
+	defer conn1.Close()
+	defer conn2.Close()
+
+	errChan := make(chan error, 2)
+
+	// Copy conn1 to conn2
+	go func() {
+		_, err := io.Copy(conn2, conn1)
+		errChan <- err
+	}()
+
+	// Copy conn2 to conn1
+	go func() {
+		_, err := io.Copy(conn1, conn2)
+		errChan <- err
+	}()
+
+	// Wait for either direction to finish or error
+	return <-errChan
+}

conn/util/compress.go

@@ -9,42 +9,41 @@ import (
 
 // 压缩数据
 // 注意：如果没有启用压缩，和数据为空不要调用该函数
-func CompressData(data []byte) (out []byte, ok bool) {
+func CompressData(data []byte) (out []byte, ok bool, err error) {
 	var buf bytes.Buffer
 	zw := gzip.NewWriter(&buf)
 	if _, err := zw.Write(data); err != nil {
 		log.Println("[CompressData Write ERROR]", err)
-		return data, false
+		return data, false, err
 	}
 	if err := zw.Close(); err != nil {
 		log.Println("[CompressData Close ERROR]", err)
-		return data, false
+		return data, false, err
 	}
 	compressedData := buf.Bytes()
 	if len(compressedData) >= len(data) {
-		// log.Println("user origin data:", string(data))
-		return data, false
+		return data, false, nil
 	} else {
-		return compressedData, true
+		return compressedData, true, nil
 	}
 }
 
-// 解压缩
-func UncompressData(compressedData []byte) (data []byte, ok bool) {
+// 解压缩，
+func UncompressData(compressedData []byte) (data []byte, ok bool, err error) {
 	if compressedData[0]&0x01 != 0 {
 		zr, err := gzip.NewReader(bytes.NewReader(compressedData[1:]))
 		if err != nil {
 			log.Println("[UncompressData NewReader ERROR]", err)
-			return compressedData, false
+			return compressedData, false, err
 		}
 		defer zr.Close()
 		uncompressedData, err := io.ReadAll(zr)
 		if err != nil {
 			log.Println("[UncompressData ReadAll ERROR]", err)
-			return compressedData, false
+			return compressedData, false, err
 		}
-		return uncompressedData, true
+		return uncompressedData, true, nil
 	} else {
-		return compressedData[1:], true
+		return compressedData[1:], false, nil
 	}
 }

conn/ws2/ws2.go

@@ -338,7 +338,7 @@ func (c *Ws2) WriteRequest(id uint16, cmd string, data []byte) error {
 	}
 	dlen := len(data)
 	if c.compress && dlen > 0 {
-		compressedData, ok := util.CompressData(data)
+		compressedData, ok, _ := util.CompressData(data)
 		ddlen := 2 + 1 + cmdLen + 1 + len(compressedData)
 		buf := make([]byte, ddlen) // 申请内存
 		index := 0
@@ -378,7 +378,7 @@ func (c *Ws2) WriteRequest(id uint16, cmd string, data []byte) error {
 func (c *Ws2) WriteResponse(id uint16, state uint8, data []byte) error {
 	dlen := len(data)
 	if c.compress && dlen > 0 {
-		compressedData, ok := util.CompressData(data)
+		compressedData, ok, _ := util.CompressData(data)
 		ddlen := 2 + 1 + 1 + len(compressedData)
 		buf := make([]byte, ddlen)
 		binary.BigEndian.PutUint16(buf[0:2], id)
@@ -456,8 +456,9 @@ func (c *Ws2) ReadMessage(deadline int) (msgType conn.MsgType, id uint16, cmd st
 	}
 	dataLen := len(data)
 	if c.compress && dataLen > 1 {
-		data, _ = util.UncompressData(data)
-		if c.cf.PrintMsg {
+		var ok bool
+		data, ok, err = util.UncompressData(data)
+		if ok && c.cf.PrintMsg {
 			log.Println("[UncompressData]", dataLen, "->", len(data))
 		}
 	}

const.go

@@ -7,8 +7,8 @@ import (
 )
 
 // 定义成功与失败的值
-const STATE_OK = 1
-const STATE_FAILED = 0
+const STATE_OK = config.STATE_OK
+const STATE_FAILED = config.STATE_FAILED
 
 const (
 	// 系统错误号定义，最低号为110，最高127
@@ -27,9 +27,9 @@ const (
 
 const (
 	// ID 号最高值，高于这个值的ID号为系统内部使用
-	ID_MAX = 65500
+	ID_MAX = config.ID_MAX
 	// 验证ID
-	ID_AUTH = 65502
+	ID_AUTH = config.ID_AUTH
 )
 
 // 转换 id 到对应的消息

filter.go

@@ -7,6 +7,8 @@ import (
 	"strings"
 )
 
+// 一些常用的过滤器
+
 // 匹配所有的频道
 func AllChannelFilter() FilterFunc {
 	return func(conn *Line) (ok bool) {

host.go

@@ -0,0 +1,147 @@
+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

@@ -332,7 +332,7 @@ func (h *Hub) sendRequest(gd *GetData) (count int, err error) {
 			channel := h.filterToChannelFunc(gd.Filter)
 			if channel == "" {
 				err = errors.New(IdMsg(NO_MATCH_CONNECT))
-				log.Println("not channel found")
+				log.Println("not channel found with cmd:", gd.Cmd)
 				return 0, err
 			}
 			err := h.ConnectToServer(channel, false, nil, false)
@@ -868,7 +868,9 @@ func (h *Hub) checkConnect() {
 	if h.connectHostFunc == nil {
 		return
 	}
+	// 检查是否正在使用代理的节点，如果可行的话将其直连
 	proxyTicker := time.NewTicker(time.Duration(h.cf.ProxyTimeout * int(time.Millisecond)))
+	// 检查连接是否还活着，如果没有的话重连
 	connectTicker := time.NewTicker(time.Millisecond * time.Duration(h.cf.ConnectCheck))
 	for {
 		select {

line.go

@@ -254,8 +254,11 @@ func (c *Line) writePump() {
 				// 超时关闭当前的连接
 				log.Println("[Connect timeout and stop it]", c.channel)
 				// 有可能连接出现问题，断开并重新连接
-				c.Close(false)
-				return
+				// 检测到浏览器会出现休眠的状态，服务端就不主动关闭，只有客户端关闭
+				if c.host != nil {
+					c.Close(false)
+					return
+				}
 			}
 			// 只需要客户端发送
 			if c.host != nil {

mapx.go

@@ -1,5 +1,7 @@
 package tinymq
 
+// 一个简单的订阅查找匹配组件
+
 /* 优化连接方式
 
 - 高效匹配字符串算法

type.go

@@ -1,14 +1,7 @@
 package tinymq
 
 import (
-	"bytes"
-	"errors"
 	"fmt"
-	"regexp"
-	"strconv"
-
-	// "regexp"
-	"strings"
 	"time"
 )
 
@@ -129,138 +122,6 @@ type GetMsg struct {
 	timer *time.Timer
 }
 
-// 连接服务结构
-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,omitempty" 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 + "@")
-	}
-	if strings.Contains(h.Host, ":") {
-		// ipv6
-		b.WriteString("[" + h.Host + "]")
-	} 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("?" + 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)
-}
-
 // 获取对应频道的一个连接地址
 type ConnectHostFunc func(channel string, hostType HostType) (hostInfo *HostInfo, err error)