still codding

dail.go

@@ -5,6 +5,7 @@ import (
 	"log"
 	"net"
 	"strconv"
+	"strings"
 
 	"git.me9.top/git/tinymq/config"
 	"git.me9.top/git/tinymq/conn"
@@ -12,20 +13,18 @@ import (
 	"git.me9.top/git/tinymq/conn/ws2"
 )
 
-// 建立连接，输出实际代码实现的协议
+// 根据协议名建立连接，输出实际代码实现的协议
 // 会一直阻塞直到连接结束
-func Dail(host *conn.HostInfo, cf *config.Config) (rawProto string, conn conn.Connect, err error) {
+func Dail(host *conn.HostInfo, cf *config.Config) (conn conn.Connect, err error) {
 	addr := net.JoinHostPort(host.Host, strconv.Itoa(int(host.Port)))
 	if host.Version == ws2.VERSION && (host.Proto == ws2.PROTO || host.Proto == ws2.PROTO_STL) {
-		rawProto = ws2.PROTO
 		if cf.PrintMsg {
-			log.Println("[ConnectToServer] connecting:", host.Proto, addr, host.Path, host.Hash)
+			log.Println("[Dail]:", host.Proto, addr, host.Path, host.Hash)
 		}
 		conn, err = ws2.Dial(cf, host.Proto, addr, host.Path, host.Hash)
 	} else if host.Version == tcp2.VERSION && host.Proto == tcp2.PROTO {
-		rawProto = tcp2.PROTO
 		if cf.PrintMsg {
-			log.Println("[ConnectToServer] connecting:", host.Proto, addr, host.Hash)
+			log.Println("[Dail]:", host.Proto, addr, host.Hash)
 		}
 		conn, err = tcp2.Dial(cf, addr, host.Hash)
 	} else {
@@ -33,3 +32,61 @@ func Dail(host *conn.HostInfo, cf *config.Config) (rawProto string, conn conn.Co
 	}
 	return
 }
+
+// 检验 AuthInfo
+func CheckAuthInfo(
+	conn conn.Connect,
+	host *conn.HostInfo,
+	localChannel string,
+	remoteChannel string,
+	authFunc AuthFunc,
+	checkAuthFunc CheckAuthFunc,
+) (err error) {
+	if err := conn.WriteAuthInfo(localChannel, authFunc(true, host.Proto, host.Version, remoteChannel, nil)); err != nil {
+		log.Println("[CheckAuthInfo] WriteAuthInfo failed:", err)
+		// conn.Close()
+		// host.Errors++
+		// host.Updated = time.Now()
+		return err
+	}
+	// 接收频道信息
+	proto, version, channel, auth, err := conn.ReadAuthInfo()
+	if err != nil {
+		log.Println("[CheckAuthInfo] ReadAuthInfo failed:", err)
+		// conn.Close()
+		// host.Errors++
+		// host.Updated = time.Now()
+		return err
+	}
+	// 检查版本和协议是否一致
+	if version != host.Version || !strings.HasPrefix(host.Proto, proto) {
+		err = fmt.Errorf("[CheckAuthInfo] version or protocol wrong: %d, %s", version, proto)
+		log.Println(err)
+		// conn.Close()
+		// host.Errors++
+		// host.Updated = time.Now()
+		return err
+	}
+	// 检查频道名称是否匹配
+	if !strings.Contains(channel, remoteChannel) {
+		err = fmt.Errorf("[CheckAuthInfo] channel want %s, get %s", remoteChannel, channel)
+		log.Println(err)
+		// conn.Close()
+		// host.Errors++
+		// host.Updated = time.Now()
+		return err
+	}
+	// 检查验证是否合法
+	if !checkAuthFunc(true, proto, version, channel, auth) {
+		err = fmt.Errorf("[CheckAuthInfo] client checkAuthFunc in proto: %s, version: %d, channel: %s, auth: %s", proto, version, channel, string(auth))
+		log.Println(err)
+		// conn.Close()
+		// host.Errors++
+		// host.Updated = time.Now()
+		return err
+	}
+	// 更新服务主机信息
+	// host.Errors = 0
+	// host.Updated = time.Now()
+	return
+}

examples/client-tcp2.go

@@ -73,6 +73,6 @@ func main() {
 	}
 	log.Println("[RESULT]<-", string(rsp.Data))
 
-	time.Sleep(time.Second * 300)
+	time.Sleep(time.Second * 30)
 	log.Println("client exit")
 }

examples/client-ws2.go

@@ -98,6 +98,6 @@ func main() {
 	time.Sleep(time.Second * 5)
 	hub.Push(remoteFilter, "push", []byte(time.Now().GoString()))
 
-	time.Sleep(time.Second * 300)
+	time.Sleep(time.Second * 30)
 	log.Println("client exit")
 }

hub.go

@@ -11,7 +11,6 @@ import (
 
 	// "regexp"
 	"strconv"
-	"strings"
 	"sync"
 	"time"
 
@@ -663,15 +662,15 @@ func (h *Hub) BindForServer(info *conn.HostInfo) (err error) {
 
 // 新建一个连接，不同的连接协议由底层自己选择
 // channel: 要连接的频道信息，需要能表达频道关键信息的部分
-func (h *Hub) ConnectToServer(channel string, force bool, host *conn.HostInfo, autoReconnect bool) (err error) {
+func (h *Hub) ConnectToServer(remoteChannel string, force bool, host *conn.HostInfo, autoReconnect bool) (err error) {
 	h.connectMutex.Lock()
 	defer h.connectMutex.Unlock()
 	// 检查当前channel是否已经存在
 	if !force {
-		line := h.ChannelToLine(channel)
+		line := h.ChannelToLine(remoteChannel)
 		if line != nil && line.state == Connected {
 			// err = fmt.Errorf("[ConnectToServer ERROR] existed channel: %s", channel)
-			log.Println("[ConnectToServer] channel existed:", channel)
+			log.Println("[ConnectToServer] channel existed:", remoteChannel)
 			return
 		}
 	}
@@ -680,7 +679,7 @@ func (h *Hub) ConnectToServer(channel string, force bool, host *conn.HostInfo, a
 			return errors.New("not connect host func found")
 		}
 		// 获取服务地址等信息
-		host, err = h.connectHostFunc(channel, Both)
+		host, err = h.connectHostFunc(remoteChannel, Both)
 		if err != nil {
 			return err
 		}
@@ -691,7 +690,7 @@ func (h *Hub) ConnectToServer(channel string, force bool, host *conn.HostInfo, a
 	}
 
 	var conn conn.Connect
-	var rawProto string
+	// var rawProto string
 	// addr := net.JoinHostPort(host.Host, strconv.Itoa(int(host.Port)))
 
 	// 添加定时器
@@ -702,7 +701,7 @@ func (h *Hub) ConnectToServer(channel string, force bool, host *conn.HostInfo, a
 
 	// 发送连接请求
 	go func() {
-		rawProto, conn, err = Dail(host, h.cf)
+		conn, err = Dail(host, h.cf)
 		if done {
 			if err != nil {
 				log.Println("[ConnectToServer] dial failed:", err)
@@ -735,72 +734,82 @@ func (h *Hub) ConnectToServer(channel string, force bool, host *conn.HostInfo, a
 	}
 
 	// 如果 host 是代理，将代理信息添加到channel中
-	localChannel := h.channel
-	if host.Proxy {
-		localChannel = localChannel + "?proxy=" + host.Host
-	}
-	// 发送验证信息
-	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++
-		host.Updated = time.Now()
-		return err
-	}
-	// 接收频道信息
-	proto, version, channel2, auth, err := conn.ReadAuthInfo()
+	// localChannel := h.channel
+	// if host.Proxy {
+	// 	localChannel = localChannel + "?proxy=" + host.Host
+	// }
+	err = CheckAuthInfo(conn, host, h.channel, remoteChannel, h.authFunc, h.checkAuthFunc)
 	if err != nil {
-		log.Println("[ConnectToServer] ReadAuthInfo failed:", err)
 		conn.Close()
 		host.Errors++
 		host.Updated = time.Now()
 		return err
-	}
-	// 检查版本和协议是否一致
-	if version != host.Version || proto != rawProto {
-		err = fmt.Errorf("[ConnectToServer] version or protocol wrong: %d, %s", version, proto)
-		log.Println(err)
-		conn.Close()
-		host.Errors++
+	} else {
+		host.Errors = 0
 		host.Updated = time.Now()
-		return err
 	}
-	// 检查频道名称是否匹配
-	if !strings.Contains(channel2, channel) {
-		err = fmt.Errorf("[ConnectToServer] channel want %s, get %s", channel, channel2)
-		log.Println(err)
-		conn.Close()
-		host.Errors++
-		host.Updated = time.Now()
-		return err
-	}
-	// 检查验证是否合法
-	if !h.checkAuthFunc(true, proto, version, channel, auth) {
-		err = fmt.Errorf("[ConnectToServer] client checkAuthFunc in proto: %s, version: %d, channel: %s, auth: %s", proto, version, channel, string(auth))
-		log.Println(err)
-		conn.Close()
-		host.Errors++
-		host.Updated = time.Now()
-		return err
-	}
-	// 更新服务主机信息
-	host.Errors = 0
-	host.Updated = time.Now()
+	// 发送验证信息
+	// 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++
+	// 	host.Updated = time.Now()
+	// 	return err
+	// }
+	// // 接收频道信息
+	// proto, version, channel2, auth, err := conn.ReadAuthInfo()
+	// if err != nil {
+	// 	log.Println("[ConnectToServer] ReadAuthInfo failed:", err)
+	// 	conn.Close()
+	// 	host.Errors++
+	// 	host.Updated = time.Now()
+	// 	return err
+	// }
+	// // 检查版本和协议是否一致
+	// if version != host.Version || proto != rawProto {
+	// 	err = fmt.Errorf("[ConnectToServer] version or protocol wrong: %d, %s", version, proto)
+	// 	log.Println(err)
+	// 	conn.Close()
+	// 	host.Errors++
+	// 	host.Updated = time.Now()
+	// 	return err
+	// }
+	// // 检查频道名称是否匹配
+	// if !strings.Contains(channel2, channel) {
+	// 	err = fmt.Errorf("[ConnectToServer] channel want %s, get %s", channel, channel2)
+	// 	log.Println(err)
+	// 	conn.Close()
+	// 	host.Errors++
+	// 	host.Updated = time.Now()
+	// 	return err
+	// }
+	// // 检查验证是否合法
+	// if !h.checkAuthFunc(true, proto, version, channel, auth) {
+	// 	err = fmt.Errorf("[ConnectToServer] client checkAuthFunc in proto: %s, version: %d, channel: %s, auth: %s", proto, version, channel, string(auth))
+	// 	log.Println(err)
+	// 	conn.Close()
+	// 	host.Errors++
+	// 	host.Updated = time.Now()
+	// 	return err
+	// }
+	// // 更新服务主机信息
+	// host.Errors = 0
+	// host.Updated = time.Now()
 
 	// 将连接加入现有连接中
 	done = false
 	h.lines.Range(func(id int, line *Line) bool {
-		if line.channel == channel {
+		if line.channel == remoteChannel {
 			if line.state == Connected {
 				if force {
 					line.Close(true)
 				} else {
-					err = fmt.Errorf("[connectToServer] channel already connected: %s", channel)
+					err = fmt.Errorf("[connectToServer] channel already connected: %s", remoteChannel)
 					log.Println(err)
 					return false
 				}
 			}
-			line.Start(channel, conn, host)
+			line.Start(remoteChannel, conn, host)
 			done = true
 			return false
 		}
@@ -811,7 +820,7 @@ func (h *Hub) ConnectToServer(channel string, force bool, host *conn.HostInfo, a
 	}
 	// 新建一个连接
 	if !done {
-		line := NewConnect(h.cf, h, channel, conn, host, autoReconnect)
+		line := NewConnect(h.cf, h, remoteChannel, conn, host, autoReconnect)
 		h.addLine(line)
 	}
 	return nil