refactor: 重构,更优雅的使用方式。

main v0.7.0
NoahLan 2 years ago
parent c65fd5961b
commit ebcbd0f88f

@ -0,0 +1,23 @@
package nnet
import (
"git.noahlan.cn/noahlan/nnet/connection"
"git.noahlan.cn/noahlan/ntools-go/core/nlog"
"net"
)
// Dial 连接服务器
func (ngin *Engine) Dial(addr string) (*connection.Connection, error) {
err := ngin.setup()
if err != nil {
nlog.Errorf("%s failed to setup server, err:%v", ngin.LogPrefix(), err)
return nil, err
}
conn, err := net.Dial("tcp", addr)
nlog.Must(err)
nlog.Infof("%s now connect to %s...", ngin.LogPrefix(), addr)
return ngin.handle(conn), nil
}

@ -1,7 +1,7 @@
package config
import (
"git.noahlan.cn/noahlan/ntools-go/core/pool"
"fmt"
"time"
)
@ -16,29 +16,20 @@ const (
type (
EngineConf struct {
ServerConf
Pool pool.Config
}
ServerConf struct {
// Protocol 协议名
// "tcp", "tcp4", "tcp6", "unix" or "unixpacket"
// 若只想开启IPv4, 使用tcp4即可
Protocol string `json:",default=tcp4"`
// Addr 服务地址
// 地址可直接使用hostname,但强烈不建议这样做,可能会同时监听多个本地IP
// 如果端口号不填或端口号为0例如"127.0.0.1:" 或 ":0",服务端将选择随机可用端口
Addr string `json:",default=0.0.0.0"`
// Name 服务端名称默认为n-net
Name string `json:",default=n-net"`
// TaskTimerPrecision // 全局任务的timer间隔
TaskTimerPrecision time.Duration `json:",default=1s"`
// Mode 运行模式
Mode string `json:",default=dev,options=[dev,test,prod]"`
// Name 引擎名称
Name string `json:",default=NL,env=ENGINE_NAME"`
}
)
// ShallLogDebug 是否应该打印 Debug 级别的日志,打印的首要条件是 nlog 的打印级别为 debug
func ShallLogDebug(mode string) bool {
return mode == DevMode || mode == TestMode
func (c EngineConf) ShallLogDebug() bool {
return c.Mode == DevMode || c.Mode == TestMode
}
func (c EngineConf) LogPrefix() string {
return fmt.Sprintf("[NNet-%s]", c.Name)
}

@ -0,0 +1,14 @@
package config
type (
TCPServerConf struct {
// Protocol 协议名
// "tcp", "tcp4", "tcp6", "unix" or "unixpacket"
// 若只想开启IPv4, 使用tcp4即可
Protocol string `json:",default=tcp4,env=TCP_PROTOCOL"`
// Addr 服务地址
// 地址可直接使用hostname,但强烈不建议这样做,可能会同时监听多个本地IP
// 如果端口号不填或端口号为0例如"127.0.0.1:" 或 ":0",服务端将选择随机可用端口
Addr string `json:",default=0.0.0.0:9876,env=TCP_ADDR"`
}
)

@ -0,0 +1,35 @@
package config
import (
"net/http"
"time"
)
type (
WSServerConf struct {
// Addr 服务地址
// 地址可直接使用hostname,但强烈不建议这样做,可能会同时监听多个本地IP
// 如果端口号不填或端口号为0例如"127.0.0.1:" 或 ":0",服务端将选择随机可用端口
Addr string `json:",default=0.0.0.0:9876,env=WS_ADDR"`
// Path 监听路径 /WebsocketPath
Path string `json:",default=/,env=WS_PATH"`
// HandshakeTimeout 握手超时时间默认0
HandshakeTimeout time.Duration `json:",default=0"`
// ReadBufferSize 读缓冲区大小
ReadBufferSize int `json:",default=2048"`
// WriteBufferSize 写缓冲区大小
WriteBufferSize int `json:",default=2048"`
// Compression 是否使用压缩协议
Compression bool `json:",default=false"`
// TLSCertificate 证书地址
TLSCertificate string `json:",optional"`
// TLS 证书key地址
TLSKey string `json:",optional"`
// check origin
CheckOrigin func(*http.Request) bool
}
)
func (c WSServerConf) IsTLS() bool {
return len(c.TLSCertificate) > 0 && len(c.TLSKey) > 0
}

@ -1,11 +1,11 @@
package core
package connection
import (
"errors"
"fmt"
"git.noahlan.cn/noahlan/nnet/entity"
"git.noahlan.cn/noahlan/nnet/packet"
"git.noahlan.cn/noahlan/nnet/scheduler"
"git.noahlan.cn/noahlan/nnet/serialize"
"git.noahlan.cn/noahlan/nnet/session"
"git.noahlan.cn/noahlan/ntools-go/core/nlog"
"git.noahlan.cn/noahlan/ntools-go/core/pool"
"net"
@ -31,16 +31,26 @@ const (
StatusClosed
)
type ConnType int
const (
ConnTypeTCP ConnType = iota // TCP connection
ConnTypeWS // Websocket connection
)
type (
connection struct {
session *session // Session
ngin *engine // engine
Connection struct {
conf Config // 配置
session *session.Session // Session
status int32 // 连接状态
conn net.Conn // low-level conn fd
isWS bool // 是否为websocket
typ ConnType // 连接类型
packer packet.Packer // 封包、拆包器
packer packet.Packer // 封包、拆包器
serializer serialize.Serializer // 消息序列化/反序列化器
pipeline Pipeline // 连接生命周期管理
handleFn func(conn *Connection, pkg packet.IPacket) // 消息处理方法
lastMid uint64 // 最近一次消息ID
@ -49,32 +59,54 @@ type (
chWrite chan []byte // 消息发送通道(二进制消息)
}
packetFn func(conn *Connection, pkg packet.IPacket)
Config struct {
LogDebug bool
LogPrefix string
}
PendingMessage struct {
header interface{}
payload interface{}
}
)
func newConnection(server *engine, conn net.Conn) *connection {
r := &connection{
ngin: server,
func NewConnection(
id int64,
conn net.Conn,
conf Config,
packerBuilder packet.PackerBuilder,
serializer serialize.Serializer,
pipeline Pipeline,
handleFn packetFn) *Connection {
r := &Connection{
conf: conf,
session: session.NewSession(id),
status: StatusStart,
conn: conn,
packer: server.packerFn(),
typ: ConnTypeTCP,
packer: packerBuilder(),
serializer: serializer,
pipeline: pipeline,
handleFn: handleFn,
lastMid: 0,
chDie: make(chan struct{}),
chSend: make(chan PendingMessage, 128),
chWrite: make(chan []byte, 128),
}
_, r.isWS = conn.(*WSConn)
// binding session
r.session = newSession(r, server.sessIdMgr.SessionID())
_, ok := conn.(*WSConn)
if ok {
r.typ = ConnTypeWS
}
return r
}
func (r *connection) Send(header, payload interface{}) (err error) {
func (r *Connection) Send(header, payload interface{}) (err error) {
defer func() {
if e := recover(); e != nil {
err = ErrBrokenPipe
@ -87,7 +119,7 @@ func (r *connection) Send(header, payload interface{}) (err error) {
return err
}
func (r *connection) SendBytes(data []byte) (err error) {
func (r *Connection) SendBytes(data []byte) (err error) {
defer func() {
if e := recover(); e != nil {
err = ErrBrokenPipe
@ -97,35 +129,35 @@ func (r *connection) SendBytes(data []byte) (err error) {
return err
}
func (r *connection) Status() int32 {
func (r *Connection) Status() int32 {
return atomic.LoadInt32(&r.status)
}
func (r *connection) SetStatus(s int32) {
func (r *Connection) SetStatus(s int32) {
atomic.StoreInt32(&r.status, s)
}
func (r *connection) Conn() (net.Conn, bool) {
return r.conn, r.isWS
func (r *Connection) Conn() (net.Conn, ConnType) {
return r.conn, r.typ
}
func (r *connection) ID() int64 {
func (r *Connection) ID() int64 {
return r.session.ID()
}
func (r *connection) Session() entity.Session {
func (r *Connection) Session() *session.Session {
return r.session
}
func (r *connection) LastMID() uint64 {
func (r *Connection) LastMID() uint64 {
return r.lastMid
}
func (r *connection) SetLastMID(mid uint64) {
func (r *Connection) SetLastMID(mid uint64) {
atomic.StoreUint64(&r.lastMid, mid)
}
func (r *connection) serve() {
func (r *Connection) Serve() {
_ = pool.Submit(func() {
r.write()
})
@ -135,15 +167,15 @@ func (r *connection) serve() {
})
}
func (r *connection) write() {
func (r *Connection) write() {
defer func() {
close(r.chSend)
close(r.chWrite)
_ = r.Close()
if r.ngin.shallLogDebug() {
if r.conf.LogDebug {
nlog.Debugf("%s [writeLoop] connection write goroutine exit, ConnID=%d, SessionUID=%s",
r.ngin.logPrefix(), r.ID(), r.session.UID())
r.conf.LogPrefix, r.ID(), r.session.UID())
}
}()
@ -151,51 +183,52 @@ func (r *connection) write() {
select {
case data := <-r.chSend:
// marshal packet body (data)
if r.ngin.serializer == nil {
if r.serializer == nil {
if _, ok := data.payload.([]byte); !ok {
nlog.Errorf("%s serializer is nil, but payload type not []byte", r.ngin.logPrefix())
nlog.Errorf("%s serializer is nil, but payload type not []byte", r.conf.LogPrefix)
break
}
} else {
payload, err := r.ngin.serializer.Marshal(data.payload)
payload, err := r.serializer.Marshal(data.payload)
if err != nil {
nlog.Errorf("%s message body marshal err: %v", r.ngin.logPrefix(), err)
nlog.Errorf("%s message body marshal err: %v", r.conf.LogPrefix, err)
break
}
data.payload = payload
}
// invoke pipeline
if pipe := r.ngin.pipeline; pipe != nil {
if pipe := r.pipeline; pipe != nil {
err := pipe.Outbound().Process(r, data)
if err != nil {
nlog.Errorf("%s pipeline err: %s", r.ngin.logPrefix(), err.Error())
nlog.Errorf("%s pipeline err: %s", r.conf.LogPrefix, err.Error())
}
}
// packet pack data
p, err := r.packer.Pack(data.header, data.payload.([]byte))
if err != nil {
nlog.Errorf("%s pack err: %s", r.ngin.logPrefix(), err.Error())
nlog.Errorf("%s pack err: %s", r.conf.LogPrefix, err.Error())
break
}
r.chWrite <- p
case data := <-r.chWrite:
// 回写数据
if _, err := r.conn.Write(data); err != nil {
nlog.Errorf("%s write data err: %s", r.ngin.logPrefix(), err.Error())
nlog.Errorf("%s write data err: %s", r.conf.LogPrefix, err.Error())
break
}
//nlog.Debugf("write data %v", data)
case <-r.chDie: // connection close signal
return
case <-r.ngin.dieChan: // application quit signal
return
// TODO
//case <-r.ngin.dieChan: // application quit signal
// return
}
}
}
func (r *connection) read() {
func (r *Connection) read() {
defer func() {
_ = r.Close()
}()
@ -205,38 +238,38 @@ func (r *connection) read() {
//nlog.Debugf("receive data %v", buf[:n])
if err != nil {
nlog.Errorf("%s [readLoop] Read message error: %s, session will be closed immediately",
r.ngin.logPrefix(), err.Error())
r.conf.LogPrefix, err.Error())
return
}
if n == 0 {
nlog.Errorf("%s [readLoop] Read empty message, session will be closed immediately",
r.ngin.logPrefix())
r.conf.LogPrefix)
return
}
if r.packer == nil {
nlog.Errorf("%s [readLoop] unexpected error: packer is nil", r.ngin.logPrefix())
nlog.Errorf("%s [readLoop] unexpected error: packer is nil", r.conf.LogPrefix)
return
}
// warning: 为性能考虑复用slice处理数据buf传入后必须要copy再处理
packets, err := r.packer.Unpack(buf[:n])
if err != nil {
nlog.Errorf("%s unpack err: %s", r.ngin.logPrefix(), err.Error())
nlog.Errorf("%s unpack err: %s", r.conf.LogPrefix, err.Error())
}
// packets 处理
for _, p := range packets {
if err := r.processPacket(p); err != nil {
nlog.Errorf("%s process packet err: %s", r.ngin.logPrefix(), err.Error())
nlog.Errorf("%s process packet err: %s", r.conf.LogPrefix, err.Error())
continue
}
}
}
}
func (r *connection) processPacket(packet packet.IPacket) error {
if pipe := r.ngin.pipeline; pipe != nil {
func (r *Connection) processPacket(packet packet.IPacket) error {
if pipe := r.pipeline; pipe != nil {
err := pipe.Inbound().Process(r, packet)
if err != nil {
return errors.New(fmt.Sprintf("pipeline process failed: %v", err.Error()))
@ -244,32 +277,34 @@ func (r *connection) processPacket(packet packet.IPacket) error {
}
if r.Status() == StatusWorking {
// HandleFunc
// 处理包消息
_ = pool.Submit(func() {
r.ngin.handler.Handle(r, packet)
r.handleFn(r, packet)
})
}
return nil
}
func (r *connection) Close() error {
func (r *Connection) DieChan() chan struct{} {
return r.chDie
}
func (r *Connection) Close() error {
if r.Status() == StatusClosed {
return ErrCloseClosedSession
}
r.SetStatus(StatusClosed)
if r.ngin.shallLogDebug() {
nlog.Debugf("%s close connection, ID: %d", r.ngin.logPrefix(), r.ID())
if r.conf.LogDebug {
nlog.Debugf("%s close connection, ID: %d", r.conf.LogPrefix, r.ID())
}
select {
case <-r.chDie:
default:
close(r.chDie)
scheduler.PushTask(func() { r.ngin.lifetime.Close(r) })
}
_ = r.ngin.connManager.Remove(r)
r.session.Close()
r.session.Close()
return r.conn.Close()
}

@ -1,8 +1,7 @@
package conn
package connection
import (
"errors"
"git.noahlan.cn/noahlan/nnet/entity"
"git.noahlan.cn/noahlan/ntools-go/core/nlog"
"sync"
"sync/atomic"
@ -27,7 +26,7 @@ type Group struct {
status int32 // group current status
name string // group name
conns map[int64]entity.NetworkEntity
conns map[int64]*Connection
}
func NewGroup(name string) *Group {
@ -35,16 +34,16 @@ func NewGroup(name string) *Group {
mu: sync.RWMutex{},
status: groupStatusWorking,
name: name,
conns: make(map[int64]entity.NetworkEntity),
conns: make(map[int64]*Connection),
}
}
// Member returns connection by specified uid
func (c *Group) Member(uid string) (entity.NetworkEntity, bool) {
c.mu.RLock()
defer c.mu.RUnlock()
func (g *Group) Member(uid string) (*Connection, bool) {
g.mu.RLock()
defer g.mu.RUnlock()
for _, e := range c.conns {
for _, e := range g.conns {
if e.Session().UID() == uid {
return e, true
}
@ -54,18 +53,18 @@ func (c *Group) Member(uid string) (entity.NetworkEntity, bool) {
}
// MemberBySID returns specified sId's connection
func (c *Group) MemberBySID(id int64) (entity.NetworkEntity, bool) {
c.mu.RLock()
defer c.mu.RUnlock()
func (g *Group) MemberBySID(id int64) (*Connection, bool) {
g.mu.RLock()
defer g.mu.RUnlock()
e, ok := c.conns[id]
e, ok := g.conns[id]
return e, ok
}
func (c *Group) Members() []entity.NetworkEntity {
var resp []entity.NetworkEntity
c.PeekMembers(func(_ int64, e entity.NetworkEntity) bool {
resp = append(resp, e)
func (g *Group) Members() []*Connection {
var resp []*Connection
g.PeekMembers(func(_ int64, c *Connection) bool {
resp = append(resp, c)
return false
})
return resp
@ -73,33 +72,33 @@ func (c *Group) Members() []entity.NetworkEntity {
// PeekMembers returns all members in current group
// fn 返回true跳过循环反之一直循环
func (c *Group) PeekMembers(fn func(sId int64, e entity.NetworkEntity) bool) {
c.mu.RLock()
defer c.mu.RUnlock()
func (g *Group) PeekMembers(fn func(sId int64, c *Connection) bool) {
g.mu.RLock()
defer g.mu.RUnlock()
for sId, e := range c.conns {
if fn(sId, e) {
for sId, c := range g.conns {
if fn(sId, c) {
break
}
}
}
// Contains check whether a UID is contained in current group or not
func (c *Group) Contains(uid string) bool {
_, ok := c.Member(uid)
func (g *Group) Contains(uid string) bool {
_, ok := g.Member(uid)
return ok
}
// Add session to group
func (c *Group) Add(e entity.NetworkEntity) error {
if c.isClosed() {
func (g *Group) Add(c *Connection) error {
if g.isClosed() {
return ErrClosedGroup
}
c.mu.Lock()
defer c.mu.Unlock()
g.mu.Lock()
defer g.mu.Unlock()
sess := e.Session()
sess := c.Session()
id := sess.ID()
// group attribute
@ -110,41 +109,41 @@ func (c *Group) Add(e entity.NetworkEntity) error {
sess.SetAttribute(groupKey, groups)
}
contains := false
for _, g := range groups {
if g == c.name {
for _, group := range groups {
if group == g.name {
contains = true
break
}
}
if !contains {
groups = append(groups, c.name)
groups = append(groups, g.name)
sess.SetAttribute(groupKey, groups)
}
} else {
sess.SetAttribute(groupKey, []string{c.name})
sess.SetAttribute(groupKey, []string{g.name})
}
if _, ok := c.conns[id]; !ok {
c.conns[id] = e
if _, ok := g.conns[id]; !ok {
g.conns[id] = c
}
nlog.Debugf("Add connection to group %s, ID=%d, UID=%s", c.name, sess.ID(), sess.UID())
nlog.Debugf("Add connection to group %s, ID=%d, UID=%s", g.name, sess.ID(), sess.UID())
return nil
}
// Leave remove specified UID related session from group
func (c *Group) Leave(e entity.NetworkEntity) error {
if c.isClosed() {
func (g *Group) Leave(c *Connection) error {
if g.isClosed() {
return ErrClosedGroup
}
if e == nil {
if c == nil {
return nil
}
sess := e.Session()
nlog.Debugf("Remove connection from group %s, UID=%s", c.name, sess.UID())
sess := c.Session()
nlog.Debugf("Remove connection from group %s, UID=%s", g.name, sess.UID())
c.mu.Lock()
defer c.mu.Unlock()
g.mu.Lock()
defer g.mu.Unlock()
if sess.Exists(groupKey) {
groups, ok := sess.Attribute(groupKey).([]string)
@ -152,7 +151,7 @@ func (c *Group) Leave(e entity.NetworkEntity) error {
groups = make([]string, 0)
sess.SetAttribute(groupKey, groups)
}
groups = c.removeGroupAttr(groups)
groups = g.removeGroupAttr(groups)
if len(groups) == 0 {
sess.RemoveAttribute(groupKey)
@ -161,28 +160,28 @@ func (c *Group) Leave(e entity.NetworkEntity) error {
}
}
delete(c.conns, sess.ID())
delete(g.conns, sess.ID())
return nil
}
func (c *Group) LeaveByUID(uid string) error {
if c.isClosed() {
func (g *Group) LeaveByUID(uid string) error {
if g.isClosed() {
return ErrClosedGroup
}
member, _ := c.Member(uid)
return c.Leave(member)
member, _ := g.Member(uid)
return g.Leave(member)
}
// LeaveAll clear all sessions in the group
func (c *Group) LeaveAll() error {
if c.isClosed() {
func (g *Group) LeaveAll() error {
if g.isClosed() {
return ErrClosedGroup
}
c.mu.Lock()
defer c.mu.Unlock()
g.mu.Lock()
defer g.mu.Unlock()
for _, e := range c.conns {
for _, e := range g.conns {
sess := e.Session()
groups, ok := sess.Attribute(groupKey).([]string)
@ -190,7 +189,7 @@ func (c *Group) LeaveAll() error {
groups = make([]string, 0)
sess.SetAttribute(groupKey, groups)
}
groups = c.removeGroupAttr(groups)
groups = g.removeGroupAttr(groups)
if len(groups) == 0 {
sess.RemoveAttribute(groupKey)
@ -198,15 +197,15 @@ func (c *Group) LeaveAll() error {
sess.SetAttribute(groupKey, groups)
}
}
c.conns = make(map[int64]entity.NetworkEntity)
g.conns = make(map[int64]*Connection)
return nil
}
// 使用移位法移除group中与name匹配的元素
func (c *Group) removeGroupAttr(groups []string) []string {
func (g *Group) removeGroupAttr(groups []string) []string {
j := 0
for _, v := range groups {
if v != c.name {
if v != g.name {
groups[j] = v
j++
}
@ -215,32 +214,32 @@ func (c *Group) removeGroupAttr(groups []string) []string {
}
// Count get current member amount in the group
func (c *Group) Count() int {
c.mu.RLock()
defer c.mu.RUnlock()
func (g *Group) Count() int {
g.mu.RLock()
defer g.mu.RUnlock()
return len(c.conns)
return len(g.conns)
}
func (c *Group) isClosed() bool {
if atomic.LoadInt32(&c.status) == groupStatusClosed {
func (g *Group) isClosed() bool {
if atomic.LoadInt32(&g.status) == groupStatusClosed {
return true
}
return false
}
// Close destroy group, which will release all resource in the group
func (c *Group) Close() error {
if c.isClosed() {
func (g *Group) Close() error {
if g.isClosed() {
return ErrCloseClosedGroup
}
if c.name == DefaultGroupName {
if g.name == DefaultGroupName {
// 默认分组不允许删除
return DeleteDefaultGroupNotAllow
}
_ = c.LeaveAll()
_ = g.LeaveAll()
atomic.StoreInt32(&c.status, groupStatusClosed)
atomic.StoreInt32(&g.status, groupStatusClosed)
return nil
}

@ -1,7 +1,6 @@
package conn
package connection
import (
"git.noahlan.cn/noahlan/nnet/entity"
"sync"
)
@ -11,38 +10,38 @@ type Manager struct {
// 分组
groups map[string]*Group
// 所有 Connection
conns map[int64]entity.NetworkEntity
conns map[int64]*Connection
}
func NewManager() *Manager {
return &Manager{
RWMutex: sync.RWMutex{},
groups: make(map[string]*Group),
conns: make(map[int64]entity.NetworkEntity),
conns: make(map[int64]*Connection),
}
}
// Store 保存连接,同时加入到指定分组,若给定分组名为空,则不进行分组操作
func (m *Manager) Store(groupName string, s entity.NetworkEntity) error {
func (m *Manager) Store(groupName string, c *Connection) error {
m.Lock()
m.conns[s.Session().ID()] = s
m.conns[c.Session().ID()] = c
m.Unlock()
group, ok := m.FindGroup(groupName)
if !ok {
group = m.NewGroup(groupName)
}
return group.Add(s)
return group.Add(c)
}
func (m *Manager) Remove(s entity.NetworkEntity) error {
func (m *Manager) Remove(c *Connection) error {
m.Lock()
defer m.Unlock()
delete(m.conns, s.Session().ID())
delete(m.conns, c.Session().ID())
// 从所有group中移除
for _, group := range m.groups {
err := group.Leave(s)
err := group.Leave(c)
if err != nil {
return err
}
@ -50,9 +49,9 @@ func (m *Manager) Remove(s entity.NetworkEntity) error {
return nil
}
func (m *Manager) RemoveFromGroup(groupName string, s entity.NetworkEntity) error {
func (m *Manager) RemoveFromGroup(groupName string, c *Connection) error {
m.Lock()
delete(m.conns, s.Session().ID())
delete(m.conns, c.Session().ID())
m.Unlock()
group, ok := m.FindGroup(groupName)
@ -60,7 +59,7 @@ func (m *Manager) RemoveFromGroup(groupName string, s entity.NetworkEntity) erro
return nil
}
return group.Leave(s)
return group.Leave(c)
}
// NewGroup 新增分组,若分组已存在,则返回现有分组
@ -89,7 +88,7 @@ func (m *Manager) FindGroup(name string) (*Group, bool) {
}
// FindConn 根据连接ID找到连接
func (m *Manager) FindConn(id int64) (entity.NetworkEntity, bool) {
func (m *Manager) FindConn(id int64) (*Connection, bool) {
m.RLock()
defer m.RUnlock()
@ -98,7 +97,7 @@ func (m *Manager) FindConn(id int64) (entity.NetworkEntity, bool) {
}
// FindConnByUID 根据连接绑定的UID找到连接
func (m *Manager) FindConnByUID(uid string) (entity.NetworkEntity, bool) {
func (m *Manager) FindConnByUID(uid string) (*Connection, bool) {
m.RLock()
defer m.RUnlock()
@ -112,12 +111,12 @@ func (m *Manager) FindConnByUID(uid string) (entity.NetworkEntity, bool) {
// PeekConn 循环所有连接
// fn 返回true跳过循环反之一直循环
func (m *Manager) PeekConn(fn func(id int64, e entity.NetworkEntity) bool) {
func (m *Manager) PeekConn(fn func(id int64, c *Connection) bool) {
m.RLock()
defer m.RUnlock()
for id, e := range m.conns {
if fn(id, e) {
for id, c := range m.conns {
if fn(id, c) {
break
}
}

@ -1,12 +1,11 @@
package pipeline
package connection
import (
"git.noahlan.cn/noahlan/nnet/entity"
"sync"
)
type (
Func func(entity entity.NetworkEntity, v interface{}) error
Func func(c *Connection, v interface{}) error
// Pipeline 消息管道
Pipeline interface {
@ -21,7 +20,7 @@ type (
Channel interface {
PushFront(h Func)
PushBack(h Func)
Process(entity entity.NetworkEntity, v interface{}) error
Process(c *Connection, v interface{}) error
}
pipelineChannel struct {
@ -30,7 +29,7 @@ type (
}
)
func New() Pipeline {
func NewPipeline() Pipeline {
return &pipeline{
outbound: &pipelineChannel{},
inbound: &pipelineChannel{},
@ -66,7 +65,7 @@ func (p *pipelineChannel) PushBack(h Func) {
}
// Process 处理所有的pipeline方法
func (p *pipelineChannel) Process(entity entity.NetworkEntity, v interface{}) error {
func (p *pipelineChannel) Process(c *Connection, v interface{}) error {
if len(p.handlers) < 1 {
return nil
}
@ -75,7 +74,7 @@ func (p *pipelineChannel) Process(entity entity.NetworkEntity, v interface{}) er
defer p.mu.RUnlock()
for _, handler := range p.handlers {
err := handler(entity, v)
err := handler(c, v)
if err != nil {
return err
}

@ -1,4 +1,4 @@
package core
package connection
import (
"github.com/gorilla/websocket"
@ -12,8 +12,8 @@ type WSConn struct {
*websocket.Conn
}
// newWSConn 新建wsConn
func newWSConn(conn *websocket.Conn) *WSConn {
// NewWSConn 新建wsConn
func NewWSConn(conn *websocket.Conn) *WSConn {
return &WSConn{Conn: conn}
}

@ -1,286 +0,0 @@
package core
import (
"errors"
"fmt"
"git.noahlan.cn/noahlan/nnet/config"
conn2 "git.noahlan.cn/noahlan/nnet/conn"
"git.noahlan.cn/noahlan/nnet/entity"
"git.noahlan.cn/noahlan/nnet/lifetime"
"git.noahlan.cn/noahlan/nnet/packet"
"git.noahlan.cn/noahlan/nnet/pipeline"
"git.noahlan.cn/noahlan/nnet/scheduler"
"git.noahlan.cn/noahlan/nnet/serialize"
"git.noahlan.cn/noahlan/ntools-go/core/nlog"
"git.noahlan.cn/noahlan/ntools-go/core/pool"
"github.com/gorilla/websocket"
"log"
"net"
"net/http"
"strings"
"time"
)
func NotFound(conn entity.NetworkEntity, _ packet.IPacket) {
nlog.Error("handler not found")
_ = conn.SendBytes([]byte("handler not found"))
}
func NotFoundHandler() Handler {
return HandlerFunc(NotFound)
}
type (
// engine TCP-engine
engine struct {
conf config.EngineConf // conf 配置
taskTimerPrecision time.Duration
middlewares []Middleware // 中间件
routes []Route // 路由
// handler 消息处理器
handler Handler
// dieChan 应用程序退出信号
dieChan chan struct{}
pipeline pipeline.Pipeline // 消息管道
lifetime *lifetime.Mgr // 连接的生命周期管理器
packerFn packet.NewPackerFunc // 封包、拆包器
serializer serialize.Serializer // 消息 序列化/反序列化
wsOpt wsOptions // websocket
connManager *conn2.Manager
sessIdMgr *sessionIDMgr
}
wsOptions struct {
IsWebsocket bool // 是否为websocket服务端
WebsocketPath string // ws地址(ws://127.0.0.1/WebsocketPath)
TLSCertificate string // TLS 证书地址 (websocket)
TLSKey string // TLS 证书key地址 (websocket)
CheckOrigin func(*http.Request) bool // check origin
}
)
func newEngine(conf config.EngineConf) *engine {
s := &engine{
conf: conf,
dieChan: make(chan struct{}),
pipeline: pipeline.New(),
middlewares: make([]Middleware, 0),
routes: make([]Route, 0),
taskTimerPrecision: conf.TaskTimerPrecision,
connManager: conn2.NewManager(),
sessIdMgr: newSessionIDMgr(),
lifetime: lifetime.NewLifetime(),
}
pool.InitPool(conf.Pool)
return s
}
func (ng *engine) shallLogDebug() bool {
return config.ShallLogDebug(ng.conf.Mode)
}
func (ng *engine) logPrefix() string {
return fmt.Sprintf("[NNet-%s]", ng.conf.Name)
}
func (ng *engine) use(middleware ...Middleware) {
ng.middlewares = append(ng.middlewares, middleware...)
}
func (ng *engine) addRoutes(route ...Route) {
ng.routes = append(ng.routes, route...)
}
func (ng *engine) bindRoutes(router Router) error {
for _, fr := range ng.routes {
if err := ng.bindRoute(router, fr); err != nil {
return err
}
}
return nil
}
func (ng *engine) bindRoute(router Router, route Route) error {
// TODO default middleware
chain := newChain()
// build chain
for _, middleware := range ng.middlewares {
chain.Append(convertMiddleware(middleware))
}
return router.Register(route.Matches, route.Handler)
}
func convertMiddleware(ware Middleware) func(Handler) Handler {
return func(next Handler) Handler {
return ware(next.Handle)
}
}
func (ng *engine) dial(addr string, router Router) (entity.NetworkEntity, error) {
ng.handler = router
if err := ng.bindRoutes(router); err != nil {
return nil, err
}
go scheduler.Schedule(ng.taskTimerPrecision)
// connection
conn, err := net.Dial("tcp", addr)
nlog.Must(err)
c := newConnection(ng, conn)
c.serve()
// hook
ng.lifetime.Open(c)
// connection manager
err = ng.connManager.Store(conn2.DefaultGroupName, c)
nlog.Must(err)
// 连接成功,客户端已启动
if ng.shallLogDebug() {
nlog.Debugf("now connect to %s.", addr)
}
return c, nil
}
func (ng *engine) serve(router Router) error {
ng.handler = router
if err := ng.bindRoutes(router); err != nil {
return err
}
go scheduler.Schedule(ng.taskTimerPrecision)
defer func() {
nlog.Infof("%s is stopping...", ng.logPrefix())
ng.shutdown()
scheduler.Close()
}()
if ng.wsOpt.IsWebsocket {
if len(ng.wsOpt.TLSCertificate) != 0 && len(ng.wsOpt.TLSKey) != 0 {
ng.listenAndServeWSTLS()
} else {
ng.listenAndServeWS()
}
} else {
ng.listenAndServe()
}
return nil
}
func (ng *engine) close() {
close(ng.dieChan)
}
func (ng *engine) shutdown() {
}
func (ng *engine) listenAndServe() {
listener, err := net.Listen(ng.conf.Protocol, ng.conf.Addr)
nlog.Must(err)
// 监听成功,服务已启动
if ng.shallLogDebug() {
nlog.Debugf("%s now listening %s at %s.", ng.logPrefix(), ng.conf.Protocol, ng.conf.Addr)
}
defer func() {
_ = listener.Close()
ng.close()
}()
for {
conn, err := listener.Accept()
if err != nil {
if errors.Is(err, net.ErrClosed) {
nlog.Errorf("%s 服务器网络错误 %+v", ng.logPrefix(), err)
return
}
nlog.Errorf("%s 监听错误 %v", ng.logPrefix(), err)
continue
}
err = pool.Submit(func() {
ng.handle(conn)
})
if err != nil {
nlog.Errorf("%s submit conn pool err: %ng", ng.logPrefix(), err.Error())
continue
}
}
}
func (ng *engine) listenAndServeWS() {
ng.setupWS()
if ng.shallLogDebug() {
nlog.Debugf("%s now listening websocket at %s.", ng.logPrefix(), ng.conf.Addr)
}
if err := http.ListenAndServe(ng.conf.Addr, nil); err != nil {
log.Fatal(err.Error())
}
}
func (ng *engine) listenAndServeWSTLS() {
ng.setupWS()
if ng.shallLogDebug() {
nlog.Debugf("%s now listening websocket with tls at %s.", ng.logPrefix(), ng.conf.Addr)
}
if err := http.ListenAndServeTLS(ng.conf.Addr, ng.wsOpt.TLSCertificate, ng.wsOpt.TLSKey, nil); err != nil {
log.Fatal(err.Error())
}
}
func (ng *engine) setupWS() {
upgrade := websocket.Upgrader{
ReadBufferSize: 1024,
WriteBufferSize: 1024,
CheckOrigin: ng.wsOpt.CheckOrigin,
}
http.HandleFunc("/"+strings.TrimPrefix(ng.wsOpt.WebsocketPath, "/"), func(w http.ResponseWriter, r *http.Request) {
conn, err := upgrade.Upgrade(w, r, nil)
if err != nil {
nlog.Errorf("%s Upgrade failure, URI=%ng, Error=%ng", ng.logPrefix(), r.RequestURI, err.Error())
return
}
err = pool.Submit(func() {
ng.handleWS(conn)
})
if err != nil {
log.Fatalf("%s submit conn pool err: %v", ng.logPrefix(), err.Error())
}
})
}
func (ng *engine) handleWS(conn *websocket.Conn) {
c := newWSConn(conn)
ng.handle(c)
}
func (ng *engine) handle(conn net.Conn) {
c := newConnection(ng, conn)
err := ng.connManager.Store(conn2.DefaultGroupName, c)
nlog.Must(err)
c.serve()
// hook
ng.lifetime.Open(c)
}
func (ng *engine) notFoundHandler(next Handler) Handler {
return HandlerFunc(func(entity entity.NetworkEntity, packet packet.IPacket) {
h := next
if next == nil {
h = NotFoundHandler()
}
// TODO write to client
h.Handle(entity, packet)
})
}

@ -1,249 +0,0 @@
package core
import (
"git.noahlan.cn/noahlan/nnet/config"
"git.noahlan.cn/noahlan/nnet/conn"
"git.noahlan.cn/noahlan/nnet/entity"
"git.noahlan.cn/noahlan/nnet/lifetime"
"git.noahlan.cn/noahlan/nnet/packet"
"git.noahlan.cn/noahlan/nnet/pipeline"
"git.noahlan.cn/noahlan/nnet/serialize"
"git.noahlan.cn/noahlan/ntools-go/core/nlog"
"net/http"
"time"
)
type (
// RunOption defines the method to customize a NNet.
RunOption func(*NNet)
Server struct {
*NNet
}
Client struct {
*NNet
}
NNet struct {
ngin *engine
router Router
}
)
// NewServer returns a server with given config of c and options defined in opts.
// Be aware that later RunOption might overwrite previous one that write the same option.
func NewServer(c config.EngineConf, opts ...RunOption) *Server {
s := &Server{
NNet: &NNet{
ngin: newEngine(c),
router: NewDefaultRouter(),
},
}
opts = append([]RunOption{WithNotFoundHandler(nil)}, opts...)
for _, opt := range opts {
opt(s.NNet)
}
return s
}
// NewClient returns a client with given config of c and options defined in opts.
// Be aware that later RunOption might overwrite previous one that write the same option.
func NewClient(c config.EngineConf, opts ...RunOption) *Client {
s := &Client{
NNet: &NNet{
ngin: newEngine(c),
router: NewDefaultRouter(),
},
}
opts = append([]RunOption{WithNotFoundHandler(nil)}, opts...)
for _, opt := range opts {
opt(s.NNet)
}
return s
}
// Start starts the NNet.
// Graceful shutdown is enabled by default.
func (s *Server) Start() {
if err := s.ngin.serve(s.router); err != nil {
nlog.Error(err)
panic(err)
}
}
// Dial start the NNet client.
// Graceful shutdown is enabled by default.
func (c *Client) Dial(addr string) entity.NetworkEntity {
e, err := c.ngin.dial(addr, c.router)
nlog.Must(err)
return e
}
// AddRoutes add given routes into the NNet.
func (s *NNet) AddRoutes(rs []Route) {
s.ngin.addRoutes(rs...)
err := s.ngin.bindRoutes(s.router)
nlog.Must(err)
}
// AddRoute adds given route into the NNet.
func (s *NNet) AddRoute(r Route) {
s.AddRoutes([]Route{r})
}
// Stop stops the NNet.
func (s *NNet) Stop() {
s.ngin.close()
}
// Use adds the given middleware in the NNet.
func (s *NNet) Use(middleware ...Middleware) {
s.ngin.use(middleware...)
}
// Pipeline returns inner pipeline
func (s *NNet) Pipeline() pipeline.Pipeline {
return s.ngin.pipeline
}
// Lifetime returns lifetime interface.
func (s *NNet) Lifetime() lifetime.Lifetime {
return s.ngin.lifetime
}
// ConnManager returns connection manager
func (s *NNet) ConnManager() *conn.Manager {
return s.ngin.connManager
}
// ToMiddleware converts the given handler to a Middleware.
func ToMiddleware(handler func(next Handler) Handler) Middleware {
return func(next HandlerFunc) HandlerFunc {
return handler(next).Handle
}
}
// WithMiddlewares adds given middlewares to given routes.
func WithMiddlewares(ms []Middleware, rs ...Route) []Route {
for i := len(ms) - 1; i >= 0; i-- {
rs = WithMiddleware(ms[i], rs...)
}
return rs
}
// WithMiddleware adds given middleware to given route.
func WithMiddleware(middleware Middleware, rs ...Route) []Route {
routes := make([]Route, len(rs))
for i := range rs {
route := rs[i]
routes[i] = Route{
Matches: route.Matches,
Handler: middleware(route.Handler),
}
}
return routes
}
func UseMiddleware(middleware ...Middleware) RunOption {
return func(server *NNet) {
server.Use(middleware...)
}
}
// WithNotFoundHandler returns a RunOption with not found handler set to given handler.
func WithNotFoundHandler(handler Handler) RunOption {
return func(server *NNet) {
notFoundHandler := server.ngin.notFoundHandler(handler)
server.router.SetNotFoundHandler(notFoundHandler)
}
}
// WithRouter 设置消息路由
func WithRouter(router Router) RunOption {
return func(server *NNet) {
server.router = router
}
}
// WithPacker 设置消息的 封包/解包 方式
func WithPacker(fn packet.NewPackerFunc) RunOption {
return func(server *NNet) {
server.ngin.packerFn = fn
}
}
// WithSerializer 设置消息的 序列化/反序列化 方式
func WithSerializer(s serialize.Serializer) RunOption {
return func(server *NNet) {
server.ngin.serializer = s
}
}
// WithTimerPrecision 设置Timer精度需在 Start 或 Dial 之前执行
// 注精度需大于1ms, 并且不能在运行时更改
// 默认精度是 time.Second
func WithTimerPrecision(precision time.Duration) RunOption {
if precision < time.Millisecond {
panic("time precision can not less than a Millisecond")
}
return func(s *NNet) {
s.ngin.taskTimerPrecision = precision
}
}
func WithPipeline(pipeline pipeline.Pipeline) RunOption {
return func(server *NNet) {
server.ngin.pipeline = pipeline
}
}
type PipelineOption func(opts pipeline.Pipeline)
func WithPipelineOpt(opts ...func(pipeline.Pipeline)) RunOption {
return func(server *NNet) {
for _, opt := range opts {
opt(server.ngin.pipeline)
}
}
}
type WSOption func(opts *wsOptions)
// WithWebsocket 开启Websocket, 参数是websocket的相关参数 nnet.WSOption
func WithWebsocket(wsOpts ...WSOption) RunOption {
return func(server *NNet) {
for _, opt := range wsOpts {
opt(&server.ngin.wsOpt)
}
server.ngin.wsOpt.IsWebsocket = true
}
}
// WithWSPath 设置websocket的path
func WithWSPath(path string) WSOption {
return func(opts *wsOptions) {
opts.WebsocketPath = path
}
}
// WithWSTLSConfig 设置websocket的证书和密钥
func WithWSTLSConfig(certificate, key string) WSOption {
return func(opts *wsOptions) {
opts.TLSCertificate = certificate
opts.TLSKey = key
}
}
func WithWSCheckOriginFunc(fn func(*http.Request) bool) WSOption {
return func(opts *wsOptions) {
if fn != nil {
opts.CheckOrigin = fn
}
}
}

@ -1,66 +0,0 @@
package core
import (
"fmt"
"git.noahlan.cn/noahlan/nnet/config"
"git.noahlan.cn/noahlan/nnet/entity"
"git.noahlan.cn/noahlan/nnet/packet"
"git.noahlan.cn/noahlan/nnet/protocol/nnet"
"git.noahlan.cn/noahlan/ntools-go/core/nlog"
"git.noahlan.cn/noahlan/ntools-go/core/pool"
"math"
"testing"
"time"
)
func TestServer(t *testing.T) {
server := NewServer(config.EngineConf{
ServerConf: config.ServerConf{
Protocol: "tcp",
Addr: "0.0.0.0:6666",
Name: "testServer",
Mode: "dev",
},
Pool: pool.Config{
PoolSize: math.MaxInt32,
ExpiryDuration: time.Second,
PreAlloc: false,
MaxBlockingTasks: 0,
Nonblocking: false,
DisablePurge: false,
},
}, nnet.WithNNetProtocol(nnet.Config{
HeartbeatInterval: 0,
HandshakeValidator: nil,
})...)
server.AddRoute(Route{
Matches: nnet.Match{
Route: "test",
Code: 1,
},
Handler: func(entity entity.NetworkEntity, pkg packet.IPacket) {
fmt.Println(pkg)
p, ok := pkg.(*nnet.Packet)
if !ok {
nlog.Error("wrong packet type")
return
}
bd := []byte("服务器接收到数据为: " + string(p.GetBody()))
// 注Response类型数据不需要Route原地返回客户端需等待
_ = entity.Send(nnet.Header{
PacketType: nnet.Data,
Length: uint32(len(bd)),
MessageHeader: nnet.MessageHeader{
MsgType: nnet.Response,
ID: p.ID,
Route: p.Route,
},
}, bd)
},
})
defer server.Stop()
server.Start()
}

@ -1,146 +0,0 @@
package core
import (
"git.noahlan.cn/noahlan/nnet/entity"
"sync"
"sync/atomic"
)
type session struct {
sync.RWMutex // 数据锁
// 网络单元
entity entity.NetworkEntity
id int64 // Session全局唯一ID
uid string // 用户ID不绑定的情况下与sid一致
data map[string]interface{} // session数据存储内存
}
func newSession(entity entity.NetworkEntity, id int64) *session {
return &session{
id: id,
entity: entity,
uid: "",
data: make(map[string]interface{}),
}
}
// ID 获取 session ID
func (s *session) ID() int64 {
return s.id
}
// UID 获取UID
func (s *session) UID() string {
return s.uid
}
// Bind 绑定uid
func (s *session) Bind(uid string) {
s.uid = uid
}
// Attribute 获取指定key对应参数
func (s *session) Attribute(key string) interface{} {
s.RLock()
defer s.RUnlock()
return s.data[key]
}
// Keys 获取所有参数key
func (s *session) Keys() []string {
s.RLock()
defer s.RUnlock()
keys := make([]string, 0, len(s.data))
for k := range s.data {
keys = append(keys, k)
}
return keys
}
// Exists 指定key是否存在
func (s *session) Exists(key string) bool {
s.RLock()
defer s.RUnlock()
_, has := s.data[key]
return has
}
// Attributes 获取所有参数
func (s *session) Attributes() map[string]interface{} {
s.RLock()
defer s.RUnlock()
return s.data
}
// RemoveAttribute 移除指定key对应参数
func (s *session) RemoveAttribute(key string) {
s.Lock()
defer s.Unlock()
delete(s.data, key)
}
// SetAttribute 设置参数
func (s *session) SetAttribute(key string, value interface{}) {
s.Lock()
defer s.Unlock()
s.data[key] = value
}
// Invalidate 清理
func (s *session) Invalidate() {
s.Lock()
defer s.Unlock()
s.id = 0
s.uid = ""
s.data = make(map[string]interface{})
}
// Close 关闭
func (s *session) Close() {
//s.entity.Close()
s.Invalidate()
}
type sessionIDMgr struct {
count int64
sid int64
}
func newSessionIDMgr() *sessionIDMgr {
return &sessionIDMgr{}
}
// Increment the connection count
func (c *sessionIDMgr) Increment() {
atomic.AddInt64(&c.count, 1)
}
// Decrement the connection count
func (c *sessionIDMgr) Decrement() {
atomic.AddInt64(&c.count, -1)
}
// Count returns the connection numbers in current
func (c *sessionIDMgr) Count() int64 {
return atomic.LoadInt64(&c.count)
}
// Reset the connection service status
func (c *sessionIDMgr) Reset() {
atomic.StoreInt64(&c.count, 0)
atomic.StoreInt64(&c.sid, 0)
}
// SessionID returns the session id
func (c *sessionIDMgr) SessionID() int64 {
return atomic.AddInt64(&c.sid, 1)
}

@ -1,20 +0,0 @@
package core
import (
"git.noahlan.cn/noahlan/nnet/entity"
"git.noahlan.cn/noahlan/nnet/packet"
)
type (
Handler interface {
Handle(entity entity.NetworkEntity, pkg packet.IPacket)
}
HandlerFunc func(entity entity.NetworkEntity, pkg packet.IPacket)
Middleware func(next HandlerFunc) HandlerFunc
)
func (f HandlerFunc) Handle(entity entity.NetworkEntity, pkg packet.IPacket) {
f(entity, pkg)
}

@ -0,0 +1,135 @@
package nnet
import (
"git.noahlan.cn/noahlan/nnet/config"
"git.noahlan.cn/noahlan/nnet/connection"
"git.noahlan.cn/noahlan/nnet/lifetime"
"git.noahlan.cn/noahlan/nnet/packet"
rt "git.noahlan.cn/noahlan/nnet/router"
"git.noahlan.cn/noahlan/nnet/scheduler"
"git.noahlan.cn/noahlan/nnet/serialize"
"git.noahlan.cn/noahlan/nnet/session"
"git.noahlan.cn/noahlan/ntools-go/core/nlog"
"github.com/panjf2000/ants/v2"
"math"
"net"
)
// Engine 引擎
type Engine struct {
config.EngineConf // 引擎配置
middlewares []rt.Middleware // 中间件
routes []rt.Route // 路由
router rt.Router // 消息处理器
dieChan chan struct{} // 应用程序退出信号
pipeline connection.Pipeline // 消息管道
packerBuilder packet.PackerBuilder // 封包、拆包器
serializer serialize.Serializer // 消息 序列化/反序列化
goPool *ants.Pool // goroutine池
connManager *connection.Manager // 连接管理器
lifetime *lifetime.Mgr // 生命周期
sessIdMgr *session.IDMgr // SessionId管理器
}
func NewEngine(conf config.EngineConf, opts ...RunOption) *Engine {
ngin := &Engine{
EngineConf: conf,
middlewares: make([]rt.Middleware, 0),
routes: make([]rt.Route, 0),
router: rt.NewDefaultRouter(),
packerBuilder: nil,
serializer: nil,
dieChan: make(chan struct{}),
pipeline: connection.NewPipeline(),
connManager: connection.NewManager(),
lifetime: lifetime.NewLifetime(),
sessIdMgr: session.NewSessionIDMgr(),
goPool: nil,
}
for _, opt := range opts {
opt(ngin)
}
if ngin.goPool == nil {
ngin.goPool, _ = ants.NewPool(math.MaxInt32)
}
return ngin
}
func (ngin *Engine) Use(middleware ...rt.Middleware) {
ngin.middlewares = append(ngin.middlewares, middleware...)
}
func (ngin *Engine) AddRoutes(rs ...rt.Route) {
ngin.routes = append(ngin.routes, rs...)
err := ngin.bindRoutes()
nlog.Must(err)
}
func (ngin *Engine) bindRoutes() error {
for _, fr := range ngin.routes {
if err := ngin.bindRoute(fr); err != nil {
return err
}
}
return nil
}
func (ngin *Engine) bindRoute(route rt.Route) error {
// TODO default middleware
chain := rt.NewChain()
// build chain
for _, middleware := range ngin.middlewares {
chain.Append(rt.ConvertMiddleware(middleware))
}
return ngin.router.Register(route.Matches, route.Handler)
}
func (ngin *Engine) setup() error {
if err := ngin.bindRoutes(); err != nil {
return err
}
if err := ngin.goPool.Submit(func() {
scheduler.Schedule(ngin.TaskTimerPrecision)
}); err != nil {
return err
}
return nil
}
func (ngin *Engine) Stop() {
nlog.Infof("%s is stopping...", ngin.LogPrefix())
close(ngin.dieChan)
scheduler.Close()
}
func (ngin *Engine) handle(conn net.Conn) *connection.Connection {
nc := connection.NewConnection(
ngin.sessIdMgr.SessionID(),
conn,
connection.Config{LogDebug: ngin.ShallLogDebug(), LogPrefix: ngin.LogPrefix()},
ngin.packerBuilder, ngin.serializer, ngin.pipeline,
ngin.router.Handle,
)
nc.Serve()
err := ngin.connManager.Store(connection.DefaultGroupName, nc)
nlog.Must(err)
// dieChan
go func() {
// lifetime
ngin.lifetime.Open(nc)
select {
case <-nc.DieChan():
scheduler.PushTask(func() { ngin.lifetime.Close(nc) })
_ = ngin.connManager.Remove(nc)
}
}()
return nc
}

@ -1,24 +0,0 @@
package entity
import "net"
type NetworkEntity interface {
// Send 主动发送消息支持自定义header,payload
Send(header, payload interface{}) error
// SendBytes 主动发送消息,消息需提前编码
SendBytes(data []byte) error
// Status 获取当前连接状态
Status() int32
// SetStatus 设置当前连接状态
SetStatus(s int32)
// Conn 获取当前底层连接还需根据返回参数2决定是否转换为WSConn
Conn() (net.Conn, bool)
// Session 获取当前连接 Session
Session() Session
// LastMID 最新消息ID
LastMID() uint64
// SetLastMID 设置消息ID
SetLastMID(mid uint64)
// Close 关闭连接
Close() error
}

@ -1,26 +0,0 @@
package entity
type Session interface {
// ID 获取 Session ID
ID() int64
// UID 获取UID
UID() string
// Bind 绑定uid
Bind(uid string)
// Attribute 获取指定key对应参数
Attribute(key string) interface{}
// Keys 获取所有参数key
Keys() []string
// Exists 指定key是否存在
Exists(key string) bool
// Attributes 获取所有参数
Attributes() map[string]interface{}
// RemoveAttribute 移除指定key对应参数
RemoveAttribute(key string)
// SetAttribute 设置参数
SetAttribute(key string, value interface{})
// Invalidate 清理
Invalidate()
// Close 关闭 Session
Close()
}

@ -1,9 +1,11 @@
package lifetime
import "git.noahlan.cn/noahlan/nnet/entity"
import (
"git.noahlan.cn/noahlan/nnet/connection"
)
type (
Handler func(entity entity.NetworkEntity)
Handler func(conn *connection.Connection)
Lifetime interface {
OnClosed(h Handler)
@ -31,22 +33,22 @@ func (lt *Mgr) OnOpen(h Handler) {
lt.onOpen = append(lt.onOpen, h)
}
func (lt *Mgr) Open(entity entity.NetworkEntity) {
func (lt *Mgr) Open(conn *connection.Connection) {
if len(lt.onOpen) <= 0 {
return
}
for _, handler := range lt.onOpen {
handler(entity)
handler(conn)
}
}
func (lt *Mgr) Close(entity entity.NetworkEntity) {
func (lt *Mgr) Close(conn *connection.Connection) {
if len(lt.onClosed) <= 0 {
return
}
for _, handler := range lt.onClosed {
handler(entity)
handler(conn)
}
}

@ -1,9 +1,10 @@
package middleware
import (
"git.noahlan.cn/noahlan/nnet/core"
"git.noahlan.cn/noahlan/nnet/entity"
"git.noahlan.cn/noahlan/nnet"
"git.noahlan.cn/noahlan/nnet/connection"
"git.noahlan.cn/noahlan/nnet/packet"
rt "git.noahlan.cn/noahlan/nnet/router"
"git.noahlan.cn/noahlan/ntools-go/core/nlog"
"sync/atomic"
"time"
@ -12,10 +13,10 @@ import (
type HeartbeatMiddleware struct {
lastAt int64
interval time.Duration
hbdFn func(entity entity.NetworkEntity) []byte
hbdFn func(conn *connection.Connection) []byte
}
func WithHeartbeat(interval time.Duration, hbdFn func(entity entity.NetworkEntity) []byte) core.RunOption {
func WithHeartbeat(interval time.Duration, hbdFn func(conn *connection.Connection) []byte) nnet.RunOption {
m := &HeartbeatMiddleware{
lastAt: time.Now().Unix(),
interval: interval,
@ -26,20 +27,20 @@ func WithHeartbeat(interval time.Duration, hbdFn func(entity entity.NetworkEntit
panic("dataFn must not be nil")
}
return func(server *core.NNet) {
server.Lifetime().OnOpen(m.start)
return func(ngin *nnet.Engine) {
ngin.Lifetime().OnOpen(m.start)
server.Use(func(next core.HandlerFunc) core.HandlerFunc {
return func(entity entity.NetworkEntity, pkg packet.IPacket) {
m.handle(entity, pkg)
ngin.Use(func(next rt.HandlerFunc) rt.HandlerFunc {
return func(conn *connection.Connection, pkg packet.IPacket) {
m.handle(conn, pkg)
next(entity, pkg)
next(conn, pkg)
}
})
}
}
func (m *HeartbeatMiddleware) start(entity entity.NetworkEntity) {
func (m *HeartbeatMiddleware) start(conn *connection.Connection) {
ticker := time.NewTicker(m.interval)
defer func() {
@ -54,7 +55,7 @@ func (m *HeartbeatMiddleware) start(entity entity.NetworkEntity) {
nlog.Errorf("Heartbeat timeout, LastTime=%d, Deadline=%d", atomic.LoadInt64(&m.lastAt), deadline)
return
}
err := entity.SendBytes(m.hbdFn(entity))
err := conn.SendBytes(m.hbdFn(conn))
if err != nil {
nlog.Errorf("Heartbeat err: %v", err)
return
@ -63,6 +64,6 @@ func (m *HeartbeatMiddleware) start(entity entity.NetworkEntity) {
}
}
func (m *HeartbeatMiddleware) handle(_ entity.NetworkEntity, _ packet.IPacket) {
func (m *HeartbeatMiddleware) handle(_ *connection.Connection, _ packet.IPacket) {
atomic.StoreInt64(&m.lastAt, time.Now().Unix())
}

@ -0,0 +1,114 @@
package nnet
import (
"git.noahlan.cn/noahlan/nnet/connection"
"git.noahlan.cn/noahlan/nnet/lifetime"
"git.noahlan.cn/noahlan/nnet/packet"
rt "git.noahlan.cn/noahlan/nnet/router"
"git.noahlan.cn/noahlan/nnet/serialize"
"git.noahlan.cn/noahlan/ntools-go/core/pool"
"github.com/panjf2000/ants/v2"
"time"
)
type (
// RunOption defines the method to customize an Engine.
RunOption func(ngin *Engine)
)
// Pipeline returns inner pipeline
func (ngin *Engine) Pipeline() connection.Pipeline {
return ngin.pipeline
}
// Lifetime returns lifetime interface.
func (ngin *Engine) Lifetime() lifetime.Lifetime {
return ngin.lifetime
}
// ConnManager returns connection manager
func (ngin *Engine) ConnManager() *connection.Manager {
return ngin.connManager
}
//////////////////////// Options
func WithMiddleware(middleware ...rt.Middleware) RunOption {
return func(ngin *Engine) {
ngin.Use(middleware...)
}
}
// WithRouter 设置消息路由
func WithRouter(router rt.Router) RunOption {
return func(ngin *Engine) {
ngin.router = router
}
}
// WithNotFoundHandler returns a RunOption with not found handler set to given handler.
func WithNotFoundHandler(handler rt.Handler) RunOption {
return func(ngin *Engine) {
ngin.router.SetNotFoundHandler(rt.NotFoundHandler(handler))
}
}
// WithTimerPrecision 设置Timer精度需在 Start 或 Dial 之前执行
// 注精度需大于1ms, 并且不能在运行时更改
// 默认精度是 time.Second
func WithTimerPrecision(precision time.Duration) RunOption {
if precision < time.Millisecond {
panic("time precision can not less than a Millisecond")
}
return func(ngin *Engine) {
ngin.TaskTimerPrecision = precision
}
}
// WithPackerBuilder 设置 消息的封包/解包构造器
func WithPackerBuilder(fn packet.PackerBuilder) RunOption {
return func(ngin *Engine) {
ngin.packerBuilder = fn
}
}
// WithSerializer 设置消息的 序列化/反序列化 方式
func WithSerializer(s serialize.Serializer) RunOption {
return func(ngin *Engine) {
ngin.serializer = s
}
}
// WithPool 设置使用自定义的工作池
func WithPool(pl *ants.Pool) RunOption {
return func(ngin *Engine) {
ngin.goPool = pl
}
}
// WithPoolCfg 设置工作池配置
func WithPoolCfg(cfg pool.Config) RunOption {
return func(ngin *Engine) {
ngin.goPool, _ = ants.NewPool(cfg.PoolSize, ants.WithOptions(cfg.Options()))
}
}
//////////////////// Pipeline
// WithPipeline 使用自定义 pipeline
func WithPipeline(pipeline connection.Pipeline) RunOption {
return func(ngin *Engine) {
ngin.pipeline = pipeline
}
}
type PipelineOption func(opts connection.Pipeline)
// WithPipelineOpt 使用默认Pipeline并设置其配置
func WithPipelineOpt(opts ...func(connection.Pipeline)) RunOption {
return func(ngin *Engine) {
for _, opt := range opts {
opt(ngin.pipeline)
}
}
}

@ -24,5 +24,6 @@ type (
Unpack(data []byte) ([]IPacket, error)
}
NewPackerFunc func() Packer
// PackerBuilder Packer构建器
PackerBuilder func() Packer
)

@ -0,0 +1,43 @@
package modbus
import "sync"
var crcTable []uint16
var mu sync.Mutex
// crcModbus 计算modbus的crc
func crcModbus(data []byte) (crc uint16) {
if crcTable == nil {
mu.Lock()
if crcTable == nil {
initCrcTable()
}
mu.Unlock()
}
crc = 0xffff
for _, v := range data {
crc = (crc >> 8) ^ crcTable[(crc^uint16(v))&0x00FF]
}
return crc
}
// initCrcTable 初始化crcTable
func initCrcTable() {
crc16IBM := uint16(0xA001)
crcTable = make([]uint16, 256)
for i := uint16(0); i < 256; i++ {
crc := uint16(0)
c := i
for j := uint16(0); j < 8; j++ {
if ((crc ^ c) & 0x0001) > 0 {
crc = (crc >> 1) ^ crc16IBM
} else {
crc = crc >> 1
}
c = c >> 1
}
crcTable[i] = crc
}
}

@ -0,0 +1,14 @@
package modbus
import (
"git.noahlan.cn/noahlan/nnet/protocol/modbus/internal"
"testing"
)
func TestCRC(t *testing.T) {
got := crcModbus([]byte{0x01, 0x04, 0x02, 0xFF, 0xFF})
expect := uint16(0x80B8)
assert := internal.NewAssert(t, "TestCRC")
assert.Equal(expect, got)
}

@ -0,0 +1,167 @@
package internal
import (
"fmt"
"reflect"
"runtime"
"testing"
)
const (
compareNotEqual int = iota - 2
compareLess
compareEqual
compareGreater
)
// Assert is a simple implementation of assertion, only for internal usage
type Assert struct {
T *testing.T
CaseName string
}
// NewAssert return instance of Assert
func NewAssert(t *testing.T, caseName string) *Assert {
return &Assert{T: t, CaseName: caseName}
}
// Equal check if expected is equal with actual
func (a *Assert) Equal(expected, actual any) {
if compare(expected, actual) != compareEqual {
makeTestFailed(a.T, a.CaseName, expected, actual)
}
}
// NotEqual check if expected is not equal with actual
func (a *Assert) NotEqual(expected, actual any) {
if compare(expected, actual) == compareEqual {
expectedInfo := fmt.Sprintf("not %v", expected)
makeTestFailed(a.T, a.CaseName, expectedInfo, actual)
}
}
// Greater check if expected is greate than actual
func (a *Assert) Greater(expected, actual any) {
if compare(expected, actual) != compareGreater {
expectedInfo := fmt.Sprintf("> %v", expected)
makeTestFailed(a.T, a.CaseName, expectedInfo, actual)
}
}
// GreaterOrEqual check if expected is greate than or equal with actual
func (a *Assert) GreaterOrEqual(expected, actual any) {
isGreatOrEqual := compare(expected, actual) == compareGreater || compare(expected, actual) == compareEqual
if !isGreatOrEqual {
expectedInfo := fmt.Sprintf(">= %v", expected)
makeTestFailed(a.T, a.CaseName, expectedInfo, actual)
}
}
// Less check if expected is less than actual
func (a *Assert) Less(expected, actual any) {
if compare(expected, actual) != compareLess {
expectedInfo := fmt.Sprintf("< %v", expected)
makeTestFailed(a.T, a.CaseName, expectedInfo, actual)
}
}
// LessOrEqual check if expected is less than or equal with actual
func (a *Assert) LessOrEqual(expected, actual any) {
isLessOrEqual := compare(expected, actual) == compareLess || compare(expected, actual) == compareEqual
if !isLessOrEqual {
expectedInfo := fmt.Sprintf("<= %v", expected)
makeTestFailed(a.T, a.CaseName, expectedInfo, actual)
}
}
// IsNil check if value is nil
func (a *Assert) IsNil(value any) {
if value != nil {
makeTestFailed(a.T, a.CaseName, nil, value)
}
}
// IsNotNil check if value is not nil
func (a *Assert) IsNotNil(value any) {
if value == nil {
makeTestFailed(a.T, a.CaseName, "not nil", value)
}
}
// compare x and y return :
// x > y -> 1, x < y -> -1, x == y -> 0, x != y -> -2
func compare(x, y any) int {
vx := reflect.ValueOf(x)
vy := reflect.ValueOf(y)
if vx.Type() != vy.Type() {
return compareNotEqual
}
switch vx.Kind() {
case reflect.Int, reflect.Int8, reflect.Int16, reflect.Int32, reflect.Int64:
xInt := vx.Int()
yInt := vy.Int()
if xInt > yInt {
return compareGreater
}
if xInt == yInt {
return compareEqual
}
if xInt < yInt {
return compareLess
}
case reflect.Uint, reflect.Uint8, reflect.Uint16, reflect.Uint32, reflect.Uint64:
xUint := vx.Uint()
yUint := vy.Uint()
if xUint > yUint {
return compareGreater
}
if xUint == yUint {
return compareEqual
}
if xUint < yUint {
return compareLess
}
case reflect.Float32, reflect.Float64:
xFloat := vx.Float()
yFloat := vy.Float()
if xFloat > yFloat {
return compareGreater
}
if xFloat == yFloat {
return compareEqual
}
if xFloat < yFloat {
return compareLess
}
case reflect.String:
xString := vx.String()
yString := vy.String()
if xString > yString {
return compareGreater
}
if xString == yString {
return compareEqual
}
if xString < yString {
return compareLess
}
default:
if reflect.DeepEqual(x, y) {
return compareEqual
}
return compareNotEqual
}
return compareNotEqual
}
// logFailedInfo make test failed and log error info
func makeTestFailed(t *testing.T, caseName string, expected, actual any) {
_, file, line, _ := runtime.Caller(2)
errInfo := fmt.Sprintf("Case %v failed. file: %v, line: %v, expected: %v, actual: %v.", caseName, file, line, expected, actual)
t.Error(errInfo)
t.FailNow()
}

@ -0,0 +1,50 @@
package internal
import (
"testing"
)
func TestAssert(t *testing.T) {
assert := NewAssert(t, "TestAssert")
assert.Equal(0, 0)
assert.NotEqual(1, 0)
assert.NotEqual("1", 1)
var uInt1 uint
var uInt2 uint
var uInt8 uint8
var uInt16 uint16
var uInt32 uint32
var uInt64 uint64
assert.NotEqual(uInt1, uInt8)
assert.NotEqual(uInt8, uInt16)
assert.NotEqual(uInt16, uInt32)
assert.NotEqual(uInt32, uInt64)
assert.Equal(uInt1, uInt2)
uInt1 = 1
uInt2 = 2
assert.Less(uInt1, uInt2)
assert.Greater(1, 0)
assert.GreaterOrEqual(1, 1)
assert.Less(0, 1)
assert.LessOrEqual(0, 0)
assert.Equal(0.1, 0.1)
assert.Greater(1.1, 0.1)
assert.Less(0.1, 1.1)
assert.Equal("abc", "abc")
assert.NotEqual("abc", "abd")
assert.Less("abc", "abd")
assert.Greater("abd", "abc")
assert.Equal([]int{1, 2, 3}, []int{1, 2, 3})
assert.NotEqual([]int{1, 2, 3}, []int{1, 2})
assert.IsNil(nil)
assert.IsNotNil("abc")
}

@ -4,22 +4,22 @@ import (
"encoding/json"
"errors"
"fmt"
"git.noahlan.cn/noahlan/nnet/core"
"git.noahlan.cn/noahlan/nnet/entity"
"git.noahlan.cn/noahlan/nnet"
"git.noahlan.cn/noahlan/nnet/connection"
"git.noahlan.cn/noahlan/nnet/packet"
"git.noahlan.cn/noahlan/ntools-go/core/nlog"
)
type OnReadyFunc func()
func WithNNetClientPipeline(onReady OnReadyFunc, packer packet.Packer) core.RunOption {
return func(server *core.NNet) {
server.Pipeline().Inbound().PushFront(func(entity entity.NetworkEntity, v interface{}) error {
func WithNNetClientPipeline(onReady OnReadyFunc, packer packet.Packer) nnet.RunOption {
return func(ngin *nnet.Engine) {
ngin.Pipeline().Inbound().PushFront(func(conn *connection.Connection, v interface{}) error {
pkg, ok := v.(*Packet)
if !ok {
return packet.ErrWrongPacketType
}
conn, _ := entity.Conn()
nc, _ := conn.Conn()
// Server to client
switch pkg.PacketType {
@ -32,22 +32,22 @@ func WithNNetClientPipeline(onReady OnReadyFunc, packer packet.Packer) core.RunO
PacketType: HandshakeAck,
MessageHeader: MessageHeader{},
}, nil)
if err := entity.SendBytes(hrd); err != nil {
if err := conn.SendBytes(hrd); err != nil {
return err
}
entity.SetStatus(core.StatusWorking)
conn.SetStatus(connection.StatusWorking)
// onReady
if onReady != nil {
onReady()
}
nlog.Debugf("connection handshake Id=%d, Remote=%s", entity.Session().ID(), conn.RemoteAddr())
nlog.Debugf("connection handshake Id=%d, Remote=%s", conn.Session().ID(), nc.RemoteAddr())
case Kick:
_ = entity.Close()
_ = conn.Close()
case Data:
status := entity.Status()
if status != core.StatusWorking {
status := conn.Status()
if status != connection.StatusWorking {
return errors.New(fmt.Sprintf("receive data on socket which not yet ACK, session will be closed immediately, remote=%s",
conn.RemoteAddr()))
nc.RemoteAddr()))
}
var lastMid uint64
@ -57,7 +57,7 @@ func WithNNetClientPipeline(onReady OnReadyFunc, packer packet.Packer) core.RunO
case Notify:
lastMid = 0
}
entity.SetLastMID(lastMid)
conn.SetLastMID(lastMid)
}
return nil
})

@ -1,8 +1,8 @@
package nnet
import (
"git.noahlan.cn/noahlan/nnet/core"
"git.noahlan.cn/noahlan/nnet/entity"
"git.noahlan.cn/noahlan/nnet"
"git.noahlan.cn/noahlan/nnet/connection"
"git.noahlan.cn/noahlan/nnet/middleware"
"git.noahlan.cn/noahlan/nnet/packet"
"git.noahlan.cn/noahlan/ntools-go/core/nlog"
@ -36,19 +36,19 @@ type (
}
)
func WithNNetClientProtocol(onReady OnReadyFunc) []core.RunOption {
func WithNNetClientProtocol(onReady OnReadyFunc) []nnet.RunOption {
router := NewRouter().(*nRouter)
packer := NewPacker(router.routeMap)
opts := []core.RunOption{
opts := []nnet.RunOption{
WithNNetClientPipeline(onReady, packer),
core.WithRouter(router),
core.WithPacker(func() packet.Packer { return NewPacker(router.routeMap) }),
nnet.WithRouter(router),
nnet.WithPackerBuilder(func() packet.Packer { return NewPacker(router.routeMap) }),
}
return opts
}
func WithNNetProtocol(config Config) []core.RunOption {
func WithNNetProtocol(config Config) []nnet.RunOption {
if config.HandshakeValidator == nil {
config.HandshakeValidator = func(data *HandshakeReq) error {
return nil
@ -61,17 +61,17 @@ func WithNNetProtocol(config Config) []core.RunOption {
}
packer := NewPacker(router.routeMap)
opts := []core.RunOption{
opts := []nnet.RunOption{
withNNetPipeline(handshakeAckData, config.HandshakeValidator, packer),
core.WithRouter(router),
core.WithPacker(func() packet.Packer { return NewPacker(router.routeMap) }),
nnet.WithRouter(router),
nnet.WithPackerBuilder(func() packet.Packer { return NewPacker(router.routeMap) }),
}
if config.HeartbeatInterval.Seconds() > 0 {
hbd, err := packer.Pack(Heartbeat, nil)
nlog.Must(err)
opts = append(opts, middleware.WithHeartbeat(config.HeartbeatInterval, func(_ entity.NetworkEntity) []byte {
opts = append(opts, middleware.WithHeartbeat(config.HeartbeatInterval, func(_ *connection.Connection) []byte {
return hbd
}))
}

@ -4,8 +4,8 @@ import (
"encoding/json"
"errors"
"fmt"
"git.noahlan.cn/noahlan/nnet/core"
"git.noahlan.cn/noahlan/nnet/entity"
"git.noahlan.cn/noahlan/nnet"
"git.noahlan.cn/noahlan/nnet/connection"
"git.noahlan.cn/noahlan/nnet/packet"
"git.noahlan.cn/noahlan/ntools-go/core/nlog"
)
@ -19,14 +19,14 @@ func withNNetPipeline(
handshakeResp *HandshakeResp,
validator HandshakeValidatorFunc,
packer packet.Packer,
) core.RunOption {
return func(server *core.NNet) {
server.Pipeline().Inbound().PushFront(func(entity entity.NetworkEntity, v interface{}) error {
) nnet.RunOption {
return func(ngin *nnet.Engine) {
ngin.Pipeline().Inbound().PushFront(func(conn *connection.Connection, v interface{}) error {
pkg, ok := v.(*Packet)
if !ok {
return packet.ErrWrongPacketType
}
conn, _ := entity.Conn()
nc, _ := conn.Conn()
switch pkg.PacketType {
case Handshake:
@ -46,20 +46,20 @@ func withNNetPipeline(
PacketType: Handshake,
MessageHeader: MessageHeader{},
}, data)
if err := entity.SendBytes(hrd); err != nil {
if err := conn.SendBytes(hrd); err != nil {
return err
}
entity.SetStatus(core.StatusPrepare)
nlog.Debugf("connection handshake Id=%d, Remote=%s", entity.Session().ID(), conn.RemoteAddr())
conn.SetStatus(connection.StatusPrepare)
nlog.Debugf("connection handshake Id=%d, Remote=%s", conn.Session().ID(), nc.RemoteAddr())
case HandshakeAck:
entity.SetStatus(core.StatusPending)
nlog.Debugf("receive handshake ACK Id=%d, Remote=%s", entity.Session().ID(), conn.RemoteAddr())
conn.SetStatus(connection.StatusPending)
nlog.Debugf("receive handshake ACK Id=%d, Remote=%s", conn.Session().ID(), nc.RemoteAddr())
case Data:
if entity.Status() < core.StatusPending {
if conn.Status() < connection.StatusPending {
return errors.New(fmt.Sprintf("receive data on socket which not yet ACK, session will be closed immediately, remote=%s",
conn.RemoteAddr()))
nc.RemoteAddr()))
}
entity.SetStatus(core.StatusWorking)
conn.SetStatus(connection.StatusWorking)
var lastMid uint64
switch pkg.MsgType {
@ -70,7 +70,7 @@ func withNNetPipeline(
default:
return fmt.Errorf("Invalid message type: %s ", pkg.MsgType.String())
}
entity.SetLastMID(lastMid)
conn.SetLastMID(lastMid)
}
return nil
})

@ -3,9 +3,9 @@ package nnet
import (
"errors"
"fmt"
"git.noahlan.cn/noahlan/nnet/core"
"git.noahlan.cn/noahlan/nnet/entity"
"git.noahlan.cn/noahlan/nnet/connection"
"git.noahlan.cn/noahlan/nnet/packet"
rt "git.noahlan.cn/noahlan/nnet/router"
"git.noahlan.cn/noahlan/ntools-go/core/nlog"
)
@ -23,8 +23,8 @@ type (
nRouter struct {
routeMap *RouteMap
handlers map[string]core.Handler
notFound core.Handler
handlers map[string]rt.Handler
notFound rt.Handler
}
)
@ -35,14 +35,14 @@ func NewRouteMap() *RouteMap {
}
}
func NewRouter() core.Router {
func NewRouter() rt.Router {
return &nRouter{
routeMap: NewRouteMap(),
handlers: make(map[string]core.Handler),
handlers: make(map[string]rt.Handler),
}
}
func (r *nRouter) Handle(entity entity.NetworkEntity, p packet.IPacket) {
func (r *nRouter) Handle(conn *connection.Connection, p packet.IPacket) {
pkg, ok := p.(*Packet)
if !ok {
nlog.Error(packet.ErrWrongPacketType)
@ -54,13 +54,13 @@ func (r *nRouter) Handle(entity entity.NetworkEntity, p packet.IPacket) {
nlog.Error("message handler not found")
return
}
r.notFound.Handle(entity, p)
r.notFound.Handle(conn, p)
return
}
handler.Handle(entity, p)
handler.Handle(conn, p)
}
func (r *nRouter) Register(matches interface{}, handler core.Handler) error {
func (r *nRouter) Register(matches interface{}, handler rt.Handler) error {
match, ok := matches.(Match)
if !ok {
return errors.New(fmt.Sprintf("the type of matches must be %T", Match{}))
@ -72,6 +72,6 @@ func (r *nRouter) Register(matches interface{}, handler core.Handler) error {
return nil
}
func (r *nRouter) SetNotFoundHandler(handler core.Handler) {
func (r *nRouter) SetNotFoundHandler(handler rt.Handler) {
r.notFound = handler
}

@ -1,20 +1,20 @@
package plain
import (
"git.noahlan.cn/noahlan/nnet/core"
"git.noahlan.cn/noahlan/nnet/entity"
"git.noahlan.cn/noahlan/nnet"
"git.noahlan.cn/noahlan/nnet/connection"
"git.noahlan.cn/noahlan/nnet/packet"
)
func withPipeline() core.RunOption {
return func(net *core.NNet) {
net.Pipeline().Inbound().PushFront(func(et entity.NetworkEntity, v interface{}) error {
func withPipeline() nnet.RunOption {
return func(ngin *nnet.Engine) {
ngin.Pipeline().Inbound().PushFront(func(conn *connection.Connection, v interface{}) error {
_, ok := v.(*Packet)
if !ok {
return packet.ErrWrongPacketType
}
if et.Status() != core.StatusWorking {
et.SetStatus(core.StatusWorking)
if conn.Status() != connection.StatusWorking {
conn.SetStatus(connection.StatusWorking)
}
return nil
})

@ -1,15 +1,15 @@
package plain
import (
"git.noahlan.cn/noahlan/nnet/core"
"git.noahlan.cn/noahlan/nnet"
"git.noahlan.cn/noahlan/nnet/packet"
)
func WithPlainProtocol() []core.RunOption {
opts := []core.RunOption{
func WithPlainProtocol() []nnet.RunOption {
opts := []nnet.RunOption{
withPipeline(),
core.WithRouter(NewRouter()),
core.WithPacker(func() packet.Packer { return NewPacker() }),
nnet.WithRouter(NewRouter()),
nnet.WithPackerBuilder(func() packet.Packer { return NewPacker() }),
}
return opts
}

@ -1,22 +1,22 @@
package plain
import (
"git.noahlan.cn/noahlan/nnet/core"
"git.noahlan.cn/noahlan/nnet/entity"
"git.noahlan.cn/noahlan/nnet/connection"
"git.noahlan.cn/noahlan/nnet/packet"
"git.noahlan.cn/noahlan/nnet/router"
"git.noahlan.cn/noahlan/ntools-go/core/nlog"
)
type Router struct {
plainHandler core.Handler
notFound core.Handler
plainHandler router.Handler
notFound router.Handler
}
func NewRouter() core.Router {
func NewRouter() router.Router {
return &Router{}
}
func (r *Router) Handle(entity entity.NetworkEntity, pkg packet.IPacket) {
func (r *Router) Handle(conn *connection.Connection, pkg packet.IPacket) {
p, ok := pkg.(*Packet)
if !ok {
nlog.Error(packet.ErrWrongPacketType)
@ -27,17 +27,17 @@ func (r *Router) Handle(entity entity.NetworkEntity, pkg packet.IPacket) {
nlog.Error("message handler not found")
return
}
r.notFound.Handle(entity, p)
r.notFound.Handle(conn, p)
return
}
r.plainHandler.Handle(entity, p)
r.plainHandler.Handle(conn, p)
}
func (r *Router) Register(_ interface{}, handler core.Handler) error {
func (r *Router) Register(_ interface{}, handler router.Handler) error {
r.plainHandler = handler
return nil
}
func (r *Router) SetNotFoundHandler(handler core.Handler) {
func (r *Router) SetNotFoundHandler(handler router.Handler) {
r.notFound = handler
}

@ -0,0 +1,30 @@
package router
// ToMiddleware converts the given handler to a Middleware.
func ToMiddleware(handler func(next Handler) Handler) Middleware {
return func(next HandlerFunc) HandlerFunc {
return handler(next).Handle
}
}
// WithMiddlewares adds given middlewares to given routes.
func WithMiddlewares(ms []Middleware, rs ...Route) []Route {
for i := len(ms) - 1; i >= 0; i-- {
rs = WithMiddleware(ms[i], rs...)
}
return rs
}
// WithMiddleware adds given middleware to given route.
func WithMiddleware(middleware Middleware, rs ...Route) []Route {
routes := make([]Route, len(rs))
for i := range rs {
route := rs[i]
routes[i] = Route{
Matches: route.Matches,
Handler: middleware(route.Handler),
}
}
return routes
}

@ -1,11 +1,20 @@
package core
package router
import (
"git.noahlan.cn/noahlan/nnet/entity"
"git.noahlan.cn/noahlan/nnet/connection"
"git.noahlan.cn/noahlan/nnet/packet"
"git.noahlan.cn/noahlan/ntools-go/core/nlog"
)
type (
Handler interface {
Handle(c *connection.Connection, pkg packet.IPacket)
}
// HandlerFunc 消息处理方法
HandlerFunc func(conn *connection.Connection, pkg packet.IPacket)
Middleware func(next HandlerFunc) HandlerFunc
Route struct {
Matches interface{} // 用于匹配的关键字段
Handler HandlerFunc // 处理方法
@ -20,11 +29,31 @@ type (
Constructor func(Handler) Handler
)
func notFound(conn *connection.Connection, _ packet.IPacket) {
nlog.Error("handler not found")
_ = conn.SendBytes([]byte("404"))
}
func NotFoundHandler(next Handler) Handler {
return HandlerFunc(func(c *connection.Connection, packet packet.IPacket) {
h := next
if next == nil {
h = HandlerFunc(notFound)
}
// TODO write to client
h.Handle(c, packet)
})
}
func (f HandlerFunc) Handle(c *connection.Connection, pkg packet.IPacket) {
f(c, pkg)
}
type Chain struct {
constructors []Constructor
}
func newChain(constructors ...Constructor) Chain {
func NewChain(constructors ...Constructor) Chain {
return Chain{append(([]Constructor)(nil), constructors...)}
}
@ -65,11 +94,11 @@ func NewDefaultRouter() Router {
return &plainRouter{}
}
func (p *plainRouter) Handle(entity entity.NetworkEntity, pkg packet.IPacket) {
func (p *plainRouter) Handle(c *connection.Connection, pkg packet.IPacket) {
if p.handler == nil {
return
}
p.handler.Handle(entity, pkg)
p.handler.Handle(c, pkg)
}
func (p *plainRouter) Register(_ interface{}, handler Handler) error {

@ -0,0 +1,7 @@
package router
func ConvertMiddleware(ware Middleware) func(Handler) Handler {
return func(next Handler) Handler {
return ware(next.Handle)
}
}

@ -0,0 +1,46 @@
package nnet
import (
"errors"
"git.noahlan.cn/noahlan/nnet/config"
"git.noahlan.cn/noahlan/ntools-go/core/nlog"
"net"
)
func (ngin *Engine) ListenTCP(conf config.TCPServerConf) error {
err := ngin.setup()
if err != nil {
nlog.Errorf("%s failed to setup server, err:%v", ngin.LogPrefix(), err)
return err
}
listener, err := net.Listen(conf.Protocol, conf.Addr)
if err != nil {
nlog.Errorf("%s failed to listening at [%s %s] %v", ngin.LogPrefix(), conf.Protocol, conf.Addr, err)
return err
}
nlog.Infof("%s now listening %s at %s...", ngin.LogPrefix(), conf.Protocol, conf.Addr)
defer func() {
_ = listener.Close()
ngin.Stop()
}()
for {
conn, err := listener.Accept()
if err != nil {
if errors.Is(err, net.ErrClosed) {
nlog.Errorf("%s connection closed, err:%v", ngin.LogPrefix(), err)
return err
}
nlog.Errorf("%s accept connection failed, err:%v", ngin.LogPrefix(), err)
continue
}
err = ngin.goPool.Submit(func() {
ngin.handle(conn)
})
if err != nil {
nlog.Errorf("%s submit conn pool err: %ng", ngin.LogPrefix(), err.Error())
continue
}
}
}

@ -0,0 +1,66 @@
package nnet
import (
"fmt"
"git.noahlan.cn/noahlan/nnet/config"
"git.noahlan.cn/noahlan/nnet/connection"
"git.noahlan.cn/noahlan/ntools-go/core/nlog"
"github.com/gorilla/websocket"
"net/http"
"os"
"strings"
)
// ListenWebsocket 开始监听Websocket
func (ngin *Engine) ListenWebsocket(conf config.WSServerConf) error {
err := ngin.setup()
if err != nil {
nlog.Errorf("%s failed to setup server, err:%v", ngin.LogPrefix(), err)
return err
}
nlog.Infof("%s now listening websocket at %s.", ngin.LogPrefix(), conf.Addr)
ngin.upgradeWebsocket(conf)
if conf.IsTLS() {
if err := http.ListenAndServeTLS(conf.Addr, conf.TLSCertificate, conf.TLSKey, nil); err != nil {
nlog.Errorf("%s failed to listening websocket with TLS at %s %v", ngin.LogPrefix(), conf.Addr, err)
return err
}
} else {
if err := http.ListenAndServe(conf.Addr, nil); err != nil {
nlog.Errorf("%s failed to listening websocket at %s %v", ngin.LogPrefix(), conf.Addr, err)
return err
}
}
return nil
}
func (ngin *Engine) handleWS(conn *websocket.Conn) {
wsConn := connection.NewWSConn(conn)
ngin.handle(wsConn)
}
func (ngin *Engine) upgradeWebsocket(conf config.WSServerConf) {
upgrade := websocket.Upgrader{
HandshakeTimeout: conf.HandshakeTimeout,
ReadBufferSize: conf.ReadBufferSize,
WriteBufferSize: conf.WriteBufferSize,
CheckOrigin: conf.CheckOrigin,
EnableCompression: conf.Compression,
}
path := fmt.Sprintf("/%s", strings.TrimPrefix(conf.Path, "/"))
http.HandleFunc(path, func(writer http.ResponseWriter, request *http.Request) {
conn, err := upgrade.Upgrade(writer, request, nil)
if err != nil {
nlog.Errorf("%s Upgrade failure, URI=%ng, Error=%ng", ngin.LogPrefix(), request.RequestURI, err.Error())
return
}
err = ngin.goPool.Submit(func() {
ngin.handleWS(conn)
})
if err != nil {
nlog.Errorf("%s submit conn pool err: %v", ngin.LogPrefix(), err.Error())
os.Exit(1)
}
})
}

@ -0,0 +1,104 @@
package session
import (
"sync"
)
type Session struct {
sync.RWMutex // 数据锁
id int64 // Session全局唯一ID
uid string // 用户ID不绑定的情况下与sid一致
data map[string]interface{} // session数据存储内存
}
func NewSession(id int64) *Session {
return &Session{
id: id,
uid: "",
data: make(map[string]interface{}),
}
}
// ID 获取 session ID
func (s *Session) ID() int64 {
return s.id
}
// UID 获取UID
func (s *Session) UID() string {
return s.uid
}
// Bind 绑定uid
func (s *Session) Bind(uid string) {
s.uid = uid
}
// Attribute 获取指定key对应参数
func (s *Session) Attribute(key string) interface{} {
s.RLock()
defer s.RUnlock()
return s.data[key]
}
// Keys 获取所有参数key
func (s *Session) Keys() []string {
s.RLock()
defer s.RUnlock()
keys := make([]string, 0, len(s.data))
for k := range s.data {
keys = append(keys, k)
}
return keys
}
// Exists 指定key是否存在
func (s *Session) Exists(key string) bool {
s.RLock()
defer s.RUnlock()
_, has := s.data[key]
return has
}
// Attributes 获取所有参数
func (s *Session) Attributes() map[string]interface{} {
s.RLock()
defer s.RUnlock()
return s.data
}
// RemoveAttribute 移除指定key对应参数
func (s *Session) RemoveAttribute(key string) {
s.Lock()
defer s.Unlock()
delete(s.data, key)
}
// SetAttribute 设置参数
func (s *Session) SetAttribute(key string, value interface{}) {
s.Lock()
defer s.Unlock()
s.data[key] = value
}
// Invalidate 清理
func (s *Session) Invalidate() {
s.Lock()
defer s.Unlock()
s.id = 0
s.uid = ""
s.data = make(map[string]interface{})
}
// Close 关闭
func (s *Session) Close() {
s.Invalidate()
}

@ -0,0 +1,38 @@
package session
import "sync/atomic"
type IDMgr struct {
count int64
sid int64
}
func NewSessionIDMgr() *IDMgr {
return &IDMgr{}
}
// Increment the connection count
func (c *IDMgr) Increment() {
atomic.AddInt64(&c.count, 1)
}
// Decrement the connection count
func (c *IDMgr) Decrement() {
atomic.AddInt64(&c.count, -1)
}
// Count returns the connection numbers in current
func (c *IDMgr) Count() int64 {
return atomic.LoadInt64(&c.count)
}
// Reset the connection service status
func (c *IDMgr) Reset() {
atomic.StoreInt64(&c.count, 0)
atomic.StoreInt64(&c.sid, 0)
}
// SessionID returns the session id
func (c *IDMgr) SessionID() int64 {
return atomic.AddInt64(&c.sid, 1)
}

@ -2,11 +2,12 @@ package main
import (
"encoding/json"
"git.noahlan.cn/noahlan/nnet"
"git.noahlan.cn/noahlan/nnet/config"
"git.noahlan.cn/noahlan/nnet/core"
"git.noahlan.cn/noahlan/nnet/entity"
"git.noahlan.cn/noahlan/nnet/connection"
"git.noahlan.cn/noahlan/nnet/packet"
"git.noahlan.cn/noahlan/nnet/protocol/nnet"
protocol_nnet "git.noahlan.cn/noahlan/nnet/protocol/nnet"
rt "git.noahlan.cn/noahlan/nnet/router"
"git.noahlan.cn/noahlan/ntools-go/core/nlog"
"git.noahlan.cn/noahlan/ntools-go/core/pool"
"math"
@ -14,80 +15,89 @@ import (
)
func runServer(addr string) {
server := core.NewServer(config.EngineConf{
ServerConf: config.ServerConf{
Protocol: "tcp",
Addr: addr,
Name: "testServer",
Mode: "dev",
},
Pool: pool.Config{
PoolSize: math.MaxInt32,
ExpiryDuration: time.Second,
PreAlloc: false,
MaxBlockingTasks: 0,
Nonblocking: false,
DisablePurge: false,
},
}, nnet.WithNNetProtocol(nnet.Config{
nginOpts := make([]nnet.RunOption, 0)
nginOpts = append(nginOpts, nnet.WithPoolCfg(pool.Config{
PoolSize: math.MaxInt32,
ExpiryDuration: time.Second,
PreAlloc: false,
MaxBlockingTasks: 0,
Nonblocking: false,
DisablePurge: false,
}))
nginOpts = append(nginOpts, protocol_nnet.WithNNetProtocol(protocol_nnet.Config{
HeartbeatInterval: 0,
HandshakeValidator: nil,
})...)
server.AddRoutes([]core.Route{
{
Matches: nnet.Match{
Route: "ping",
Code: 1,
},
Handler: func(et entity.NetworkEntity, pkg packet.IPacket) {
nlog.Info("client ping, server pong -> ")
err := et.Send(nnet.Header{
PacketType: nnet.Data,
MessageHeader: nnet.MessageHeader{
MsgType: nnet.Request,
ID: 1,
Route: "pong",
},
}, []byte("1"))
nlog.Must(err)
},
ngin := nnet.NewEngine(config.EngineConf{
TaskTimerPrecision: 0,
Mode: "dev",
Name: "NL",
}, nginOpts...)
ngin.AddRoutes(rt.Route{
Matches: protocol_nnet.Match{
Route: "ping",
Code: 1,
},
Handler: func(conn *connection.Connection, pkg packet.IPacket) {
nlog.Info("client ping, server pong -> ")
err := conn.Send(protocol_nnet.Header{
PacketType: protocol_nnet.Data,
MessageHeader: protocol_nnet.MessageHeader{
MsgType: protocol_nnet.Request,
ID: 1,
Route: "pong",
},
}, []byte("1"))
nlog.Must(err)
},
})
defer server.Stop()
server.Start()
defer ngin.Stop()
err := ngin.ListenTCP(config.TCPServerConf{
Protocol: "tcp",
Addr: addr,
})
if err != nil {
return
}
}
func runClient(addr string) (client *core.Client, et entity.NetworkEntity) {
func runClient(addr string) (*nnet.Engine, *connection.Connection) {
chReady := make(chan struct{})
client = core.NewClient(config.EngineConf{
Pool: pool.Config{
PoolSize: math.MaxInt32,
ExpiryDuration: time.Second,
PreAlloc: false,
MaxBlockingTasks: 0,
Nonblocking: false,
DisablePurge: false,
},
}, nnet.WithNNetClientProtocol(func() {
nginOpts := make([]nnet.RunOption, 0)
nginOpts = append(nginOpts, nnet.WithPoolCfg(pool.Config{
PoolSize: math.MaxInt32,
ExpiryDuration: time.Second,
PreAlloc: false,
MaxBlockingTasks: 0,
Nonblocking: false,
DisablePurge: false,
}))
nginOpts = append(nginOpts, protocol_nnet.WithNNetClientProtocol(func() {
chReady <- struct{}{}
})...)
client.AddRoutes([]core.Route{
{
Matches: nnet.Match{
Route: "test.client",
Code: 1,
},
Handler: func(et entity.NetworkEntity, pkg packet.IPacket) {
nlog.Info("client hahaha")
},
ngin := nnet.NewEngine(config.EngineConf{
TaskTimerPrecision: 0,
Mode: "dev",
Name: "NL",
}, nginOpts...)
ngin.AddRoutes(rt.Route{
Matches: protocol_nnet.Match{
Route: "test.client",
Code: 1,
},
Handler: func(conn *connection.Connection, pkg packet.IPacket) {
nlog.Info("client hahaha")
},
})
et = client.Dial(addr)
conn, err := ngin.Dial(addr)
nlog.Must(err)
handshake, err := json.Marshal(&nnet.HandshakeReq{
handshake, err := json.Marshal(&protocol_nnet.HandshakeReq{
Version: "1.0.0",
Type: "test",
ClientId: "a",
@ -98,10 +108,10 @@ func runClient(addr string) (client *core.Client, et entity.NetworkEntity) {
})
nlog.Must(err)
packer := nnet.NewPacker(nnet.NewRouteMap())
hsd, err := packer.Pack(nnet.Header{
PacketType: nnet.Handshake,
MessageHeader: nnet.MessageHeader{
packer := protocol_nnet.NewPacker(protocol_nnet.NewRouteMap())
hsd, err := packer.Pack(protocol_nnet.Header{
PacketType: protocol_nnet.Handshake,
MessageHeader: protocol_nnet.MessageHeader{
MsgType: 0,
ID: 0,
Route: "",
@ -109,9 +119,9 @@ func runClient(addr string) (client *core.Client, et entity.NetworkEntity) {
}, handshake)
nlog.Must(err)
err = et.SendBytes(hsd)
err = conn.SendBytes(hsd)
nlog.Must(err)
<-chReady
return
return ngin, conn
}

@ -1,10 +1,10 @@
package main
import (
"git.noahlan.cn/noahlan/nnet/core"
"git.noahlan.cn/noahlan/nnet/entity"
"git.noahlan.cn/noahlan/nnet/connection"
"git.noahlan.cn/noahlan/nnet/packet"
"git.noahlan.cn/noahlan/nnet/protocol/nnet"
rt "git.noahlan.cn/noahlan/nnet/router"
"git.noahlan.cn/noahlan/ntools-go/core/nlog"
"sync"
"testing"
@ -15,13 +15,13 @@ func TestServer(t *testing.T) {
}
func TestClient(t *testing.T) {
client, et := runClient("127.0.0.1:6666")
client.AddRoute(core.Route{
ngin, et := runClient("127.0.0.1:6666")
ngin.AddRoutes(rt.Route{
Matches: nnet.Match{
Route: "pong",
Code: 2,
},
Handler: func(et entity.NetworkEntity, pkg packet.IPacket) {
Handler: func(conn *connection.Connection, pkg packet.IPacket) {
nlog.Info("server pong, client ping ->")
_ = et.Send(nnet.Header{
PacketType: nnet.Data,

Loading…
Cancel
Save