tinymq/dail.go

package tinymq

import (
	"fmt"
	"log"
	"net"
	"strconv"
	"strings"

	"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"
)

// 根据协议名建立连接，输出实际代码实现的协议
// 会一直阻塞直到连接结束
func Dail(host *conn.HostInfo, cf *config.Config) (conn conn.Connect, err error) {
	addr := net.JoinHostPort(host.Host, strconv.Itoa(int(host.Port)))
	if host.Version == ws2.VERSION && (host.Proto == ws2.PROTO || host.Proto == ws2.PROTO_STL) {
		if cf.PrintMsg {
			log.Println("[Dail]:", host.Proto, addr, host.Path, host.Hash)
		}
		conn, err = ws2.Dial(cf, host.Proto, addr, host.Path, host.Hash)
	} else if host.Version == tcp2.VERSION && host.Proto == tcp2.PROTO {
		if cf.PrintMsg {
			log.Println("[Dail]:", host.Proto, addr, host.Hash)
		}
		conn, err = tcp2.Dial(cf, addr, host.Hash)
	} else {
		err = fmt.Errorf("not correct protocol and version found in: %+v", host)
	}
	return
}

// 检验 AuthInfo
func CheckAuthInfo(
	conn conn.Connect,
	host *conn.HostInfo,
	localChannel string,
	remoteChannel string,
	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()
		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()
		return err
	}
	// 检查版本和协议是否一致
	if version != host.Version || !strings.HasPrefix(host.Proto, proto) {
		err = fmt.Errorf("[CheckAuthInfo] 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)
		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))
		log.Println(err)
		// conn.Close()
		// host.Errors++
		// host.Updated = time.Now()
		return err
	}
	// 更新服务主机信息
	// host.Errors = 0
	// host.Updated = time.Now()
	return
}