tinymq/line.go

package tinymq

import (
	"log"
	"net"
	"runtime/debug"
	"strings"
	"sync"
	"time"

	"git.me9.top/git/tinymq/config"
	"git.me9.top/git/tinymq/conn"
)

// 建立一个虚拟连接，包括生成命令，发送命令和响应命令

type Line struct {
	cf             *config.Config
	hub            *Hub
	conn           conn.Connect
	channel        string         // 连接对端的频道
	state          ConnectState   // 连接状态
	host           *conn.HostInfo // 如果有值说明是客户端，就是主动连接端
	autoReconnect  bool           // 指示是否自动重连
	pingID         uint16         // 只有客户端使用
	pingWrongCount uint8          // 记录 ping id 反馈错误次数，超过3次则重新连接

	proxyConn    conn.Connect   // 代理转发连接
	proxyChannel string         // 代理频道
	proxyHost    *conn.HostInfo // 代理地址

	Extra sync.Map // 附加临时信息，由应用端决定具体内容，断线会自动清理
	Keep  sync.Map // 附加固定信息，由应用端决定具体内容，断线不会自动清理，不过也不能保证一直保持有值，断线后可能会被系统清理

	// 当前连接的管道
	sendRequest  chan *RequestData  // 发送请求数据
	sendResponse chan *ResponseData // 发送回应包
	pingRequest  chan *PingData     // Ping请求
	closeConnect chan bool          // 关闭连接信号，true表示外部进行关闭，false表示连接已经出问题或超时关闭

	lastRead time.Time // 记录最后一次读到数据的时间，在客户端如果超时则重连

	started time.Time // 开始时间
	// updated time.Time // 更新时间
}

// 获取当前连接信息
func (c *Line) Host() *conn.HostInfo {
	return c.host
}

// 获取开始时间
func (c *Line) Started() time.Time {
	return c.started
}

// 获取更新时间
// 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
}

// 设置频道名
// 检查是否包含@，只替换@前面部分
func (c *Line) SetChannelName(name string) {
	if strings.Contains(name, "@") {
		c.channel = name
	} else {
		if inx := strings.Index(c.channel, "@"); inx >= 0 {
			c.channel = name + c.channel[inx:]
		} else {
			c.channel = name + "@" + c.channel
		}
	}
}

// 删除频道名，用于退出登录等操作
func (c *Line) RemoveChannelName() {
	if inx := strings.Index(c.channel, "@"); inx >= 0 {
		c.channel = c.channel[inx+1:]
	}
}

// 频道是否相等，不包括@前面部分和参数部分
func (c *Line) IsChannelEqual(channel string) bool {
	if inx := strings.Index(channel, "@"); inx >= 0 {
		channel = channel[inx+1:]
	}
	if inx := strings.Index(channel, "?"); inx > 0 {
		channel = channel[0:inx]
	}
	origin := c.channel
	if inx := strings.Index(origin, "@"); inx >= 0 {
		origin = origin[inx+1:]
	}
	if inx := strings.Index(origin, "?"); inx > 0 {
		origin = origin[0:inx]
	}
	return channel == origin
}

// 获取远程的地址
func (c *Line) RemoteIP() net.IP {
	defer recover() //避免连接已经关闭而引起panic
	return c.conn.RemoteIP()

}

// 获取本地的地址
func (c *Line) LocalIP() net.IP {
	defer recover() //避免连接已经关闭而引起panic
	return c.conn.LocalIP()
}

// 获取通讯消息ID号
func (c *Line) getPingID() uint16 {
	c.pingID++
	if c.pingID <= 0 || c.pingID >= config.ID_MAX {
		c.pingID = 1
	}
	return c.pingID
}

// 读信息循环通道，采用新线程
func (c *Line) readPump() {
	for {
		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 {
				c.Close(false)
				if !strings.Contains(err.Error(), "EOF") {
					log.Println("[readPump]", err)
				}
				return
			}
			// 记录最后读到数据的时间
			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:
				c.SetState(StateProxing)
				// 建立代理连接
				host, err := conn.ParseUrl(string(data))
				if err != nil {
					log.Println("[MsgProxyConnect] ParseUrl:", err)
					c.Close(false)
					return
				}
				proxyConn, err := DailWithTimeout(c.hub.ctx, host, c.cf)
				if err != nil {
					log.Println("[MsgProxyConnect] DailWithTimeout:", err)
					// 发送失败状态包
					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 error:", msgType, id, cmd, state, string(data))
			}
		}
	}
}

// 代理情况下的消息转发
// 会一直阻塞
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 {
				c.Close(false)
				return
			}
			// 转发消息到代理端
			err = c.conn.WriteRawPackage(buf, recycle)
			if err != nil {
				log.Println("[proxyDump] WriteRawPackage:", err)
				return
			}
		} else if c.state == StateClosed || c.state == StateDisconnected {
			return
		} else {
			time.Sleep(time.Second)
		}
	}
}

// 检查管道并处理不同的消息，新go程调用
// 为了防止多线程的冲突，主要的处理都在这里进行
func (c *Line) writePump() {
	pingTicker := time.NewTicker(time.Duration(c.cf.PingInterval * int(time.Millisecond)))
	// 定义恢复函数
	defer func() {
		pingTicker.Stop()
		c.conn.Close()
		// 检查是否需要重新连接
		if c.autoReconnect && c.host != nil && c.state != StateClosed && c.state != StateConnected {
			go func() {
				defer func() {
					if err := recover(); err != nil {
						c.SetState(StateDisconnected)
						log.Println(err)
						debug.PrintStack()
					}
				}()
				c.host.Errors++
				c.host.Updated = time.Now()

				delay := 5 * time.Second * time.Duration(c.host.Errors)
				log.Println("[Line] will reconnect with delay:", delay)
				time.Sleep(delay)
				if c.host.Errors < 2 {
					c.hub.ConnectToServerX(c.channel, false, c.host)
				} else {
					c.hub.ConnectToServerX(c.channel, false, nil)
				}
			}()
		}
	}()
	// 清空closeConnect
	c.cleanClose()
	// 开始处理信息循环
	for {
		select {
		case request := <-c.sendRequest: // 发送请求包
			err := c.conn.WriteRequest(request.Id, request.Cmd, request.Data)
			if err != nil {
				log.Println("[writePump WriteRequest]", err)
				return
			}
		case response := <-c.sendResponse: // 接收到的响应包
			// 发送响应数据
			err := c.conn.WriteResponse(response.Id, response.State, response.Data)
			if err != nil {
				log.Println("[writePump WriteResponse]", err)
				return
			}
		case ping := <-c.pingRequest: // 发送ping包到网络
			if c.cf.PrintPing {
				log.Println("[PING]<-", c.channel, ping.Id)
			}
			// 只有服务器端需要回应ping包
			if c.host == nil {
				if c.cf.PrintPing {
					log.Println("[RESP PING]->", c.channel, ping.Id)
				}
				err := c.conn.WritePing(ping.Id)
				if err != nil {
					log.Println("[ping] failed", err)
					return
				}
			} else {
				// 检查 ping id 是否正确
				if c.pingID == ping.Id {
					c.pingWrongCount = 0
				} else {
					c.pingWrongCount++
					if c.pingWrongCount > 3 {
						log.Println("[wrong ping id]", ping.Id)
						c.Close(false)
						return
					}
				}
			}
		case <-pingTicker.C:
			// 检查是否已经很久时间没有使用连接了
			dr := time.Since(c.lastRead).Milliseconds()
			if dr > int64(c.cf.PingInterval)*5 {
				// 超时关闭当前的连接
				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 >= int64(c.cf.PingInterval) {
					id := c.getPingID()
					// 发送ping数据包
					if c.cf.PrintPing {
						log.Println("[SEND PING]->", c.channel, id)
					}
					err := c.conn.WritePing(id)
					if err != nil {
						log.Println("[writePump WritePing]", err)
						return
					}
				}
			}
		case <-c.closeConnect:
			c.cleanClose()
			// 退出循环
			return
		}
	}
}

// 关闭连接
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 {
		c.SetState(StateClosed)
		c.hub.removeLine(c)
	} else if c.state != StateClosed {
		c.SetState(StateDisconnected)
		go c.hub.cleanDeadConnect()
	}
	c.Extra.Clear()
	// c.updated = time.Now()
}

// 清空余留下来的管道消息
func (c *Line) cleanClose() {
	for {
		select {
		case <-c.closeConnect:
		default:
			return
		}
	}
}

// 连接开始运行
func (c *Line) Start(channel string, conn conn.Connect, host *conn.HostInfo) {
	now := time.Now()
	// c.updated = now
	c.lastRead = now // 避免默认为0时被清理
	c.channel = channel
	c.conn = conn
	c.host = host
	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()
	}
}

// 请求入口函数
func NewConnect(
	cf *config.Config,
	hub *Hub,
	channel string,
	proxyChannel string,
	conn conn.Connect,
	host *conn.HostInfo,
	autoReconnect bool,
) *Line {
	now := time.Now()
	cc := &Line{
		cf:            cf,
		channel:       channel,
		hub:           hub,
		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),
		closeConnect: make(chan bool, 5),

		lastRead: now, // 避免默认为0时被清理
		started:  now,
		// updated:  now,
	}
	cc.Start(channel, conn, host)
	return cc
}