事件模型：
image.png
核心代码：
/**
启动tcpServer
*/
func startTcpServer(port string) {
addr := gxnet.HostAddress2("localhost", port)
server := getty.NewTCPServer(
getty.WithLocalAddress(addr),
)
// run server
server.RunEventLoop(newSession)
log.Info("server bind addr{%s} ok!", port)
}
func newSession(session getty.Session) error {
var (
ok bool
tcpConn *net.TCPConn
)
if tcpConn, ok = session.Conn().(*net.TCPConn); !ok {
panic(fmt.Sprintf("%s, session.conn{%#v} is not tcp connection\n", session.Stat(), session.Conn()))
}
tcpConn.SetKeepAlive(true)
session.SetPkgHandler(irisPkgHandler)
session.SetEventListener(irisMsgHandler)
session.SetCronPeriod(1000 * 10)
log.Debug("client new session:%s\n", session.Stat())
return nil
}
設置包处理器, 用于消息的编解码，需实现ReadWriter 接口，让用户可以自己实现具体的序列化/反序列化逻辑
session.SetPkgHandler(irisPkgHandler)
設置監控接口, 通过这些接口可以知道每个网络连接在每个阶段的状态， 需实现EventListener接口，
session.SetEventListener(irisMsgHandler)
- 「OnOpen」：连接建立时提供给用户使用，若当前连接总数超过用户设定的连接数，则可以返回一个非 nil 的 error，Getty 就会在初始阶段关闭这个连接。
- 「OnError」：用于连接有异常时的监控，Getty 执行这个接口后关闭连接
- 「OnClose」：用于连接关闭时的监控，Getty 执行这个接口后关闭连接。
- 「OnMessage」：当 Getty 调用 Reader 接口成功从 TCP流/UDP/WebSocket 网络中解析出一个 package 后，通过这个接口把数据包交给用户处理。
- 「OnCron」：定时接口，用户可以在这里接口函数中执行心跳检测等一些定时逻辑。
设置定时接口的触发周期
session.SetCronPeriod(1000 * 10)
完整示例：
package main
import (
"flag"
"fmt"
getty "github.com/AlexStocks/getty/transport"
gxnet "github.com/AlexStocks/goext/net"
log "github.com/sirupsen/logrus"
conf "iris/config"
"iris/global"
"iris/tcp/handler"
"iris/web/router"
"net"
"net/http"
"time"
)
var (
cfgPath string
irisMsgHandler = handler.NewEchoMessageHandler()
irisPkgHandler = handler.NewEchoPackageHandler()
)
func init() {
flag.StringVar(&cfgPath, "c", "./etc/config.yaml", "")
flag.Parse()
}
func main() {
println("starting.....")
err := conf.Init(cfgPath)
if err != nil {
log.WithFields(log.Fields{"cfg_path": cfgPath}).WithError(err).Error("[main] config init error")
return
}
log.Infof("conf:%v", conf.GConfig)
//初始化全局变量
global.Init()
go startTcpServer(conf.GConfig.TcpServer.Port)
router, err := router.InitRouter()
if err != nil {
log.WithError(err).Error("[main] init router error")
return
}
server := &http.Server{
Addr: ":" + conf.GConfig.WebServer.Port,
Handler: router,
ReadTimeout: time.Second * 60,
WriteTimeout: time.Second * 60,
MaxHeaderBytes: http.DefaultMaxHeaderBytes,
}
err = server.ListenAndServe()
if err != nil {
log.WithError(err).Error("[main] start web server error")
return
}
}
/**
启动tcpServer
*/
func startTcpServer(port string) {
addr := gxnet.HostAddress2("localhost", port)
server := getty.NewTCPServer(
getty.WithLocalAddress(addr),
)
// run server
server.RunEventLoop(newSession)
log.Info("server bind addr{%s} ok!", port)
}
func newSession(session getty.Session) error {
var (
ok bool
tcpConn *net.TCPConn
)
if tcpConn, ok = session.Conn().(*net.TCPConn); !ok {
panic(fmt.Sprintf("%s, session.conn{%#v} is not tcp connection\n", session.Stat(), session.Conn()))
}
tcpConn.SetKeepAlive(true)
session.SetPkgHandler(irisPkgHandler)
session.SetEventListener(irisMsgHandler)
session.SetCronPeriod(1000 * 10)
log.Debug("client new session:%s\n", session.Stat())
return nil
}
定义 handler：
package handler
import (
"encoding/hex"
getty "github.com/AlexStocks/getty/transport"
log "github.com/sirupsen/logrus"
"iris/constant"
"iris/global"
"iris/utils"
"strings"
)
type IrisMessageHandler struct{}
func NewEchoMessageHandler() *IrisMessageHandler {
return &IrisMessageHandler{}
}
func (h *IrisMessageHandler) OnOpen(session getty.Session) error {
log.Info("client connected:" + session.RemoteAddr())
return nil
}
func (h *IrisMessageHandler) OnError(session getty.Session, err error) {
log.Errorf("session{%s} got error{%v}.", session.Stat(), err)
}
func (h *IrisMessageHandler) OnClose(session getty.Session) {
log.Infof("session{%s} is closing......", session.Stat())
}
func (h *IrisMessageHandler) OnMessage(session getty.Session, pkg interface{}) {
log.Infof("get echo package{%s}", pkg)
}
func (h *IrisMessageHandler) OnCron(session getty.Session) {
//心跳处理
log.Info("心跳处理")
}
定义包处理器, 用于消息的编解码，需实现ReadWriter 接口，让用户可以自己实现具体的序列化/反序列化逻辑
type IrisPackageHandler struct{}
func NewEchoPackageHandler() *IrisPackageHandler {
return &IrisPackageHandler{}
}
func (h *IrisPackageHandler) Read(ss getty.Session, data []byte) (interface{}, int, error) {
log.Info("receive:" + string(data))
return string(data), len(data), nil
}
func (h *IrisPackageHandler) Write(ss getty.Session, pkg interface{}) ([]byte, error) {
log.Info("send:" + pkg.(string))
return hex.DecodeString(pkg.(string))
}
