still codding for proxy

dail.go

@@ -33,8 +33,8 @@ func Dail(host *conn.HostInfo, cf *config.Config) (conn conn.Connect, err error)
 	return
 }
 
-// 检验 AuthInfo
-func CheckAuthInfo(
+// 客户端检验 AuthInfo
+func ClientCheckAuthInfo(
 	conn conn.Connect,
 	host *conn.HostInfo,
 	localChannel string,
@@ -42,51 +42,34 @@ func CheckAuthInfo(
 	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()
+		log.Println("[ClientCheckAuthInfo] WriteAuthInfo failed:", err)
 		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()
+		log.Println("[ClientCheckAuthInfo] ReadAuthInfo failed:", err)
 		return err
 	}
 	// 检查版本和协议是否一致
 	if version != host.Version || !strings.HasPrefix(host.Proto, proto) {
-		err = fmt.Errorf("[CheckAuthInfo] version or protocol wrong: %d, %s", version, proto)
+		err = fmt.Errorf("[ClientCheckAuthInfo] 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)
+		err = fmt.Errorf("[ClientCheckAuthInfo] 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))
+		err = fmt.Errorf("[ClientCheckAuthInfo] failed 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/server.go

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

hub.go

@@ -168,10 +168,10 @@ func (h *Hub) ConnectRange(fn func(id int, line *Line) bool) {
 }
 
 // 获取当前在线的数量
-func (h *Hub) ConnectNum() int {
+func (h *Hub) ConnectedNum() int {
 	var count int
 	h.lines.Range(func(id int, line *Line) bool {
-		if line.state == Connected {
+		if line.state == StateConnected {
 			count++
 		}
 		return true
@@ -180,10 +180,10 @@ func (h *Hub) ConnectNum() int {
 }
 
 // 获取所有的在线连接频道
-func (h *Hub) AllChannel() []string {
+func (h *Hub) AllConnectedChannel() []string {
 	cs := make([]string, 0)
 	h.lines.Range(func(id int, line *Line) bool {
-		if line.state == Connected {
+		if line.state == StateConnected {
 			cs = append(cs, line.channel)
 		}
 		return true
@@ -193,10 +193,10 @@ func (h *Hub) AllChannel() []string {
 
 // 获取所有连接频道和连接时长
 // 为了避免定义数据结构麻烦，采用|隔开, 频道名|连接开始时间
-func (h *Hub) AllChannelWithStarted() []string {
+func (h *Hub) AllConnectedChannelWithStarted() []string {
 	cs := make([]string, 0)
 	h.lines.Range(func(id int, line *Line) bool {
-		if line.state == Connected {
+		if line.state == StateConnected {
 			cs = append(cs, fmt.Sprintf("%s|%d", line.channel, line.updated.UnixMilli()))
 		}
 		return true
@@ -205,9 +205,9 @@ func (h *Hub) AllChannelWithStarted() []string {
 }
 
 // 获取频道并通过函数过滤，如果返回 false 将终止
-func (h *Hub) ChannelToFunc(fn func(string) bool) {
+func (h *Hub) ConnectedChannelToFunc(fn func(string) bool) {
 	h.lines.Range(func(id int, line *Line) bool {
-		if line.state == Connected {
+		if line.state == StateConnected {
 			return fn(line.channel)
 		}
 		return true
@@ -255,9 +255,10 @@ func (h *Hub) sendRequest(gd *GetData) (count int, err error) {
 		log.Println(err)
 		return 0, err
 	}
+	// 发送数据到网络
 	doit := func(_ int, line *Line) bool {
 		// 检查连接是否OK
-		if line.state != Connected {
+		if line.state != StateConnected {
 			return true
 		}
 		// 验证连接是否达到发送数据的要求
@@ -635,7 +636,7 @@ func (h *Hub) BindForServer(info *conn.HostInfo) (err error) {
 		// 将连接加入现有连接中
 		done := false
 		h.lines.Range(func(id int, line *Line) bool {
-			if line.state == Disconnected && line.host == nil && line.IsChannelEqual(channel) {
+			if line.state == StateDisconnected && line.host == nil && line.IsChannelEqual(channel) {
 				line.Start(channel, conn, nil)
 				done = true
 				return false
@@ -668,7 +669,7 @@ func (h *Hub) ConnectToServer(remoteChannel string, force bool, host *conn.HostI
 	// 检查当前channel是否已经存在
 	if !force {
 		line := h.ChannelToLine(remoteChannel)
-		if line != nil && line.state == Connected {
+		if line != nil && line.state == StateConnected {
 			// err = fmt.Errorf("[ConnectToServer ERROR] existed channel: %s", channel)
 			log.Println("[ConnectToServer] channel existed:", remoteChannel)
 			return
@@ -684,7 +685,7 @@ func (h *Hub) ConnectToServer(remoteChannel string, force bool, host *conn.HostI
 			return err
 		}
 		if host == nil {
-			// 如果地址为空表示不需要连接，自己返回
+			// 如果地址为空表示不需要连接，直接返回
 			return
 		}
 	}
@@ -738,7 +739,7 @@ func (h *Hub) ConnectToServer(remoteChannel string, force bool, host *conn.HostI
 	// if host.Proxy {
 	// 	localChannel = localChannel + "?proxy=" + host.Host
 	// }
-	err = CheckAuthInfo(conn, host, h.channel, remoteChannel, h.authFunc, h.checkAuthFunc)
+	err = ClientCheckAuthInfo(conn, host, h.channel, remoteChannel, h.authFunc, h.checkAuthFunc)
 	if err != nil {
 		conn.Close()
 		host.Errors++
@@ -748,59 +749,12 @@ func (h *Hub) ConnectToServer(remoteChannel string, force bool, host *conn.HostI
 		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 == remoteChannel {
-			if line.state == Connected {
+			if line.state == StateConnected {
 				if force {
 					line.Close(true)
 				} else {
@@ -890,7 +844,7 @@ func (h *Hub) checkConnect() {
 			})
 		case <-connectTicker.C:
 			h.lines.Range(func(id int, line *Line) bool {
-				if line.autoReconnect && line.host != nil && line.state == Disconnected {
+				if line.autoReconnect && line.host != nil && line.state == StateDisconnected {
 					err := h.ConnectToServer(line.channel, true, nil, true)
 					if err != nil {
 						log.Println("[connectTicker ConnectToServer]", err)
@@ -908,7 +862,7 @@ func (h *Hub) checkConnect() {
 func (h *Hub) Quit() {
 	h.cancel()
 	h.lines.Range(func(id int, line *Line) bool {
-		if line.state == Connected {
+		if line.state == StateConnected {
 			line.Close(true)
 		}
 		return true

line.go

@@ -109,20 +109,15 @@ func (c *Line) IsChannelEqual(channel string) bool {
 
 // 获取远程的地址
 func (c *Line) RemoteIP() net.IP {
-	if c.state == Connected {
-		return c.conn.RemoteIP()
-	} else {
-		return nil
-	}
+	defer recover() //避免连接已经关闭而引起panic
+	return c.conn.RemoteIP()
+
 }
 
 // 获取本地的地址
 func (c *Line) LocalIP() net.IP {
-	if c.state == Connected {
-		return c.conn.LocalIP()
-	} else {
-		return nil
-	}
+	defer recover() //避免连接已经关闭而引起panic
+	return c.conn.LocalIP()
 }
 
 // 获取通讯消息ID号
@@ -137,38 +132,62 @@ func (c *Line) getPingID() uint16 {
 // 读信息循环通道，采用新线程
 func (c *Line) readPump() {
 	for {
-		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)
+		switch c.state {
+		case StateConnected: // 已经连接好啦，正常通信状态
+			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
 			}
-			c.Close(false)
-			return
-		}
-		// 记录最后读到数据的时间
-		c.lastRead = time.Now()
-		switch msgType {
-		case conn.MsgPing:
-			// ping或pong包
-			c.pingRequest <- &PingData{
-				Id: id,
+			// 记录最后读到数据的时间
+			c.lastRead = time.Now()
+			switch msgType {
+			case conn.MsgPing:
+				// ping或pong包
+				c.pingRequest <- &PingData{
+					Id: id,
+				}
+			case conn.MsgRequest:
+				// 请求数据包
+				go c.hub.requestFromNet(&RequestData{
+					Id:   id,
+					Cmd:  cmd,
+					Data: data,
+					conn: c,
+				})
+			case conn.MsgResponse:
+				// 网络回应数据包
+				go c.hub.outResponse(&ResponseData{
+					Id:    id,
+					State: state & 0x7F,
+					Data:  data,
+					conn:  c,
+				})
+			case conn.MsgProxyConnect:
+				// TODO: 建立代理连接
+			}
+		case StateProxied: // 已经处于代理状态
+			buf, recycle, err := c.conn.ReadRawPackage(c.cf.LongReadWait)
+			if err != nil {
+				if !strings.Contains(err.Error(), "EOF") {
+					log.Println("[StateProxied] ReadRawPackage:", err)
+				}
+				c.Close(false)
+				return
 			}
-		case conn.MsgRequest:
-			// 请求数据包
-			go c.hub.requestFromNet(&RequestData{
-				Id:   id,
-				Cmd:  cmd,
-				Data: data,
-				conn: c,
-			})
-		case conn.MsgResponse:
-			// 网络回应数据包
-			go c.hub.outResponse(&ResponseData{
-				Id:    id,
-				State: state & 0x7F,
-				Data:  data,
-				conn:  c,
-			})
+			// 转发消息到代理端
+			err = c.proxyConn.WriteRawPackage(buf, recycle)
+			if err != nil {
+				log.Println("[StateProxied] WriteRawPackage:", err)
+				return
+			}
+		case StateClosed:
+			return
+		default:
+			time.Sleep(time.Second)
 		}
 	}
 }
@@ -182,11 +201,11 @@ func (c *Line) writePump() {
 		pingTicker.Stop()
 		c.conn.Close()
 		// 检查是否需要重新连接
-		if c.autoReconnect && c.host != nil && c.state != Closed && c.state != Connected {
+		if c.autoReconnect && c.host != nil && c.state != StateClosed && c.state != StateConnected {
 			go func() {
 				defer func() {
 					if err := recover(); err != nil {
-						c.state = Disconnected
+						c.state = StateDisconnected
 						log.Println(err)
 						debug.PrintStack()
 					}
@@ -293,20 +312,20 @@ func (c *Line) Close(quick bool) {
 	c.conn.Close()
 	c.closeConnect <- quick
 	if quick {
-		if c.state != Closed {
-			c.state = Closed
+		if c.state != StateClosed {
+			c.state = StateClosed
 			c.hub.connectStatusFunc(c)
 		}
 		c.hub.removeLine(c)
 	} else {
-		if c.state != Disconnected {
-			c.state = Disconnected
+		if c.state != StateDisconnected {
+			c.state = StateDisconnected
 			c.hub.connectStatusFunc(c)
 			go c.hub.cleanDeadConnect()
 		}
 	}
 	c.Extra.Clear()
-	c.updated = time.Now()
+	// c.updated = time.Now()
 }
 
 // 清空余留下来的管道消息
@@ -328,7 +347,7 @@ func (c *Line) Start(channel string, conn conn.Connect, host *conn.HostInfo) {
 	c.channel = channel
 	c.conn = conn
 	c.host = host
-	c.state = Connected
+	c.state = StateConnected
 	go c.readPump()
 	go c.writePump()
 	c.hub.connectStatusFunc(c)

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.state != Connected && line.updated.UnixMilli() < expired {
+		if line.updated.UnixMilli() < expired {
 			delete(m.lineMap, id)
 		}
 	}
 	for i := len(m.randLines) - 1; i >= 0; i-- {
 		line := m.randLines[i]
-		if line.state != Connected && line.updated.UnixMilli() < expired {
+		if line.updated.UnixMilli() < expired {
 			l := len(m.randLines) - 1
 			m.randLines[i] = m.randLines[l]
 			m.randLines = m.randLines[0:l]

type.go

@@ -49,24 +49,24 @@ type GetData struct {
 type ConnectState byte
 
 const (
-	Disconnected ConnectState = iota
-	Connected
-	Proxing // 正在进行Proxy请求
-	Proxied // Proxy 连接成功状态
-	Closed
+	StateDisconnected ConnectState = iota
+	StateConnected
+	StateProxing // 正在进行Proxy请求
+	StateProxied // Proxy 连接成功状态
+	StateClosed  // 已经被关闭的连接
 )
 
 func (t ConnectState) String() string {
 	switch t {
-	case Disconnected:
+	case StateDisconnected:
 		return "Disconnected"
-	case Connected:
+	case StateConnected:
 		return "Connected"
-	case Proxing:
+	case StateProxing:
 		return "Proxing"
-	case Proxied:
+	case StateProxied:
 		return "Proxied"
-	case Closed:
+	case StateClosed:
 		return "Closed"
 	default:
 		return fmt.Sprintf("Unknown ConnectState (%d)", t)