finish proxy function, will try to debug

config/config.go

@@ -21,7 +21,7 @@ func NewConfig() *Config {
 		ConnectTimeout:       30 * 1000,
 		PingInterval:         61 * 1000,
 		WriteWait:            60 * 1000,
-		ReadWait:             30 * 1000,
+		ReadWait:             60 * 1000,
 		LongReadWait:         150 * 1000,
 		CleanDeadConnectWait: 3600 * 1000,
 		PrintPing:            false,

dail.go

@@ -88,35 +88,40 @@ func ClientCheckAuthInfo(
 	remoteChannel string,
 	authFunc AuthFunc,
 	checkAuthFunc CheckAuthFunc,
-) (err error) {
+) (needProxy bool, err error) {
 	// 发送验证信息
-	if err := conn.WriteAuthInfo(localChannel, authFunc(true, host.Proto, host.Version, remoteChannel, nil)); err != nil {
+	if err := conn.WriteAuthInfo(localChannel, authFunc(host, host.Proto, host.Version, remoteChannel, nil)); err != nil {
 		log.Println("[ClientCheckAuthInfo] WriteAuthInfo failed:", err)
-		return err
+		return false, err
 	}
 	// 接收频道信息
 	proto, version, channel, auth, err := conn.ReadAuthInfo()
 	if err != nil {
 		log.Println("[ClientCheckAuthInfo] ReadAuthInfo failed:", err)
-		return err
+		return false, err
 	}
 	// 检查版本和协议是否一致
 	if version != host.Version || !strings.HasPrefix(host.Proto, proto) {
 		err = fmt.Errorf("[ClientCheckAuthInfo] version or protocol wrong: %d, %s", version, proto)
 		log.Println(err)
-		return err
+		return false, err
 	}
 	// 检查频道名称是否匹配
-	if !strings.Contains(channel, remoteChannel) {
+	if !strings.HasPrefix(channel, remoteChannel) {
+		// 代理节点
+		if host.Proxy {
+			// 还需要再建立连接
+			return true, nil
+		}
 		err = fmt.Errorf("[ClientCheckAuthInfo] channel want %s, get %s", remoteChannel, channel)
 		log.Println(err)
-		return err
+		return false, err
 	}
 	// 检查验证是否合法
-	if !checkAuthFunc(true, proto, version, channel, auth) {
+	if !checkAuthFunc(host, proto, version, channel, auth) {
 		err = fmt.Errorf("[ClientCheckAuthInfo] failed in proto: %s, version: %d, channel: %s, auth: %s", proto, version, channel, string(auth))
 		log.Println(err)
-		return err
+		return false, err
 	}
 	return
 }

examples/client-tcp2.go

@@ -31,12 +31,12 @@ func main() {
 		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) {
-			log.Println("[AuthFunc]", client, proto, version, channel, string(remoteAuth))
+		func(host *conn.HostInfo, proto string, version uint8, channel string, remoteAuth []byte) (auth []byte) {
+			log.Println("[AuthFunc]", host, proto, version, channel, string(remoteAuth))
 			return []byte("tinymq-client")
 		},
-		func(client bool, proto string, version uint8, channel string, auth []byte) bool {
-			log.Println("[CheckAuthFunc]", client, proto, version, channel, string(auth))
+		func(host *conn.HostInfo, proto string, version uint8, channel string, auth []byte) bool {
+			log.Println("[CheckAuthFunc]", host, proto, version, channel, string(auth))
 			return string(auth) == "tinymq-server"
 		},
 		func(conn *tinymq.Line) {
@@ -44,6 +44,7 @@ func main() {
 		},
 	)
 
+	// 从过滤器到频道字符串
 	hub.SetFilterToChannelFunc(func(filter tinymq.FilterFunc) (channel string) {
 		return remoteChannel
 	})

examples/client-ws2.go

@@ -34,12 +34,12 @@ func main() {
 		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) {
-			log.Println("[AuthFunc]", client, proto, version, channel, string(remoteAuth))
+		func(host *conn.HostInfo, proto string, version uint8, channel string, remoteAuth []byte) (auth []byte) {
+			log.Println("[AuthFunc]", host, proto, version, channel, string(remoteAuth))
 			return []byte("tinymq-client")
 		},
-		func(client bool, proto string, version uint8, channel string, auth []byte) bool {
-			log.Println("[CheckAuthFunc]", client, proto, version, channel, string(auth))
+		func(host *conn.HostInfo, proto string, version uint8, channel string, auth []byte) bool {
+			log.Println("[CheckAuthFunc]", host, proto, version, channel, string(auth))
 			return string(auth) == "tinymq-server"
 		},
 		func(conn *tinymq.Line) {

examples/server.go

@@ -27,12 +27,12 @@ func main() {
 		cf,
 		localChannel,
 		nil,
-		func(client bool, proto string, version uint8, channel string, remoteAuth []byte) (auth []byte) {
-			log.Println("[AuthFunc]", client, proto, version, channel, string(remoteAuth))
+		func(host *conn.HostInfo, proto string, version uint8, channel string, remoteAuth []byte) (auth []byte) {
+			log.Println("[AuthFunc]", host, proto, version, channel, string(remoteAuth))
 			return []byte("tinymq-server")
 		},
-		func(client bool, proto string, version uint8, channel string, auth []byte) bool {
-			log.Println("[CheckAuthFunc]", client, proto, version, channel, string(auth))
+		func(host *conn.HostInfo, proto string, version uint8, channel string, auth []byte) bool {
+			log.Println("[CheckAuthFunc]", host, proto, version, channel, string(auth))
 			return string(auth) == "tinymq-client"
 		},
 		func(conn *tinymq.Line) {

hub.go

@@ -621,14 +621,14 @@ func (h *Hub) BindForServer(info *conn.HostInfo) (err error) {
 			conn.Close()
 			return
 		}
-		if !h.checkAuthFunc(false, proto, version, channel, auth) {
+		if !h.checkAuthFunc(nil, proto, version, channel, auth) {
 			err = fmt.Errorf("[server checkAuthFunc ERROR] in proto: %s, version: %d, channel: %s, auth: %s", proto, version, channel, string(auth))
 			log.Println(err)
 			conn.Close()
 			return
 		}
 		// 发送频道信息
-		if err := conn.WriteAuthInfo(h.channel, h.authFunc(false, proto, version, channel, auth)); err != nil {
+		if err := conn.WriteAuthInfo(h.channel, h.authFunc(nil, proto, version, channel, auth)); err != nil {
 			log.Println("[WriteAuthInfo ERROR]", err)
 			conn.Close()
 			return
@@ -661,6 +661,47 @@ func (h *Hub) BindForServer(info *conn.HostInfo) (err error) {
 	return errors.New("not connect protocol and version found")
 }
 
+// 代理连接，在已经连接的基础上进行代理连接
+func (h *Hub) ProxyConnect(connect conn.Connect, remoteChannel string) (err error) {
+	host, err := h.connectHostFunc(remoteChannel, HostTypeDirect)
+	if err != nil {
+		return err
+	}
+	// 发送代理连接
+	buf := conn.ProxyConnectPackageEncode(host.Url())
+	if err := connect.WriteRawPackage(buf, true); err != nil {
+		log.Println("[ConnectToServer] WriteRawPackage error:", err)
+		return err
+	}
+	// 读取代理结果
+	msgType, _, _, _, data, err := connect.ReadMessage(h.cf.LongReadWait)
+	if err != nil {
+		log.Println("[ConnectToServer] ReadRawPackage error:", err)
+		return err
+	}
+	if msgType != conn.MsgProxyResult {
+		return errors.New("not found MsgProxyResult")
+	}
+	if len(data) > 0 {
+		log.Println("[ConnectToServer] proxy result error:", string(data))
+		return errors.New(string(data))
+	}
+	// 进行auth验证
+	needProxy, err := ClientCheckAuthInfo(connect, host, h.channel, remoteChannel, h.authFunc, h.checkAuthFunc)
+	if err != nil {
+		host.Errors++
+		host.Updated = time.Now()
+		return err
+	} else {
+		host.Errors = 0
+		host.Updated = time.Now()
+	}
+	if needProxy {
+		return errors.New("proxy with proxy")
+	}
+	return nil
+}
+
 // 新建一个连接，不同的连接协议由底层自己选择
 // channel: 要连接的频道信息，需要能表达频道关键信息的部分
 func (h *Hub) ConnectToServer(remoteChannel string, force bool, host *conn.HostInfo, autoReconnect bool) (err error) {
@@ -680,7 +721,7 @@ func (h *Hub) ConnectToServer(remoteChannel string, force bool, host *conn.HostI
 			return errors.New("not connect host func found")
 		}
 		// 获取服务地址等信息
-		host, err = h.connectHostFunc(remoteChannel, Both)
+		host, err = h.connectHostFunc(remoteChannel, HostTypeBoth)
 		if err != nil {
 			return err
 		}
@@ -690,64 +731,15 @@ func (h *Hub) ConnectToServer(remoteChannel string, force bool, host *conn.HostI
 		}
 	}
 
-	conn, err := DailWithTimeout(h.ctx, host, h.cf)
+	connect, err := DailWithTimeout(h.ctx, host, h.cf)
 	if err != nil {
 		log.Println("[DailWithTimeout] error:", err)
 		return err
 	}
 
-	// var conn conn.Connect
-	// // 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))
-	// defer cancel()
-	// taskCh := make(chan bool)
-	// done := false // 指示是否已经完成
-
-	// // 发送连接请求
-	// go func() {
-	// 	conn, err = Dail(host, h.cf)
-	// 	if done {
-	// 		if err != nil {
-	// 			log.Println("[ConnectToServer] dial failed:", err)
-	// 		}
-	// 		if conn != nil {
-	// 			conn.Close()
-	// 		}
-	// 	} else {
-	// 		taskCh <- err == nil
-	// 	}
-	// }()
-
-	// select {
-	// case ok := <-taskCh:
-	// 	cancel()
-	// 	if !ok || err != nil || conn == nil {
-	// 		log.Println("[ConnectToServer] connect failed:", err)
-	// 		host.Errors++
-	// 		host.Updated = time.Now()
-	// 		if err == nil {
-	// 			err = errors.New("unknown error")
-	// 		}
-	// 		return err
-	// 	}
-	// case <-ctx.Done():
-	// 	done = true
-	// 	return errors.New("timeout")
-	// case <-h.ctx.Done():
-	// 	return errors.New("quit")
-	// }
-
-	// 如果 host 是代理，将代理信息添加到channel中
-	// localChannel := h.channel
-	// if host.Proxy {
-	// 	localChannel = localChannel + "?proxy=" + host.Host
-	// }
-	err = ClientCheckAuthInfo(conn, host, h.channel, remoteChannel, h.authFunc, h.checkAuthFunc)
+	needProxy, err := ClientCheckAuthInfo(connect, host, h.channel, remoteChannel, h.authFunc, h.checkAuthFunc)
 	if err != nil {
-		conn.Close()
+		connect.Close()
 		host.Errors++
 		host.Updated = time.Now()
 		return err
@@ -756,6 +748,16 @@ func (h *Hub) ConnectToServer(remoteChannel string, force bool, host *conn.HostI
 		host.Updated = time.Now()
 	}
 
+	// 判断是否是代理节点，如果是代理节点还需要进行下一步的连接
+	if needProxy {
+		err := h.ProxyConnect(connect, remoteChannel)
+		if err != nil {
+			connect.Close()
+			log.Println("[ProxyConnect] error:", err)
+			return err
+		}
+	}
+
 	// 将连接加入现有连接中
 	done := false
 	h.lines.Range(func(id int, line *Line) bool {
@@ -769,7 +771,7 @@ func (h *Hub) ConnectToServer(remoteChannel string, force bool, host *conn.HostI
 					return false
 				}
 			}
-			line.Start(remoteChannel, conn, host)
+			line.Start(remoteChannel, connect, host)
 			done = true
 			return false
 		}
@@ -780,7 +782,7 @@ func (h *Hub) ConnectToServer(remoteChannel string, force bool, host *conn.HostI
 	}
 	// 新建一个连接
 	if !done {
-		line := NewConnect(h.cf, h, remoteChannel, conn, host, autoReconnect)
+		line := NewConnect(h.cf, h, remoteChannel, connect, host, autoReconnect)
 		h.addLine(line)
 	}
 	return nil
@@ -789,13 +791,24 @@ func (h *Hub) ConnectToServer(remoteChannel string, force bool, host *conn.HostI
 // 重试方式连接服务
 // 将会一直阻塞直到连接成功
 func (h *Hub) ConnectToServerX(channel string, force bool, host *conn.HostInfo) {
+	count := 0
 	for {
 		if host == nil {
 			if h.connectHostFunc == nil {
 				log.Println("[ConnectToServerX] not connect host func found")
 				return
 			}
-			hx, err := h.connectHostFunc(channel, Both)
+			// 通过尝试次数来获取不同类型的服务节点
+			var hostType HostType
+			if count < 2 {
+				hostType = HostTypeDirect
+			} else if count < 5 {
+				hostType = HostTypeBoth
+			} else {
+				hostType = HostTypeProxy
+			}
+			count++
+			hx, err := h.connectHostFunc(channel, hostType)
 			if err == nil {
 				err := h.ConnectToServer(channel, force, hx, true)
 				if err == nil {
@@ -834,7 +847,7 @@ func (h *Hub) checkConnect() {
 			now := time.Now().UnixMilli()
 			h.lines.Range(func(id int, line *Line) bool {
 				if line.host != nil && line.host.Proxy && now-line.updated.UnixMilli() > int64(h.cf.ProxyTimeout) {
-					host, err := h.connectHostFunc(line.channel, Direct)
+					host, err := h.connectHostFunc(line.channel, HostTypeDirect)
 					if err != nil {
 						log.Println("[proxyTicker connectHostFunc]", err)
 						return false

line.go

@@ -344,6 +344,10 @@ func (c *Line) Close(quick bool) {
 			go c.hub.cleanDeadConnect()
 		}
 	}
+	if c.proxyConn != nil {
+		c.proxyConn.Close()
+		// c.proxyConn = nil
+	}
 	c.Extra.Clear()
 	// c.updated = time.Now()
 }

type.go

@@ -77,9 +77,9 @@ func (t ConnectState) String() string {
 type HostType byte
 
 const (
-	Direct HostType = iota
-	Proxy
-	Both
+	HostTypeDirect HostType = iota
+	HostTypeProxy
+	HostTypeBoth
 )
 
 // 请求数据包
@@ -134,10 +134,11 @@ type GetMsg struct {
 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 AuthFunc func(host *conn.HostInfo, proto string, version uint8, channel string, remoteAuth []byte) (auth []byte)
 
-// 认证合法性函数
-type CheckAuthFunc func(client bool, proto string, version uint8, channel string, auth []byte) bool
+// 认证合法性函数，只有客户端有host信息
+// host 后面的部分是远程返回回来的信息
+type CheckAuthFunc func(host *conn.HostInfo, proto string, version uint8, channel string, auth []byte) bool
 
 // 验证发送的数据条件是否满足
 type CheckConnectOkFunc = func(line *Line, data *GetData) bool