tinymq/hub.go

package tinymq

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"log"
	"math/rand"
	"net"

	// "regexp"
	"strconv"
	"strings"
	"sync"
	"time"

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

// 类似一个插座的功能，管理多个连接
// 一个hub即可以是客户端，同时也可以是服务端
// 为了简化流程和让通讯更加迅速，不再重发和缓存结果，采用超时的方式告诉应用层。

// 截取部分字符串
func subStr(str string, length int) string {
	if len(str) <= length {
		return str
	}
	return str[0:length] + "..."
}

type Hub struct {
	sync.Mutex
	ctx        context.Context // 为了方便退出而建立
	cancel     context.CancelFunc
	cf         *config.Config
	globalID   uint16
	channel    string       // 本地频道信息
	middle     []MiddleFunc // 中间件
	lines      *Mapx        // 记录当前的连接，统一管理
	subscribes sync.Map     // [cmd]->[]*SubscribeData   //注册绑定频道的函数，用于响应请求
	msgCache   sync.Map     //  map[uint16]*GetMsg //请求的回应记录，key为id

	// 客户端需要用的函数（服务端可为空）
	connectHostFunc ConnectHostFunc // 获取对应频道的一个连接地址

	// 返回认证信息，发送到对方
	authFunc AuthFunc // 获取认证信息，用于发送给对方

	// 核对发送过来的认证信息
	checkAuthFunc CheckAuthFunc // 核对认证是否合法

	// 连接状态变化时调用的函数
	connectStatusFunc ConnectStatusFunc

	// 验证发送数据的条件是否满足 (可空)
	checkConnectOkFunc CheckConnectOkFunc

	// 通过过滤函数获取一个频道信息 (可空)
	filterToChannelFunc FilterToChannelFunc

	// 上次清理异常连接时间戳
	lastCleanDeadConnect int64
}

func (h *Hub) SetCheckConnectOkFunc(fn CheckConnectOkFunc) {
	h.checkConnectOkFunc = fn
}

func (h *Hub) SetFilterToChannelFunc(fn FilterToChannelFunc) {
	h.filterToChannelFunc = fn
}

// 转换数据
func (h *Hub) convertData(data any) (reqData []byte, err error) {
	switch data := data.(type) {
	case []byte:
		reqData = data
	case string:
		reqData = []byte(data)
	case func() ([]byte, error):
		reqData, err = data()
		if err != nil {
			log.Println(err.Error())
			return nil, err
		}
	default:
		if data != nil {
			// 自动转换数据为json格式
			reqData, err = json.Marshal(data)
			if err != nil {
				log.Println(err.Error())
				return nil, err
			}
		}
	}
	return
}

// 清理异常连接
func (h *Hub) cleanDeadConnect() {
	now := time.Now().UnixMilli()
	expired := now - int64(h.cf.CleanDeadConnectWait)
	if h.lastCleanDeadConnect < expired {
		h.lastCleanDeadConnect = now
		h.lines.DeleteInvalidLines(expired)
	}
}

// 获取通讯消息ID号
func (h *Hub) GetID() uint16 {
	h.Lock()
	defer h.Unlock()
	h.globalID++
	if h.globalID <= 0 || h.globalID >= config.ID_MAX {
		h.globalID = 1
	}
	for {
		// 检查是否在请求队列中存在对应的id
		if _, ok := h.msgCache.Load(h.globalID); ok {
			h.globalID++
			if h.globalID <= 0 || h.globalID >= config.ID_MAX {
				h.globalID = 1
			}
		} else {
			break
		}
	}
	return h.globalID
}

// 添加中间件
// 如果中间件函数返回为空，表示处理完成，通过
// 如果中间件函数返回 NEXT_MIDDLE，表示需要下一个中间件函数处理；如果没有下一函数则默认通过
func (h *Hub) UseMiddle(middleFunc MiddleFunc) {
	h.middle = append(h.middle, middleFunc)
}

// 注册频道，其中频道为正则表达式字符串
func (h *Hub) Subscribe(filter FilterFunc, cmd string, backFunc SubscribeBackFunc) (err error) {
	if filter == nil {
		return errors.New("filter function can not be nil")
	}
	reg := &SubscribeData{
		Filter:   filter,
		Cmd:      cmd,
		BackFunc: backFunc,
	}
	sub, ok := h.subscribes.Load(cmd)
	if ok {
		h.subscribes.Store(cmd, append(sub.([]*SubscribeData), reg))
		return
	}
	regs := make([]*SubscribeData, 1)
	regs[0] = reg
	h.subscribes.Store(cmd, regs)
	return
}

// 遍历频道列表
// 如果 fn 返回 false，则 range 停止迭代
func (h *Hub) ConnectRange(fn func(id int, line *Line) bool) {
	h.lines.Range(fn)
}

// 获取当前在线的数量
func (h *Hub) ConnectNum() int {
	var count int
	h.lines.Range(func(id int, line *Line) bool {
		if line.state == Connected {
			count++
		}
		return true
	})
	return count
}

// 获取所有的在线连接频道
func (h *Hub) AllChannel() []string {
	cs := make([]string, 0)
	h.lines.Range(func(id int, line *Line) bool {
		if line.state == Connected {
			cs = append(cs, line.channel)
		}
		return true
	})
	return cs
}

// 获取所有连接频道和连接时长
// 为了避免定义数据结构麻烦，采用|隔开, 频道名|连接开始时间
func (h *Hub) AllChannelWithStarted() []string {
	cs := make([]string, 0)
	h.lines.Range(func(id int, line *Line) bool {
		if line.state == Connected {
			cs = append(cs, fmt.Sprintf("%s|%d", line.channel, line.updated.UnixMilli()))
		}
		return true
	})
	return cs
}

// 获取频道并通过函数过滤，如果返回 false 将终止
func (h *Hub) ChannelToFunc(fn func(string) bool) {
	h.lines.Range(func(id int, line *Line) bool {
		if line.state == Connected {
			return fn(line.channel)
		}
		return true
	})
}

// 从 channel 获取连接
func (h *Hub) ChannelToLine(channel string) (line *Line) {
	h.lines.Range(func(id int, l *Line) bool {
		if l.IsChannelEqual(channel) {
			line = l
			return false
		}
		return true
	})
	return
}

// 返回请求结果
func (h *Hub) outResponse(response *ResponseData) {
	defer recover() //避免管道已经关闭而引起panic
	id := response.Id
	t, ok := h.msgCache.Load(id)
	if ok {
		// 删除数据缓存
		h.msgCache.Delete(id)
		gm := t.(*GetMsg)
		// 停止定时器
		if !gm.timer.Stop() {
			select {
			case <-gm.timer.C:
			default:
			}
		}
		// 回应数据到上层
		gm.out <- response
	}
}

// 发送数据到网络接口
// 返回发送的数量
func (h *Hub) sendRequest(gd *GetData) (count int, err error) {
	outData, err := h.convertData(gd.Data)
	if err != nil {
		log.Println(err)
		return 0, err
	}
	doit := func(_ int, line *Line) bool {
		// 检查连接是否OK
		if line.state != Connected {
			return true
		}
		// 验证连接是否达到发送数据的要求
		if h.checkConnectOkFunc != nil && !h.checkConnectOkFunc(line, gd) {
			return true
		}
		if gd.Filter(line) {
			var id uint16
			if gd.backchan != nil {
				id = h.GetID()
				timeout := gd.Timeout
				if timeout <= 0 {
					timeout = h.cf.WriteWait
				}
				fn := func(id uint16, conn *Line) func() {
					return func() {
						go h.outResponse(&ResponseData{
							Id:    id,
							State: config.GET_TIMEOUT,
							Data:  fmt.Appendf(nil, "[%s] %s %s", IdMsg(config.GET_TIMEOUT), conn.channel, gd.Cmd),
							conn:  conn,
						})
						// 检查是否已经很久时间没有使用连接了
						if time.Since(conn.lastRead) > time.Duration(h.cf.PingInterval*3*int(time.Millisecond)) {
							// 超时关闭当前的连接
							log.Println("get message timeout", conn.channel)
							// 有可能连接出现问题，断开并重新连接
							conn.Close(false)
							return
						}
					}
				}(id, line)
				// 将要发送的请求缓存
				gm := &GetMsg{
					out:   gd.backchan,
					timer: time.AfterFunc(time.Millisecond*time.Duration(timeout), fn),
				}
				h.msgCache.Store(id, gm)
			}
			// 组织数据并发送到Connect
			line.sendRequest <- &RequestData{
				Id:       id,
				Cmd:      gd.Cmd,
				Data:     outData,
				timeout:  gd.Timeout,
				backchan: gd.backchan,
				conn:     line,
			}
			if h.cf.PrintMsg {
				log.Println("[SEND]->", id, line.channel, "["+gd.Cmd+"]", subStr(string(outData), 200))
			}
			count++
			if gd.Max > 0 && count >= gd.Max {
				return false
			}
		}
		return true
	}
	// 如果没有发送到消息，延时重连直到超时
	for i := 0; i <= gd.Timeout; i += 500 {
		if gd.Rand {
			h.lines.RandRange(doit, i == 0)
		} else {
			h.lines.Range(doit)
		}
		if count > 0 {
			break
		}
		// 如果是客户端，并且有机会自动连接，则尝试自动连接
		if i == 0 && h.connectHostFunc != nil && h.filterToChannelFunc != nil {
			channel := h.filterToChannelFunc(gd.Filter)
			if channel == "" {
				err = errors.New(IdMsg(NO_MATCH_CONNECT))
				log.Println("not channel found")
				return 0, err
			}
			err := h.ConnectToServer(channel, false, nil, false)
			if err != nil {
				log.Println(err)
				return 0, err
			}
		} else {
			time.Sleep(time.Millisecond * 400) // 故意将时间缩小一点
		}
	}
	return
}

// 执行网络发送过来的命令
func (h *Hub) requestFromNet(request *RequestData) {
	cmd := request.Cmd
	channel := request.conn.channel
	if h.cf.PrintMsg {
		log.Println("[REQU]<-", request.Id, channel, "["+cmd+"]", subStr(string(request.Data), 200))
	}
	// 执行中间件
	for _, mdFunc := range h.middle {
		rsp := mdFunc(request)
		if rsp != nil {
			// NEXT_MIDDLE 表示当前的函数没有处理完成，还需要下个中间件处理
			if rsp.State == config.NEXT_MIDDLE {
				continue
			}
			// 返回消息
			if request.Id != 0 {
				rsp.Id = request.Id
				request.conn.sendResponse <- rsp
			}
			return
		} else {
			break
		}
	}
	sub, ok := h.subscribes.Load(cmd)
	if ok {
		subs := sub.([]*SubscribeData)
		// 倒序查找是为了新增的频道响应函数优先执行
		for i := len(subs) - 1; i >= 0; i-- {
			rg := subs[i]
			if rg.Filter(request.conn) {
				state, data := rg.BackFunc(request)
				// NEXT_SUBSCRIBE 表示当前的函数没有处理完成，还需要下个注册函数处理
				if state == config.NEXT_SUBSCRIBE {
					continue
				}
				var byteData []byte
				switch data := data.(type) {
				case []byte:
					byteData = data
				case string:
					byteData = []byte(data)
				default:
					if data != nil {
						// 自动转换数据为json格式
						var err error
						byteData, err = json.Marshal(data)
						if err != nil {
							log.Println(err.Error())
							state = config.CONVERT_FAILED
							byteData = fmt.Appendf(nil, "[%s] %s %s", IdMsg(config.CONVERT_FAILED), request.conn.channel, request.Cmd)
						}
					}
				}
				// 如果id为0表示不需要回应
				if request.Id != 0 {
					request.conn.sendResponse <- &ResponseData{
						Id:    request.Id,
						State: state,
						Data:  byteData,
					}
					if h.cf.PrintMsg {
						log.Println("[RESP]->", request.Id, channel, "["+cmd+"]", state, subStr(string(byteData), 200))
					}
				}
				return
			}
		}
	}

	log.Println("[not match command]", channel, cmd)
	// 返回没有匹配的消息
	request.conn.sendResponse <- &ResponseData{
		Id:    request.Id,
		State: config.NO_MATCH_CMD,
		Data:  fmt.Appendf(nil, "[%s] Channel: %s, Cmd: %s", IdMsg(config.NO_MATCH_CMD), channel, cmd),
	}
}

// 请求频道并获取数据，采用回调的方式返回结果
// 当前调用将会阻塞，直到命令都执行结束，最后返回执行的数量
// 如果 backFunc 返回为 false 则提前结束
// 最大数量和超时时间如果为0的话表示使用默认值
func (h *Hub) GetWithStruct(gd *GetData, backFunc GetBackFunc) (count int, err error) {
	if gd.Filter == nil {
		return 0, errors.New("not filter found")
	}
	if gd.Timeout <= 0 {
		gd.Timeout = h.cf.WriteWait
	}
	if gd.backchan == nil {
		gd.backchan = make(chan *ResponseData, 32)
	}
	sendMax, err := h.sendRequest(gd)
	if err != nil {
		return 0, err
	}
	if sendMax <= 0 {
		return 0, nil
	}
	// 避免出现异常时线程无法退出
	timer := time.NewTimer(time.Millisecond * time.Duration(gd.Timeout+h.cf.WriteWait*2))
	defer func() {
		if !timer.Stop() {
			select {
			case <-timer.C:
			default:
			}
		}
		close(gd.backchan)
	}()
	for {
		select {
		case rp := <-gd.backchan:
			if rp == nil || rp.conn == nil {
				// 可能是已经退出了
				return
			}
			ch := rp.conn.channel
			if h.cf.PrintMsg {
				log.Println("[RECV]<-", rp.Id, ch, "["+gd.Cmd+"]", rp.State, subStr(string(rp.Data), 200))
			}
			count++
			// 如果这里返回为false这跳出循环
			if backFunc != nil && !backFunc(rp) {
				return
			}
			if count >= sendMax {
				return
			}
		case <-timer.C:
			return
		case <-h.ctx.Done():
			return
		}
	}
}

// 请求频道并获取数据，采用回调的方式返回结果
// 当前调用将会阻塞，直到命令都执行结束，最后返回执行的数量
// 如果 backFunc 返回为 false 则提前结束
func (h *Hub) Get(filter FilterFunc, cmd string, data any, backFunc GetBackFunc) (count int, err error) {
	return h.GetWithStruct(&GetData{
		Filter: filter,
		Cmd:    cmd,
		Data:   data,
	}, backFunc)
}

// 获取一个数据，阻塞等待到超时间隔
func (h *Hub) GetOneWithStruct(gd *GetData) (response *ResponseData) {
	if gd.Filter == nil {
		return &ResponseData{
			State: config.NO_MATCH_FILTER,
			Data:  []byte(IdMsg(config.NO_MATCH_FILTER)),
		}
	}
	gd.Max = 1
	h.GetWithStruct(gd, func(rp *ResponseData) (ok bool) {
		response = rp
		return false
	})
	if response == nil {
		return &ResponseData{
			State: config.NO_MATCH_CONNECT,
			Data:  []byte(IdMsg(config.NO_MATCH_CONNECT)),
		}
	}
	return
}

// 只获取一个频道的数据，阻塞等待到默认超时间隔
// 如果没有结果将返回 NO_MATCH
func (h *Hub) GetOne(filter FilterFunc, cmd string, data any) (response *ResponseData) {
	return h.GetOneWithStruct(&GetData{
		Filter: filter,
		Cmd:    cmd,
		Data:   data,
		Max:    1,
	})
}

func (h *Hub) GetRandOne(filter FilterFunc, cmd string, data any) (response *ResponseData) {
	return h.GetOneWithStruct(&GetData{
		Filter: filter,
		Cmd:    cmd,
		Data:   data,
		Max:    1,
		Rand:   true,
	})
}

// 只获取一个频道的数据，阻塞等待到指定超时间隔
// 如果没有结果将返回 NO_MATCH
func (h *Hub) GetOneWithTimeout(filter FilterFunc, cmd string, data any, timeout int) (response *ResponseData) {
	return h.GetOneWithStruct(&GetData{
		Filter:  filter,
		Cmd:     cmd,
		Data:    data,
		Max:     1,
		Timeout: timeout,
	})
}

func (h *Hub) GetRandOneWithTimeout(filter FilterFunc, cmd string, data any, timeout int) (response *ResponseData) {
	return h.GetOneWithStruct(&GetData{
		Filter:  filter,
		Cmd:     cmd,
		Data:    data,
		Max:     1,
		Timeout: timeout,
		Rand:    true,
	})
}

// 推送消息出去，不需要返回数据
func (h *Hub) Push(filter FilterFunc, cmd string, data any) {
	h.PushWithMax(filter, cmd, data, 0)
}

// 推送最大对应数量的消息出去，不需要返回数据
func (h *Hub) PushWithMax(filter FilterFunc, cmd string, data any, max int) {
	// 排除空频道
	if filter == nil {
		return
	}
	gd := &GetData{
		Filter:   filter,
		Cmd:      cmd,
		Data:     data,
		Max:      max,
		Timeout:  h.cf.WriteWait,
		backchan: nil,
	}
	h.sendRequest(gd)
}

// 增加连接
func (h *Hub) addLine(line *Line) {
	if h.lines.Exist(line) {
		log.Println("connect have exist")
		// 连接已经存在，直接返回
		return
	}
	h.lines.Store(line)
}

// 删除连接
func (h *Hub) removeLine(line *Line) {
	line.Close(true)
	h.lines.Delete(line)
}

// 获取指定连接的连接持续时间
func (h *Hub) ConnectDuration(line *Line) time.Duration {
	return time.Since(line.started)
}

// 绑定端口，建立服务
// 需要程序运行时调用
func (h *Hub) BindForServer(info *HostInfo) (err error) {
	doConnectFunc := func(conn conn.Connect) {
		proto, version, channel, auth, err := conn.ReadAuthInfo()
		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(false, 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(false, proto, version, channel, auth)); err != nil {
			log.Println("[WriteAuthInfo ERROR]", err)
			conn.Close()
			return
		}
		// 将连接加入现有连接中
		done := false
		h.lines.Range(func(id int, line *Line) bool {
			if line.state == Disconnected && line.host == nil && line.IsChannelEqual(channel) {
				line.Start(channel, conn, nil)
				done = true
				return false
			}
			return true
		})
		// 新建一个连接
		if !done {
			line := NewConnect(h.cf, h, channel, conn, nil, false)
			h.addLine(line)
		}
	}
	if info.Version == ws2.VERSION && info.Proto == ws2.PROTO {
		bind := ""
		if info.Bind != "" {
			bind = net.JoinHostPort(info.Bind, strconv.Itoa(int(info.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 errors.New("not connect protocol and version found")
}

// 新建一个连接，不同的连接协议由底层自己选择
// channel: 要连接的频道信息，需要能表达频道关键信息的部分
func (h *Hub) ConnectToServer(channel string, force bool, host *HostInfo, autoReconnect bool) (err error) {
	// 检查当前channel是否已经存在
	if !force {
		line := h.ChannelToLine(channel)
		if line != nil && line.state == Connected {
			// err = fmt.Errorf("[ConnectToServer ERROR] existed channel: %s", channel)
			log.Println("[ConnectToServer] channel existed:", channel)
			return
		}
	}
	if host == nil {
		if h.connectHostFunc == nil {
			return errors.New("not connect host func found")
		}
		// 获取服务地址等信息
		host, err = h.connectHostFunc(channel, Both)
		if err != nil {
			return err
		}
		if host == nil {
			// 如果地址为空表示不需要连接，自己返回
			return
		}
	}

	var conn conn.Connect
	var runProto string
	addr := net.JoinHostPort(host.Host, strconv.Itoa(int(host.Port)))

	// 添加定时器
	ctx, cancel := context.WithTimeout(context.Background(), time.Millisecond*time.Duration(h.cf.ConnectTimeout))
	defer cancel()
	taskCh := make(chan bool)
	done := false

	go func() {
		if host.Version == ws2.VERSION && (host.Proto == ws2.PROTO || host.Proto == ws2.PROTO_STL) {
			runProto = ws2.PROTO
			if h.cf.PrintMsg {
				log.Println("[ConnectToServer] connecting:", host.Proto, addr, host.Path, host.Hash)
			}
			conn, err = ws2.Dial(h.cf, host.Proto, addr, host.Path, host.Hash)
		} else if host.Version == tcp2.VERSION && host.Proto == tcp2.PROTO {
			runProto = tcp2.PROTO
			if h.cf.PrintMsg {
				log.Println("[ConnectToServer] connecting:", host.Proto, addr, host.Hash)
			}
			conn, err = tcp2.Dial(h.cf, addr, host.Hash)
		} else {
			err = fmt.Errorf("not correct protocol and version found in: %+v", host)
		}
		if done {
			if err != nil {
				log.Println("[ConnectToServer] dial failed:", err)
			}
			if conn != nil {
				conn.Close()
			}
		} else {
			taskCh <- err == nil
		}
	}()

	select {
	case ok := <-taskCh:
		cancel()
		if !ok || err != nil || conn == nil {
			log.Println("[ConnectToServer] connect failed:", err)
			host.Errors++
			host.Updated = time.Now()
			if err == nil {
				err = errors.New("unknown error")
			}
			return err
		}
	case <-ctx.Done():
		done = true
		return errors.New("timeout")
	case <-h.ctx.Done():
		return errors.New("quit")
	}

	// 如果 host 是代理，将代理信息添加到channel中
	localChannel := h.channel
	if host.Proxy {
		localChannel = localChannel + "?proxy=" + host.Host
	}
	// 发送验证信息
	if err := conn.WriteAuthInfo(localChannel, h.authFunc(true, runProto, 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 != runProto {
		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.state == Connected {
				if force {
					line.Close(true)
				} else {
					err = fmt.Errorf("[connectToServer] channel already connected: %s", channel)
					log.Println(err)
					return false
				}
			}
			line.Start(channel, conn, host)
			done = true
			return false
		}
		return true
	})
	if err != nil {
		return err
	}
	// 新建一个连接
	if !done {
		line := NewConnect(h.cf, h, channel, conn, host, autoReconnect)
		h.addLine(line)
	}
	return nil
}

// 重试方式连接服务
// 将会一直阻塞直到连接成功
func (h *Hub) ConnectToServerX(channel string, force bool, host *HostInfo) {
	if host == nil {
		if h.connectHostFunc == nil {
			log.Println("[ConnectToServerX] not connect host func found")
			return
		}
		host, _ = h.connectHostFunc(channel, Direct)
		if host == nil {
			log.Println("[ConnectToServerX] not host found for channel:", channel)
			return
		}
	}
	for {
		err := h.ConnectToServer(channel, force, host, true)
		if err == nil {
			return
		}
		log.Println("[ConnectToServerX] connect failed:", channel, host.Url(), err)
		// 产生一个随机数避免刹间重连过载
		delay := time.Duration(rand.Intn(h.cf.ConnectTimeout)+(h.cf.ConnectTimeout/2)) * time.Millisecond
		log.Println("[ConnectToServerX] will reconnect with delay:", delay)
		host = nil
		time.Sleep(delay)
	}
}

// 检测处理连接状态，只在客户端有效
func (h *Hub) checkConnect() {
	// 检查客户端获取主机地址函数
	if h.connectHostFunc == nil {
		return
	}
	proxyTicker := time.NewTicker(time.Duration(h.cf.ProxyTimeout * int(time.Millisecond)))
	connectTicker := time.NewTicker(time.Millisecond * time.Duration(h.cf.ConnectCheck))
	for {
		select {
		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) {
					host, err := h.connectHostFunc(line.channel, Direct)
					if err != nil {
						log.Println("[proxyTicker connectHostFunc]", err)
						return false
					}
					if host != nil {
						err = h.ConnectToServer(line.channel, true, host, line.autoReconnect)
						if err != nil {
							log.Println("[checkProxyConnect ConnectToServer]", err)
						}
					}
				}
				return true
			})
		case <-connectTicker.C:
			h.lines.Range(func(id int, line *Line) bool {
				if line.autoReconnect && line.host != nil && line.state == Disconnected {
					err := h.ConnectToServer(line.channel, true, nil, true)
					if err != nil {
						log.Println("[connectTicker ConnectToServer]", err)
					}
				}
				return true
			})
		case <-h.ctx.Done():
			return
		}
	}
}

// 退出所有的连接
func (h *Hub) Quit() {
	h.cancel()
	h.lines.Range(func(id int, line *Line) bool {
		if line.state == Connected {
			line.Close(true)
		}
		return true
	})
}

// 建立一个集线器
// connectFunc 用于监听连接状态的函数，可以为nil
func NewHub(
	cf *config.Config,
	channel string,
	// 客户端需要用的函数，提供连接的主机信息 (服务端可空)
	connectHostFunc ConnectHostFunc,
	// 验证函数，获取认证信息，用于发送给对方
	authFunc AuthFunc,
	// 核对发送过来的认证信息
	checkAuthFunc CheckAuthFunc,
	// 连接状态变化时调用的函数
	connectStatusFunc ConnectStatusFunc,
	// 验证发送数据的条件是否满足 (可为空)
	// checkConnectOkFunc CheckConnectOkFunc,
) (h *Hub) {
	if cf == nil {
		cf = config.NewConfig()
	}
	ctx, cancel := context.WithCancel(context.Background())
	h = &Hub{
		ctx:               ctx,
		cancel:            cancel,
		cf:                cf,
		globalID:          uint16(time.Now().UnixNano()) % config.ID_MAX,
		channel:           channel,
		middle:            make([]MiddleFunc, 0),
		lines:             NewMapx(),
		connectHostFunc:   connectHostFunc,
		authFunc:          authFunc,
		checkAuthFunc:     checkAuthFunc,
		connectStatusFunc: connectStatusFunc,
		// checkConnectOkFunc:   checkConnectOkFunc,
		lastCleanDeadConnect: time.Now().UnixMilli(),
	}
	go h.checkConnect()
	return h
}