You cannot select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
nnet/conn/conn.go

148 lines
2.6 KiB
Go

package conn
import (
"git.noahlan.cn/noahlan/nnet/session"
"net"
"sync/atomic"
)
type (
Connection struct {
session *session.Session // Session
status int32 // 连接状态
conn net.Conn // low-level conn fd
typ ConnType // 连接类型
lastMid uint64 // 最近一次消息ID
chDie chan struct{} // 停止通道
chSend chan PendingMessage // 消息发送通道(结构化消息)
chWrite chan []byte // 消息发送通道(二进制消息)
}
)
func NewConnection(id int64, rawC net.Conn) *Connection {
r := &Connection{
session: session.NewSession(id),
status: StatusStart,
conn: rawC,
typ: ConnTypeTCP,
lastMid: 0,
chDie: make(chan struct{}),
chSend: make(chan PendingMessage, 128),
chWrite: make(chan []byte, 128),
}
if _, ok := rawC.(*WSConn); ok {
r.typ = ConnTypeWS
return r
}
if _, ok := rawC.(*SerialConn); ok {
r.typ = ConnTypeSerial
return r
}
return r
}
func (r *Connection) Send(header, payload any) (err error) {
defer func() {
if e := recover(); e != nil {
err = ErrBrokenPipe
}
}()
r.chSend <- PendingMessage{
Header: header,
Payload: payload,
}
return err
}
func (r *Connection) SendBytes(data []byte) (err error) {
defer func() {
if e := recover(); e != nil {
err = ErrBrokenPipe
}
}()
r.chWrite <- data
return err
}
func (r *Connection) Status() int32 {
return atomic.LoadInt32(&r.status)
}
func (r *Connection) SetStatus(s int32) {
atomic.StoreInt32(&r.status, s)
}
func (r *Connection) Type() ConnType {
return r.typ
}
func (r *Connection) Conn() net.Conn {
return r.conn
}
func (r *Connection) WsConn() *WSConn {
if r.typ == ConnTypeWS {
return r.conn.(*WSConn)
}
return nil
}
func (r *Connection) SerialConn() *SerialConn {
if r.typ == ConnTypeSerial {
return r.conn.(*SerialConn)
}
return nil
}
func (r *Connection) ID() int64 {
return r.session.ID()
}
func (r *Connection) Session() *session.Session {
return r.session
}
func (r *Connection) LastMID() uint64 {
return r.lastMid
}
func (r *Connection) SetLastMID(mid uint64) {
atomic.StoreUint64(&r.lastMid, mid)
}
func (r *Connection) ChDie() chan struct{} {
return r.chDie
}
func (r *Connection) ChSend() chan PendingMessage {
return r.chSend
}
func (r *Connection) ChWrite() chan []byte {
return r.chWrite
}
func (r *Connection) Close() error {
if r.Status() == StatusClosed {
return ErrCloseClosedSession
}
r.SetStatus(StatusClosed)
select {
case <-r.chDie:
close(r.chSend)
close(r.chWrite)
default:
close(r.chDie)
}
r.session.Close()
return r.conn.Close()
}