still codding

conn/tcp2/tcp2.go

@@ -353,7 +353,9 @@ func (c *Tcp2) WritePing(id uint16) error {
 
 // 获取信息
 func (c *Tcp2) ReadMessage(deadline int) (msgType conn.MsgType, id uint16, cmd string, state uint8, data []byte, err error) {
-	msg, recycle, err := c.ReadRawPackage(deadline)
+	var msg []byte
+	var recycle bool
+	msg, recycle, err = c.ReadRawPackage(deadline)
 	if err != nil {
 		return
 	}

conn/ws2/ws2.go

@@ -287,7 +287,9 @@ func (c *Ws2) WritePing(id uint16) error {
 
 // 获取信息
 func (c *Ws2) ReadMessage(deadline int) (msgType conn.MsgType, id uint16, cmd string, state uint8, data []byte, err error) {
-	msg, recycle, err := c.ReadRawPackage(deadline)
+	var msg []byte
+	var recycle bool
+	msg, recycle, err = c.ReadRawPackage(deadline)
 	if err != nil {
 		return
 	}

dail.go

@@ -81,6 +81,7 @@ func DailWithTimeout(ctx context.Context, host *conn.HostInfo, cf *config.Config
 }
 
 // 客户端检验 AuthInfo
+// 如果返回的 channel 有值，说明需要代理连接
 func ClientCheckAuthInfo(
 	conn conn.Connect,
 	host *conn.HostInfo,
@@ -88,40 +89,74 @@ func ClientCheckAuthInfo(
 	remoteChannel string,
 	authFunc AuthFunc,
 	checkAuthFunc CheckAuthFunc,
-) (needProxy bool, err error) {
+) (channel string, err error) {
 	// 发送验证信息
 	if err := conn.WriteAuthInfo(localChannel, authFunc(host, host.Proto, host.Version, remoteChannel, nil)); err != nil {
 		log.Println("[ClientCheckAuthInfo] WriteAuthInfo failed:", err)
-		return false, err
+		return "", err
 	}
 	// 接收频道信息
 	proto, version, channel, auth, err := conn.ReadAuthInfo()
 	if err != nil {
 		log.Println("[ClientCheckAuthInfo] ReadAuthInfo failed:", err)
-		return false, err
+		return "", 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 false, err
+		return "", err
 	}
-	// 检查频道名称是否匹配
-	if !strings.HasPrefix(channel, remoteChannel) {
+	// 检查频道名称是否匹配，只匹配最终的频道
+	if !strings.HasPrefix(channel, strings.Split(remoteChannel, "<-")[0]) {
 		// 代理节点
 		if host.Proxy {
 			// 还需要再建立连接
-			return true, nil
+			return channel, nil
 		}
 		err = fmt.Errorf("[ClientCheckAuthInfo] channel want %s, get %s", remoteChannel, channel)
 		log.Println(err)
-		return false, err
+		return "", err
 	}
 	// 检查验证是否合法
 	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 false, err
+		return "", err
+	}
+	return "", nil
+}
+
+// 服务端验证
+func ServerCheckAuthInfo(
+	conn conn.Connect,
+	host *conn.HostInfo,
+	localChannel string,
+	authFunc AuthFunc,
+	checkAuthFunc CheckAuthFunc,
+) (channel string, err error) {
+	var proto string
+	var version uint8
+	var auth []byte
+	proto, version, channel, auth, err = conn.ReadAuthInfo()
+	if err != nil {
+		log.Println("[BindForServer ReadAuthInfo ERROR]", err)
+		return
+	}
+	if version != host.Version || !strings.HasPrefix(host.Proto, proto) {
+		err = fmt.Errorf("wrong version (%d, %d) or protocol (%s, %s)", version, host.Version, proto, host.Proto)
+		log.Println(err)
+		return
+	}
+	if !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)
+		return
+	}
+	// 发送频道信息
+	if err = conn.WriteAuthInfo(localChannel, authFunc(nil, proto, version, channel, auth)); err != nil {
+		log.Println("[WriteAuthInfo ERROR]", err)
+		return
 	}
 	return
 }

examples/client-tcp2.go

@@ -40,7 +40,7 @@ func main() {
 			return string(auth) == "tinymq-server"
 		},
 		func(conn *tinymq.Line) {
-			log.Println("connect state", conn.Channel(), conn.State(), time.Since(conn.Updated()))
+			log.Println("[Connect state change]", conn.Channel(), conn.State(), time.Since(conn.Started()))
 		},
 	)
 
@@ -74,6 +74,7 @@ func main() {
 	}
 	log.Println("[RESULT]<-", string(rsp.Data))
 
+	log.Println("test finished")
 	time.Sleep(time.Second * 30)
 	log.Println("client exit")
 }

examples/client-ws2-proxy.go

@@ -0,0 +1,120 @@
+//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/ws2/proxy"
+	remoteChannel := "/tinymq/server"
+	remoteFilter := tinymq.StrChannelFilter(remoteChannel)
+
+	hostProxy := &conn.HostInfo{
+		Proto:   "ws",
+		Version: 2,
+		Host:    "127.0.0.1",
+		Port:    44211,
+		// Path:    "/tinymq",
+		Path:  "/tinymq-xor",
+		Hash:  "xor:1qaz2wsx3edc",
+		Proxy: true,
+	}
+
+	hostServer := &conn.HostInfo{
+		Proto:   "ws",
+		Version: 2,
+		Host:    "127.0.0.1",
+		Port:    34211,
+		// Path:    "/tinymq",
+		Path: "/tinymq-xor",
+		Hash: "xor:1qaz2wsx3edc",
+	}
+
+	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.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(remoteChannel, true, nil, false)
+	if err != nil {
+		log.Fatalln("[client ConnectToServer ERROR]", err)
+	}
+
+	log.Println("start get data >>>")
+	count, err := hub.Get(remoteFilter, "hello", "hello in get model", func(response *tinymq.ResponseData) (ok bool) {
+		log.Println("get state and data: ", response.State, string(response.Data))
+		return true
+	})
+	log.Println("end get data with count and err:", count, err)
+
+	// 获取信息
+	rsp := hub.GetOne(remoteFilter, "hello", "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))
+	rsp = hub.GetOne(remoteFilter, "hello", func() ([]byte, error) {
+		return []byte("hello from data function"), nil
+	})
+	if rsp.State != tinymq.STATE_OK {
+		log.Println("error state:", rsp.State)
+		return
+	}
+	log.Println("[RESULT]<-", string(rsp.Data))
+	// 获取长数据
+	rsp = hub.GetOne(remoteFilter, "bigdata", nil)
+	if rsp.State != tinymq.STATE_OK {
+		log.Println("error state:", rsp.State)
+		return
+	} else {
+		log.Println("get bigdata ok")
+	}
+
+	time.Sleep(time.Second * 5)
+	hub.Push(remoteFilter, "push", []byte(time.Now().GoString()))
+
+	log.Println("test finished")
+
+	time.Sleep(time.Second * 30)
+	log.Println("client exit")
+}

examples/client-ws2.go

@@ -43,7 +43,7 @@ func main() {
 			return string(auth) == "tinymq-server"
 		},
 		func(conn *tinymq.Line) {
-			log.Println("connect state", conn.Channel(), conn.State(), time.Since(conn.Updated()))
+			log.Println("[Connect state change]", conn.Channel(), conn.State(), time.Since(conn.Started()))
 		},
 	)
 
@@ -98,6 +98,7 @@ func main() {
 	time.Sleep(time.Second * 5)
 	hub.Push(remoteFilter, "push", []byte(time.Now().GoString()))
 
+	log.Println("test finished")
 	time.Sleep(time.Second * 30)
 	log.Println("client exit")
 }

examples/proxy.go

@@ -0,0 +1,81 @@
+//go:build ignore
+// +build ignore
+
+package main
+
+import (
+	"log"
+	"os"
+	"os/signal"
+	"syscall"
+	"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/proxy"
+
+	var hub *tinymq.Hub
+
+	hub = tinymq.NewHub(
+		cf,
+		localChannel,
+		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-proxy")
+		},
+		func(host *conn.HostInfo, proto string, version uint8, channel string, auth []byte) bool {
+			log.Println("[CheckAuthFunc]", host, proto, version, channel, string(auth))
+			return true
+		},
+		func(conn *tinymq.Line) {
+			log.Println("[Connect state change]", conn.Channel(), conn.State(), time.Since(conn.Started()))
+		},
+	)
+
+	// tcp2协议
+	bindTpv2Info := &conn.HostInfo{
+		Proto:   "tcp",
+		Version: 2,
+		Bind:    "127.0.0.1",
+		Port:    44222,
+		Hash:    "xor:1qaz2wsx3",
+	}
+	hub.BindForServer(bindTpv2Info)
+
+	// ws2协议
+	bindws2Info := &conn.HostInfo{
+		Proto:   "ws",
+		Version: 2,
+		Bind:    "127.0.0.1",
+		Port:    44211,
+		Path:    "/tinymq-xor",
+		Hash:    "xor:1qaz2wsx3edc",
+	}
+	hub.BindForServer(bindws2Info)
+
+	// ws2协议，没有加密算法
+	bindInfo := &conn.HostInfo{
+		Proto:   "ws",
+		Version: 2,
+		// Bind:      "127.0.0.1",
+		Port: 44211,
+		Path: "/tinymq",
+	}
+	hub.BindForServer(bindInfo)
+
+	// 初始化一个channel
+	exit := make(chan os.Signal, 3)
+	//notify方法用来监听收到的信号
+	signal.Notify(exit, os.Interrupt, syscall.SIGINT, syscall.SIGTERM)
+
+	sig := <-exit
+	log.Println("[Exist with]", sig.String())
+}

examples/server.go

@@ -36,7 +36,7 @@ func main() {
 			return string(auth) == "tinymq-client"
 		},
 		func(conn *tinymq.Line) {
-			log.Println("[Connect state change]", conn.Channel(), conn.State(), time.Since(conn.Updated()))
+			log.Println("[Connect state change]", conn.Channel(), conn.State(), time.Since(conn.Started()))
 			if conn.State() == tinymq.StateConnected {
 				go hub.Get(remoteFilter, "hello", []byte("hello from server push"),
 					func(response *tinymq.ResponseData) (ok bool) {

hub.go

@@ -197,7 +197,7 @@ func (h *Hub) AllConnectedChannelWithStarted() []string {
 	cs := make([]string, 0)
 	h.lines.Range(func(id int, line *Line) bool {
 		if line.state == StateConnected {
-			cs = append(cs, fmt.Sprintf("%s|%d", line.channel, line.updated.UnixMilli()))
+			cs = append(cs, fmt.Sprintf("%s|%d", line.channel, line.started.UnixMilli()))
 		}
 		return true
 	})
@@ -337,8 +337,8 @@ func (h *Hub) sendRequest(gd *GetData) (count int, err error) {
 			}
 			err := h.ConnectToServer(channel, false, nil, false)
 			if err != nil {
-				log.Println(err)
-				return 0, err
+				time.Sleep(time.Millisecond * 400)
+				continue
 			}
 		} else {
 			time.Sleep(time.Millisecond * 400) // 故意将时间缩小一点
@@ -608,73 +608,60 @@ func (h *Hub) ConnectDuration(line *Line) time.Duration {
 
 // 绑定端口，建立服务
 // 需要程序运行时调用
-func (h *Hub) BindForServer(info *conn.HostInfo) (err error) {
+func (h *Hub) BindForServer(host *conn.HostInfo) (err error) {
+	// 验证并加入到hub里
 	doConnectFunc := func(conn conn.Connect) {
-		proto, version, channel, auth, err := conn.ReadAuthInfo()
+		channel, err := ServerCheckAuthInfo(conn, host, h.channel, h.authFunc, h.checkAuthFunc)
 		if err != nil {
-			log.Println("[BindForServer ReadAuthInfo ERROR]", err)
-			conn.Close()
-			return
-		}
-		if version != info.Version || proto != info.Proto {
-			log.Println("wrong version or protocol: ", version, proto)
-			conn.Close()
-			return
-		}
-		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(nil, proto, version, channel, auth)); err != nil {
-			log.Println("[WriteAuthInfo ERROR]", err)
+			log.Println("[BindForServer ServerCheckAuthInfo] error:", err)
 			conn.Close()
 			return
 		}
 		// 将连接加入现有连接中
-		done := false
-		h.lines.Range(func(id int, line *Line) bool {
-			if line.state == StateDisconnected && line.host == nil && line.IsChannelEqual(channel) {
-				line.Start(channel, conn, nil)
-				done = true
+		var line *Line
+		h.lines.Range(func(id int, ln *Line) bool {
+			if ln.state == StateDisconnected && ln.host == nil && ln.IsChannelEqual(channel) {
+				line = ln
+				ln.proxyChannel = ""
+				ln.Start(channel, conn, nil)
 				return false
 			}
 			return true
 		})
 		// 新建一个连接
-		if !done {
-			line := NewConnect(h.cf, h, channel, conn, nil, false)
+		if line == nil {
+			line = NewConnect(h.cf, h, channel, "", conn, nil, false)
 			h.addLine(line)
 		}
 	}
-	if info.Version == ws2.VERSION && info.Proto == ws2.PROTO {
+	// 通过不同的协议标识执行不同的库
+	if host.Version == ws2.VERSION && host.Proto == ws2.PROTO {
 		bind := ""
-		if info.Bind != "" {
-			bind = net.JoinHostPort(info.Bind, strconv.Itoa(int(info.Port)))
+		if host.Bind != "" {
+			bind = net.JoinHostPort(host.Bind, strconv.Itoa(int(host.Port)))
 		}
-		return ws2.Server(h.cf, bind, info.Path, info.Hash, doConnectFunc)
-	} else if info.Version == tcp2.VERSION && info.Proto == tcp2.PROTO {
-		return tcp2.Server(h.cf, net.JoinHostPort(info.Bind, strconv.Itoa(int(info.Port))), info.Hash, doConnectFunc)
+		return ws2.Server(h.cf, bind, host.Path, host.Hash, doConnectFunc)
+	} else if host.Version == tcp2.VERSION && host.Proto == tcp2.PROTO {
+		return tcp2.Server(h.cf, net.JoinHostPort(host.Bind, strconv.Itoa(int(host.Port))), host.Hash, doConnectFunc)
 	}
 	return errors.New("not connect protocol and version found")
 }
 
 // 代理连接，在已经连接的基础上进行代理连接
-func (h *Hub) ProxyConnect(connect conn.Connect, remoteChannel string) (err error) {
+func (h *Hub) ClientProxyConnect(line *Line, 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 {
+	if err := line.conn.WriteRawPackage(buf, true); err != nil {
 		log.Println("[ConnectToServer] WriteRawPackage error:", err)
 		return err
 	}
 	// 读取代理结果
-	msgType, _, _, _, data, err := connect.ReadMessage(h.cf.LongReadWait)
+	msgType, _, _, _, data, err := line.conn.ReadMessage(h.cf.LongReadWait)
 	if err != nil {
 		log.Println("[ConnectToServer] ReadRawPackage error:", err)
 		return err
@@ -687,7 +674,7 @@ func (h *Hub) ProxyConnect(connect conn.Connect, remoteChannel string) (err erro
 		return errors.New(string(data))
 	}
 	// 进行auth验证
-	needProxy, err := ClientCheckAuthInfo(connect, host, h.channel, remoteChannel, h.authFunc, h.checkAuthFunc)
+	proxyChannel, err := ClientCheckAuthInfo(line.conn, host, h.channel, remoteChannel, h.authFunc, h.checkAuthFunc)
 	if err != nil {
 		host.Errors++
 		host.Updated = time.Now()
@@ -696,8 +683,8 @@ func (h *Hub) ProxyConnect(connect conn.Connect, remoteChannel string) (err erro
 		host.Errors = 0
 		host.Updated = time.Now()
 	}
-	if needProxy {
-		return errors.New("proxy with proxy")
+	if proxyChannel != "" {
+		return errors.New("proxy with proxy error with channel: " + proxyChannel)
 	}
 	return nil
 }
@@ -711,7 +698,6 @@ func (h *Hub) ConnectToServer(remoteChannel string, force bool, host *conn.HostI
 	if !force {
 		line := h.ChannelToLine(remoteChannel)
 		if line != nil && line.state == StateConnected {
-			// err = fmt.Errorf("[ConnectToServer ERROR] existed channel: %s", channel)
 			log.Println("[ConnectToServer] channel existed:", remoteChannel)
 			return
 		}
@@ -737,7 +723,7 @@ func (h *Hub) ConnectToServer(remoteChannel string, force bool, host *conn.HostI
 		return err
 	}
 
-	needProxy, err := ClientCheckAuthInfo(connect, host, h.channel, remoteChannel, h.authFunc, h.checkAuthFunc)
+	proxyChannel, err := ClientCheckAuthInfo(connect, host, h.channel, remoteChannel, h.authFunc, h.checkAuthFunc)
 	if err != nil {
 		connect.Close()
 		host.Errors++
@@ -748,31 +734,22 @@ 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 {
-		if line.channel == remoteChannel {
-			if line.state == StateConnected {
+	var line *Line
+	h.lines.Range(func(id int, ln *Line) bool {
+		if ln.channel == remoteChannel {
+			if ln.state == StateConnected {
 				if force {
-					line.Close(true)
+					ln.Close(true)
 				} else {
 					err = fmt.Errorf("[connectToServer] channel already connected: %s", remoteChannel)
 					log.Println(err)
 					return false
 				}
 			}
+			line = ln
+			line.proxyChannel = proxyChannel
 			line.Start(remoteChannel, connect, host)
-			done = true
 			return false
 		}
 		return true
@@ -781,8 +758,8 @@ func (h *Hub) ConnectToServer(remoteChannel string, force bool, host *conn.HostI
 		return err
 	}
 	// 新建一个连接
-	if !done {
-		line := NewConnect(h.cf, h, remoteChannel, connect, host, autoReconnect)
+	if line == nil {
+		line = NewConnect(h.cf, h, remoteChannel, proxyChannel, connect, host, autoReconnect)
 		h.addLine(line)
 	}
 	return nil
@@ -846,7 +823,7 @@ func (h *Hub) checkConnect() {
 		case <-proxyTicker.C:
 			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) {
+				if line.host != nil && now-line.lastRead.UnixMilli() > int64(h.cf.ProxyTimeout) {
 					host, err := h.connectHostFunc(line.channel, HostTypeDirect)
 					if err != nil {
 						log.Println("[proxyTicker connectHostFunc]", err)

line.go

@@ -25,8 +25,9 @@ type Line struct {
 	pingID         uint16         // 只有客户端使用
 	pingWrongCount uint8          // 记录 ping id 反馈错误次数，超过3次则重新连接
 
-	proxyConn conn.Connect   // 代理转发连接
-	proxyHost *conn.HostInfo // 代理地址
+	proxyConn    conn.Connect   // 代理转发连接
+	proxyChannel string         // 代理频道
+	proxyHost    *conn.HostInfo // 代理地址
 
 	Extra sync.Map // 附加临时信息，由应用端决定具体内容，断线会自动清理
 	Keep  sync.Map // 附加固定信息，由应用端决定具体内容，断线不会自动清理，不过也不能保证一直保持有值，断线后可能会被系统清理
@@ -40,7 +41,7 @@ type Line struct {
 	lastRead time.Time // 记录最后一次读到数据的时间，在客户端如果超时则重连
 
 	started time.Time // 开始时间
-	updated time.Time // 更新时间
+	// updated time.Time // 更新时间
 }
 
 // 获取当前连接信息
@@ -54,15 +55,25 @@ func (c *Line) Started() time.Time {
 }
 
 // 获取更新时间
-func (c *Line) Updated() time.Time {
-	return c.updated
-}
+// func (c *Line) Updated() time.Time {
+// 	return c.updated
+// }
 
 // 获取当前连接状态
 func (c *Line) State() ConnectState {
 	return c.state
 }
 
+// 设置状态
+func (c *Line) SetState(state ConnectState) {
+	if c.state != state {
+		c.state = state
+		if c.hub != nil && c.hub.connectStatusFunc != nil {
+			c.hub.connectStatusFunc(c)
+		}
+	}
+}
+
 // 获取频道名
 func (c *Line) Channel() string {
 	return c.channel
@@ -132,14 +143,30 @@ func (c *Line) getPingID() uint16 {
 // 读信息循环通道，采用新线程
 func (c *Line) readPump() {
 	for {
-		switch c.state {
-		case StateConnected: // 已经连接好啦，正常通信状态
+		if c.state == StateProxied && c.proxyConn != nil {
+			buf, recycle, err := c.conn.ReadRawPackage(c.cf.LongReadWait)
+			if err != nil {
+				if !strings.Contains(err.Error(), "EOF") {
+					log.Println("[readPump] ReadRawPackage:", err)
+				}
+				c.Close(false)
+				return
+			}
+			c.lastRead = time.Now()
+			// 转发消息到代理端
+			err = c.proxyConn.WriteRawPackage(buf, recycle)
+			if err != nil {
+				log.Println("[readPump] WriteRawPackage:", err)
+				return
+			}
+		} else if c.state == StateClosed || c.state == StateDisconnected {
+			return
+		} else {
 			msgType, id, cmd, state, data, err := c.conn.ReadMessage(c.cf.LongReadWait)
 			if err != nil {
 				if !strings.Contains(err.Error(), "EOF") {
 					log.Println("[readPump]", err)
 				}
-				c.Close(false)
 				return
 			}
 			// 记录最后读到数据的时间
@@ -167,8 +194,8 @@ func (c *Line) readPump() {
 					conn:  c,
 				})
 			case conn.MsgProxyConnect:
-				c.state = StateProxing
-				// TODO: 建立代理连接
+				c.SetState(StateProxing)
+				// 建立代理连接
 				host, err := conn.ParseUrl(string(data))
 				if err != nil {
 					log.Println("[MsgProxyConnect] ParseUrl:", err)
@@ -178,35 +205,47 @@ func (c *Line) readPump() {
 				proxyConn, err := DailWithTimeout(c.hub.ctx, host, c.cf)
 				if err != nil {
 					log.Println("[MsgProxyConnect] DailWithTimeout:", err)
-					// 回退到正常连接状态
-					c.state = StateConnected
 					// 发送失败状态包
-					conn.ProxyResultPackageEncode(err.Error())
-				} else {
-					c.proxyConn = proxyConn
-					c.proxyHost = host
-					c.state = StateProxied
-					conn.ProxyResultPackageEncode("") // not news is good news
+					c.conn.WriteRawPackage(conn.ProxyResultPackageEncode(err.Error()), true)
+					c.Close(false)
+					return
 				}
+				c.proxyConn = proxyConn
+				c.proxyHost = host
+				c.SetState(StateProxied)
+				// not news is good news
+				c.conn.WriteRawPackage(conn.ProxyResultPackageEncode(""), true)
+				go c.proxyDump()
+			default:
+				log.Println("[readPump] unknown msg:", msgType, id, cmd, state, string(data))
 			}
-		case StateProxied: // 已经处于代理状态
-			buf, recycle, err := c.conn.ReadRawPackage(c.cf.LongReadWait)
+		}
+	}
+}
+
+// 代理情况下的消息转发
+// 会一直阻塞
+func (c *Line) proxyDump() {
+	defer c.proxyConn.Close()
+	for {
+		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("[StateProxied] ReadRawPackage:", err)
+					log.Println("[proxyDump] ReadRawPackage:", err)
 				}
 				c.Close(false)
 				return
 			}
 			// 转发消息到代理端
-			err = c.proxyConn.WriteRawPackage(buf, recycle)
+			err = c.conn.WriteRawPackage(buf, recycle)
 			if err != nil {
-				log.Println("[StateProxied] WriteRawPackage:", err)
+				log.Println("[proxyDump] WriteRawPackage:", err)
 				return
 			}
-		case StateClosed:
+		} else if c.state == StateClosed || c.state == StateDisconnected {
 			return
-		default:
+		} else {
 			time.Sleep(time.Second)
 		}
 	}
@@ -225,7 +264,7 @@ func (c *Line) writePump() {
 			go func() {
 				defer func() {
 					if err := recover(); err != nil {
-						c.state = StateDisconnected
+						c.SetState(StateDisconnected)
 						log.Println(err)
 						debug.PrintStack()
 					}
@@ -291,21 +330,22 @@ func (c *Line) writePump() {
 			}
 		case <-pingTicker.C:
 			// 检查是否已经很久时间没有使用连接了
-			dr := time.Since(c.lastRead)
-			if dr > time.Duration(c.cf.PingInterval*3*int(time.Millisecond)) {
+			dr := time.Since(c.lastRead).Milliseconds()
+			if dr > int64(c.cf.PingInterval)*5 {
 				// 超时关闭当前的连接
-				log.Println("[Connect timeout]", c.channel)
-				// 有可能连接出现问题，断开并重新连接
-				// 检测到浏览器会出现休眠的状态，服务端就不主动关闭，只有客户端关闭
-				if c.host != nil {
-					c.Close(false)
-					return
+				if c.cf.PrintMsg {
+					log.Println("[Connect timeout]", c.channel)
 				}
+				// 有可能连接出现问题，断开并重新连接
+				// if c.host != nil || c.proxyHost != nil {
+				c.Close(false)
+				return
+				// }
 			}
 			// 只需要客户端发送
 			if c.host != nil {
 				// 发送ping包
-				if dr >= time.Duration(c.cf.PingInterval*int(time.Millisecond)) {
+				if dr >= int64(c.cf.PingInterval) {
 					id := c.getPingID()
 					// 发送ping数据包
 					if c.cf.PrintPing {
@@ -330,23 +370,19 @@ func (c *Line) writePump() {
 func (c *Line) Close(quick bool) {
 	defer recover() //避免管道已经关闭而引起panic
 	c.conn.Close()
+	if c.proxyConn != nil {
+		c.proxyConn.Close()
+		c.proxyConn = nil
+	}
+	c.proxyChannel = ""
+	c.proxyHost = nil
 	c.closeConnect <- quick
 	if quick {
-		if c.state != StateClosed {
-			c.state = StateClosed
-			c.hub.connectStatusFunc(c)
-		}
+		c.SetState(StateClosed)
 		c.hub.removeLine(c)
-	} else {
-		if c.state != StateDisconnected {
-			c.state = StateDisconnected
-			c.hub.connectStatusFunc(c)
-			go c.hub.cleanDeadConnect()
-		}
-	}
-	if c.proxyConn != nil {
-		c.proxyConn.Close()
-		// c.proxyConn = nil
+	} else if c.state != StateClosed {
+		c.SetState(StateDisconnected)
+		go c.hub.cleanDeadConnect()
 	}
 	c.Extra.Clear()
 	// c.updated = time.Now()
@@ -366,15 +402,28 @@ func (c *Line) cleanClose() {
 // 连接开始运行
 func (c *Line) Start(channel string, conn conn.Connect, host *conn.HostInfo) {
 	now := time.Now()
-	c.updated = now
+	// c.updated = now
 	c.lastRead = now // 避免默认为0时被清理
 	c.channel = channel
 	c.conn = conn
 	c.host = host
-	c.state = StateConnected
-	go c.readPump()
-	go c.writePump()
-	c.hub.connectStatusFunc(c)
+	if c.host != nil && c.proxyChannel != "" {
+		// 进行代理连接
+		c.SetState(StateProxing)
+		err := c.hub.ClientProxyConnect(c, c.channel+"<-"+c.proxyChannel)
+		if err != nil {
+			log.Println("[ProxyConnect] error:", err)
+			c.Close(false)
+		} else {
+			c.SetState(StateProxied)
+		}
+		go c.readPump()
+		go c.writePump()
+	} else {
+		c.SetState(StateConnected)
+		go c.readPump()
+		go c.writePump()
+	}
 }
 
 // 请求入口函数
@@ -382,6 +431,7 @@ func NewConnect(
 	cf *config.Config,
 	hub *Hub,
 	channel string,
+	proxyChannel string,
 	conn conn.Connect,
 	host *conn.HostInfo,
 	autoReconnect bool,
@@ -394,6 +444,8 @@ func NewConnect(
 		pingID:        uint16(now.UnixNano()) % config.ID_MAX,
 		autoReconnect: autoReconnect,
 
+		proxyChannel: proxyChannel,
+
 		sendRequest:  make(chan *RequestData, 32),
 		sendResponse: make(chan *ResponseData, 32),
 		pingRequest:  make(chan *PingData, 5),
@@ -401,7 +453,7 @@ func NewConnect(
 
 		lastRead: now, // 避免默认为0时被清理
 		started:  now,
-		updated:  now,
+		// updated:  now,
 	}
 	cc.Start(channel, conn, host)
 	return cc

mapx.go

@@ -90,13 +90,13 @@ func (m *Mapx) DeleteInvalidLines(expired int64) {
 	m.Lock()
 	defer m.Unlock()
 	for id, line := range m.lineMap {
-		if line.updated.UnixMilli() < expired {
+		if line.lastRead.UnixMilli() < expired {
 			delete(m.lineMap, id)
 		}
 	}
 	for i := len(m.randLines) - 1; i >= 0; i-- {
 		line := m.randLines[i]
-		if line.updated.UnixMilli() < expired {
+		if line.lastRead.UnixMilli() < expired {
 			l := len(m.randLines) - 1
 			m.randLines[i] = m.randLines[l]
 			m.randLines = m.randLines[0:l]