finish proxy code

examples/client-tcp2-proxy.go

@@ -0,0 +1,93 @@
+//go:build ignore
+// +build ignore
+
+package main
+
+import (
+	"log"
+	"time"
+
+	"git.me9.top/git/tinymq"
+	"git.me9.top/git/tinymq/config"
+	"git.me9.top/git/tinymq/conn"
+)
+
+func main() {
+	cf := config.NewConfig()
+	localChannel := "/tinymq/client/tcp2/proxy"
+	remoteChannel := "/tinymq/server"
+	remoteFilter := tinymq.StrChannelFilter(remoteChannel)
+	hostProxy := &conn.HostInfo{
+		Proto:   "tcp",
+		Version: 2,
+		Host:    "127.0.0.1",
+		Port:    44222,
+		Hash:    "xor:1qaz2wsx3",
+		Proxy:   true,
+	}
+
+	hostServer := &conn.HostInfo{
+		Proto:   "tcp",
+		Version: 2,
+		Host:    "127.0.0.1",
+		Port:    34222,
+		Hash:    "xor:1qaz2wsx3",
+	}
+
+	hub := tinymq.NewHub(
+		cf,
+		localChannel,
+		func(channel string, hostType tinymq.HostType) (hostInfo *conn.HostInfo, err error) {
+			log.Println("[connectHostFunc]", channel, hostType)
+			if hostType == tinymq.HostTypeDirect {
+				return hostServer, nil
+			}
+			return hostProxy, nil
+		},
+		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(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) {
+			log.Println("[Connect state change]", conn.Channel(), conn.State(), time.Since(conn.Started()))
+		},
+	)
+
+	// 从过滤器到频道字符串
+	hub.SetFilterToChannelFunc(func(filter tinymq.FilterFunc) (channel string) {
+		return remoteChannel
+	})
+
+	// 订阅频道
+	hub.Subscribe(remoteFilter, "hello", func(request *tinymq.RequestData) (state uint8, result any) {
+		log.Println("[client RECV]<-", string(request.Data))
+		return 1, "tiny client"
+	},
+	)
+	hub.Subscribe(remoteFilter, "nodata", func(request *tinymq.RequestData) (state uint8, result any) {
+		log.Println("[client RECV]<-", string(request.Data))
+		return 1, nil
+	},
+	)
+
+	// err := hub.ConnectToServer("/tinymq/server", true, nil)
+	// if err != nil {
+	// 	log.Fatalln("[client ConnectToServer ERROR]", err)
+	// }
+
+	// 获取信息
+	rsp := hub.GetOne(remoteFilter, "hello", []byte("hello from client, hello from client, hello from client"))
+	if rsp.State != tinymq.STATE_OK {
+		log.Println("error state:", rsp.State)
+		return
+	}
+	log.Println("[RESULT]<-", string(rsp.Data))
+
+	log.Println("test finished")
+	time.Sleep(time.Second * 10)
+	log.Println("client exit")
+}

examples/client-tcp2.go

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

examples/client-ws2-proxy.go

@@ -115,6 +115,6 @@ func main() {
 
 	log.Println("test finished")
 
-	time.Sleep(time.Second * 30)
+	time.Sleep(time.Second * 10)
 	log.Println("client exit")
 }

examples/client-ws2.go

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

hub.go

@@ -168,10 +168,10 @@ func (h *Hub) ConnectRange(fn func(id int, line *Line) bool) {
 }
 
 // 获取当前在线的数量
-func (h *Hub) ConnectedNum() int {
+func (h *Hub) ConnectNum(state ConnectState) int {
 	var count int
 	h.lines.Range(func(id int, line *Line) bool {
-		if line.state == StateConnected {
+		if line.state == state {
 			count++
 		}
 		return true
@@ -180,10 +180,10 @@ func (h *Hub) ConnectedNum() int {
 }
 
 // 获取所有的在线连接频道
-func (h *Hub) AllConnectedChannel() []string {
+func (h *Hub) AllConnectChannel(state ConnectState) []string {
 	cs := make([]string, 0)
 	h.lines.Range(func(id int, line *Line) bool {
-		if line.state == StateConnected {
+		if line.state == state {
 			cs = append(cs, line.channel)
 		}
 		return true
@@ -193,10 +193,10 @@ func (h *Hub) AllConnectedChannel() []string {
 
 // 获取所有连接频道和连接时长
 // 为了避免定义数据结构麻烦，采用|隔开, 频道名|连接开始时间
-func (h *Hub) AllConnectedChannelWithStarted() []string {
+func (h *Hub) AllConnectChannelWithStarted(state ConnectState) []string {
 	cs := make([]string, 0)
 	h.lines.Range(func(id int, line *Line) bool {
-		if line.state == StateConnected {
+		if line.state == state {
 			cs = append(cs, fmt.Sprintf("%s|%d", line.channel, line.started.UnixMilli()))
 		}
 		return true
@@ -205,9 +205,9 @@ func (h *Hub) AllConnectedChannelWithStarted() []string {
 }
 
 // 获取频道并通过函数过滤，如果返回 false 将终止
-func (h *Hub) ConnectedChannelToFunc(fn func(string) bool) {
+func (h *Hub) ConnectChannelToFunc(state ConnectState, fn func(string) bool) {
 	h.lines.Range(func(id int, line *Line) bool {
-		if line.state == StateConnected {
+		if line.state == state {
 			return fn(line.channel)
 		}
 		return true
@@ -258,7 +258,7 @@ func (h *Hub) sendRequest(gd *GetData) (count int, err error) {
 	// 发送数据到网络
 	doit := func(_ int, line *Line) bool {
 		// 检查连接是否OK
-		if line.state != StateConnected {
+		if line.state != StateConnected && line.state != StateProxied {
 			return true
 		}
 		// 验证连接是否达到发送数据的要求
@@ -284,7 +284,7 @@ func (h *Hub) sendRequest(gd *GetData) (count int, err error) {
 						// 检查是否已经很久时间没有使用连接了
 						if time.Since(conn.lastRead) > time.Duration(h.cf.PingInterval*3*int(time.Millisecond)) {
 							// 超时关闭当前的连接
-							log.Println("get message timeout", conn.channel)
+							log.Println("get message timeout error:", conn.channel)
 							// 有可能连接出现问题，断开并重新连接
 							conn.Close(false)
 							return
@@ -697,7 +697,7 @@ func (h *Hub) ConnectToServer(remoteChannel string, force bool, host *conn.HostI
 	// 检查当前channel是否已经存在
 	if !force {
 		line := h.ChannelToLine(remoteChannel)
-		if line != nil && line.state == StateConnected {
+		if line != nil && (line.state == StateConnected || line.state == StateProxied) {
 			log.Println("[ConnectToServer] channel existed:", remoteChannel)
 			return
 		}
@@ -858,7 +858,7 @@ func (h *Hub) checkConnect() {
 func (h *Hub) Quit() {
 	h.cancel()
 	h.lines.Range(func(id int, line *Line) bool {
-		if line.state == StateConnected {
+		if line.state == StateConnected || line.state == StateProxied {
 			line.Close(true)
 		}
 		return true

line.go

@@ -164,6 +164,7 @@ func (c *Line) readPump() {
 		} else {
 			msgType, id, cmd, state, data, err := c.conn.ReadMessage(c.cf.LongReadWait)
 			if err != nil {
+				c.Close(false)
 				if !strings.Contains(err.Error(), "EOF") {
 					log.Println("[readPump]", err)
 				}
@@ -217,7 +218,7 @@ func (c *Line) readPump() {
 				c.conn.WriteRawPackage(conn.ProxyResultPackageEncode(""), true)
 				go c.proxyDump()
 			default:
-				log.Println("[readPump] unknown msg:", msgType, id, cmd, state, string(data))
+				log.Println("[readPump] unknown msg error:", msgType, id, cmd, state, string(data))
 			}
 		}
 	}
@@ -231,9 +232,6 @@ func (c *Line) proxyDump() {
 		if c.state == StateProxied && c.proxyConn != nil {
 			buf, recycle, err := c.proxyConn.ReadRawPackage(c.cf.LongReadWait)
 			if err != nil {
-				if !strings.Contains(err.Error(), "EOF") {
-					log.Println("[proxyDump] ReadRawPackage:", err)
-				}
 				c.Close(false)
 				return
 			}