gnet

package module
v0.0.0-...-f657754 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Feb 9, 2025 License: MIT Imports: 23 Imported by: 0

README

gnet

Go Report Card Go Reference codecov Mentioned in Awesome Go

中文说明

High performance network library,especially for game servers

Features

  • MultiThread, nonblocking
  • Default support protobuf
  • Optimize receiving and dispatching using lockless RingBuffer, which can improve performance by 5x for some cases
  • Easy to implement custom encoding and decoding
  • rpc
  • Support Tcp,WebSocket(ws and wss)

Usage

run a server

codec := gnet.NewProtoCodec(nil)
handler := gnet.NewDefaultConnectionHandler(codec)
handler.Register(gnet.PacketCommand(pb.CmdTest_Cmd_TestMessage), onTestMessage, new(pb.TestMessage))
listenerConfig := &gnet.ListenerConfig{
    AcceptConfig: gnet.DefaultConnectionConfig,
}
listenerConfig.AcceptConfig.Codec = codec
listenerConfig.AcceptConfig.Handler = handler
gnet.GetNetMgr().NewListener(ctx, "localhost:10001", listenerConfig)

run a client

codec := gnet.NewProtoCodec(nil)
handler := gnet.NewDefaultConnectionHandler(codec)
handler.Register(gnet.PacketCommand(pb.CmdTest_Cmd_TestMessage), onTestMessage, new(pb.TestMessage))
connectionConfig := gnet.DefaultConnectionConfig
connectionConfig.Codec = clientCodec
connectionConfig.Handler = clientHandler
connector := gnet.GetNetMgr().NewConnector(ctx, "localhost:10001", &connectionConfig, nil)
connector.SendPacket(gnet.NewProtoPacket(gnet.PacketCommand(pb.CmdTest_Cmd_TestMessage),
    &pb.TestMessage{
        Name: "hello",
    }))

Encoding and decoding(https://github.com/shinysimple/gnet/blob/main/codec.go)

gnet divide TCP stream based decoding into three layers

Layer1:subcontracting stream, format:|Length|Data|,after receiving full packet content, hand it over to the next layer for processing

Layer2:Decoding the data from Layer1,such as decryption,decompression,etc

Layer3:protobuf deserialize,generate proto.Message

encode

decode

Use RingBuffer to increase performance

ringbuffer-performance

rpc

Rpc send a request to target and block wait reply,similar to grpc-go,but gnet use command id instead of method name

request := gnet.NewProtoPacket(cmd, &pb.HelloRequest{
    Name: "hello",
})
reply := new(pb.HelloReply)
err := connection.Rpc(request, reply)
if err != nil {
    return
}
logger.Info("reply:%v", reply)

goroutine

connection_goroutine

Examples

example/helloworld: a simple example use protobuf message

example/data_packet: a simple example use DataPacket

example/custom_packet: how to extend custom packet struct

example/tcp_connection_simple: use protobuf message without RingBuffer

example/packet_size: send big packet which size is bigger than RingBuffer's cap

example/websocket: a simple example use websocket

example/rpc: how to use rpc

example/simulate_game: a performance test with a game application scenario

Client Connector Library

C#: gnet_csharp

Project

game db&cache framework

distributed game server framework

gnet is also used in our commercial online game projects

Documentation

Index

Constants

View Source 
const (
	DebugLevel int8 = iota - 1
	InfoLevel
	WarnLevel
	ErrorLevel
)

日志级别,参考zap 

log level
View Source 
const (
	// 默认包头长度
	// the default packet header size
	DefaultPacketHeaderSize = int(unsafe.Sizeof(DefaultPacketHeader{}))

	// 数据包长度限制(16M)
	// the default packet data size limit
	MaxPacketDataSize = 0x00FFFFFF

	// packet header contains rpcCallId
	RpcCall uint8 = 1 << 0
	// compress packet data
	Compress uint8 = 1 << 1
	// packet contains error code
	ErrorCode uint8 = 1 << 2
)
View Source 
const (
	SimplePacketHeaderSize = int(unsafe.Sizeof(SimplePacketHeader{}))
)

Variables

View Source 
var (
	ErrBufferFull   = errors.New("buffer is full")
	ErrNotSupport   = errors.New("not support")
	ErrPacketLength = errors.New("packet length error")
	// 数据包长度超出设置
	ErrPacketLengthExceed = errors.New("packet length exceed")
	ErrReadRemainPacket   = errors.New("read remain packet data error")
	ErrPacketDataNotRead  = errors.New("packet data not read")
)
View Source 
var (
	DefaultConnectionConfig = ConnectionConfig{
		SendPacketCacheCap: 16,
		SendBufferSize:     4096,
		RecvBufferSize:     4096,
		MaxPacketSize:      MaxPacketDataSize,
		RecvTimeout:        7,
		HeartBeatInterval:  3,
	}
)
View Source 
var (
	// rpc默认超时时间
	DefaultRpcTimeout = time.Second * 3
)

Functions

func GetCurrentTimeStamp 

func GetCurrentTimeStamp() int64

获取当前时间戳(秒) 

get current timestamp

func LogStack 

func LogStack()

func NewConnectionId 

func NewConnectionId() uint32

func SetLogLevel 

func SetLogLevel(level int8)

func SetLogger 

func SetLogger(w Logger, level int8)

Types

type AcceptConnectionCreator 

type AcceptConnectionCreator func(conn net.Conn, config *ConnectionConfig) Connection

type Codec 

type Codec interface {
	// 包头长度
	// 应用层可以自己扩展包头长度
	// the packet header size (without packet data)
	PacketHeaderSize() uint32

	// 创建消息头
	// generate a packet header
	// packet can be nil
	// packetData is the data after encode,can be nil
	CreatePacketHeader(connection Connection, packet Packet, packetData []byte) PacketHeader

	// encoding
	Encode(connection Connection, packet Packet) []byte

	// decoding
	Decode(connection Connection, data []byte) (newPacket Packet, err error)
}

连接的编解码接口 interface for encoding and decoding

type Connection 

type Connection interface {
	// unique id
	GetConnectionId() uint32

	// is connector
	IsConnector() bool

	// send a packet(proto.Message)
	//  NOTE: 调用Send(command,message)之后,不要再对message进行读写!
	//  NOTE: do not read or modify message after call Send
	Send(command PacketCommand, message proto.Message, opts ...SendOption) bool

	// send a packet(Packet)
	//  NOTE:调用SendPacket(packet)之后,不要再对packet进行读写!
	//  NOTE: do not read or modify Packet after call SendPacket
	SendPacket(packet Packet, opts ...SendOption) bool

	// 超时发包,超时未发送则丢弃,适用于某些允许丢弃的数据包
	//  try send a packet with Timeout
	TrySendPacket(packet Packet, timeout time.Duration, opts ...SendOption) bool

	// Rpc send a request to target and block wait reply
	Rpc(request Packet, reply proto.Message, opts ...SendOption) error

	// is connected
	IsConnected() bool

	// codec for this connection
	GetCodec() Codec

	// set codec
	SetCodec(codec Codec)

	// handler for this connection
	GetHandler() ConnectionHandler

	// LocalAddr returns the local network address.
	LocalAddr() net.Addr

	// RemoteAddr returns the remote network address.
	RemoteAddr() net.Addr

	// close this connection
	Close()

	// 获取关联数据
	// get the associated tag
	GetTag() interface{}

	// 设置关联数据
	// set the associated tag
	SetTag(tag interface{})

	// connect to target server
	//  address format ip:port
	Connect(address string) bool

	// 开启读写协程
	// start the read&write goroutine
	Start(ctx context.Context, netMgrWg *sync.WaitGroup, onClose func(connection Connection))
}

interface for Connection

type ConnectionConfig 

type ConnectionConfig struct {
	// 发包缓存chan大小(缓存数据包chan容量)
	// capacity for send packet chan
	SendPacketCacheCap uint32

	// 发包Buffer大小(byte)
	// size of send RingBuffer (byte)
	SendBufferSize uint32

	// 收包Buffer大小(byte)
	// size of recv RingBuffer (byte)
	RecvBufferSize uint32

	// 最大包体大小设置(byte),不包含PacketHeader
	// 允许该值大于SendBufferSize和RecvBufferSize
	//  max size of packet (byte), not include PacketHeader's size
	//  allow MaxPacketSize lager than SendBufferSize and RecvBufferSize
	MaxPacketSize uint32

	// 收包超时设置(秒)
	//  if the connection dont recv packet for RecvTimeout seconds,the connection will close
	//  if RecvTimeout is zero,it will not check Timeout
	RecvTimeout uint32

	// 心跳包发送间隔(秒),对connector有效
	//  heartbeat packet sending interval(seconds)
	//  only valid for connector
	HeartBeatInterval uint32

	// 发包超时设置(秒)
	//  net.Conn.SetWriteDeadline
	WriteTimeout uint32

	Codec Codec

	Handler ConnectionHandler

	// ws或wss的http路径,如"/ws"或"/wss"
	Path string

	// "ws"或"wss"
	Scheme string
}

connection options

type ConnectionCreator 

type ConnectionCreator func(config *ConnectionConfig) Connection

type ConnectionHandler 

type ConnectionHandler interface {
	// 连接成功或失败
	//  after connect
	OnConnected(connection Connection, success bool)

	// 断开连接
	//  when disconnected
	OnDisconnected(connection Connection)

	// 收到一个完整数据包
	// 在收包协程中调用
	//  after recv a full packet, calling in the read goroutine
	OnRecvPacket(connection Connection, packet Packet)

	// 创建一个心跳包(只对connector有效)
	// 在connector的发包协程中调用
	//  generate a heartbeat packet, calling int the connector's write goroutine
	CreateHeartBeatPacket(connection Connection) Packet
}

handler for Connection

type DataPacket 

type DataPacket struct {
	// contains filtered or unexported fields
}

只包含一个[]byte的数据包 

packet which only have a byte array

func NewDataPacket 

func NewDataPacket(data []byte) *DataPacket

func NewDataPacketWithHeader 

func NewDataPacketWithHeader(header PacketHeader, data []byte) *DataPacket

func (*DataPacket) Clone 

func (p *DataPacket) Clone() Packet

deep copy

func (*DataPacket) Command 

func (p *DataPacket) Command() PacketCommand

func (*DataPacket) ErrorCode 

func (p *DataPacket) ErrorCode() uint32

NOTE: 暂未实现ErrorCode

func (*DataPacket) GetStreamData 

func (p *DataPacket) GetStreamData() []byte

func (*DataPacket) Message 

func (p *DataPacket) Message() proto.Message

func (*DataPacket) RpcCallId 

func (p *DataPacket) RpcCallId() uint32

NOTE: 暂未实现rpcCallId

func (*DataPacket) SetErrorCode 

func (p *DataPacket) SetErrorCode(code uint32) *DataPacket

NOTE: 暂未实现ErrorCode

func (*DataPacket) SetRpcCallId 

func (p *DataPacket) SetRpcCallId(rpcCallId uint32)

NOTE: 暂未实现rpcCallId

type DefaultCodec 

type DefaultCodec struct {
	RingBufferCodec
}

默认编解码,只做长度和数据的解析 default codec, which format is length + data

func NewDefaultCodec 

func NewDefaultCodec() *DefaultCodec

type DefaultConnectionHandler 

type DefaultConnectionHandler struct {
	// 注册消息的处理函数map
	//  registered map of PacketCommand and PacketHandler
	PacketHandlers map[PacketCommand]PacketHandler
	// 未注册消息的处理函数
	//  packetHandler for unregistered PacketCommand
	UnRegisterHandler PacketHandler
	// contains filtered or unexported fields
}

default ConnectionHandler for Proto

func NewDefaultConnectionHandler 

func NewDefaultConnectionHandler(protoCodec Codec) *DefaultConnectionHandler

func (*DefaultConnectionHandler) CreateHeartBeatPacket 

func (h *DefaultConnectionHandler) CreateHeartBeatPacket(connection Connection) Packet

func (*DefaultConnectionHandler) GetCodec 

func (h *DefaultConnectionHandler) GetCodec() Codec

func (*DefaultConnectionHandler) GetPacketHandler 

func (h *DefaultConnectionHandler) GetPacketHandler(packetCommand PacketCommand) PacketHandler

func (*DefaultConnectionHandler) OnConnected 

func (h *DefaultConnectionHandler) OnConnected(connection Connection, success bool)

func (*DefaultConnectionHandler) OnDisconnected 

func (h *DefaultConnectionHandler) OnDisconnected(connection Connection)

func (*DefaultConnectionHandler) OnRecvPacket 

func (h *DefaultConnectionHandler) OnRecvPacket(connection Connection, packet Packet)

func (*DefaultConnectionHandler) Register 

func (h *DefaultConnectionHandler) Register(packetCommand PacketCommand, handler PacketHandler, protoMessage proto.Message)

注册消息号和消息回调,proto.Message的映射 handler在TcpConnection的read协程中被调用 

register PacketCommand,PacketHandler,proto.Message

func (*DefaultConnectionHandler) RegisterHeartBeat 

func (h *DefaultConnectionHandler) RegisterHeartBeat(heartBeatPacketCreator PacketCreator)

注册心跳包(只对connector有效) 

register heartBeatPacketCreator, only valid for connector

func (*DefaultConnectionHandler) SetOnConnectedFunc 

func (h *DefaultConnectionHandler) SetOnConnectedFunc(onConnectedFunc func(connection Connection, success bool))

set connected callback

func (*DefaultConnectionHandler) SetOnDisconnectedFunc 

func (h *DefaultConnectionHandler) SetOnDisconnectedFunc(onDisconnectedFunc func(connection Connection))

set disconnected callback

func (*DefaultConnectionHandler) SetUnRegisterHandler 

func (h *DefaultConnectionHandler) SetUnRegisterHandler(unRegisterHandler PacketHandler)

未注册消息的处理函数 unRegisterHandler在TcpConnection的read协程中被调用 

register the PacketHandler for unRegister PacketCommand

type DefaultPacketHeader 

type DefaultPacketHeader struct {
	// (flags << 24) | len
	// flags [0,255)
	// len [0,16M)
	LenAndFlags uint32
}

默认包头,支持小于16M的数据包 

default packet header

func NewDefaultPacketHeader 

func NewDefaultPacketHeader(len uint32, flags uint8) *DefaultPacketHeader

func (*DefaultPacketHeader) AddFlags 

func (h *DefaultPacketHeader) AddFlags(flag uint8)

func (*DefaultPacketHeader) Flags 

func (h *DefaultPacketHeader) Flags() uint8

标记 [0,0xFF]

func (*DefaultPacketHeader) HasFlag 

func (h *DefaultPacketHeader) HasFlag(flag uint8) bool

func (*DefaultPacketHeader) Len 

func (h *DefaultPacketHeader) Len() uint32

包体长度,不包含包头的长度 

packet body length (without packet header's length)
[0,0x00FFFFFF]

func (*DefaultPacketHeader) ReadFrom 

func (h *DefaultPacketHeader) ReadFrom(packetHeaderData []byte)

从字节流读取数据,len(messageHeaderData)>=MessageHeaderSize 使用小端字节序 

parse LenAndFlags from stream data

func (*DefaultPacketHeader) SetFlags 

func (h *DefaultPacketHeader) SetFlags(flags uint8)

func (*DefaultPacketHeader) WriteTo 

func (h *DefaultPacketHeader) WriteTo(packetHeaderData []byte)

写入字节流,使用小端字节序 

write LenAndFlags to stream data

type Listener 

type Listener interface {
	GetListenerId() uint32

	GetConnection(connectionId uint32) Connection

	// 广播消息
	//  broadcast packet to accepted connections
	Broadcast(packet Packet)

	// Addr returns the listener's network address.
	Addr() net.Addr

	Close()
}

interface for Listener

type ListenerConfig 

type ListenerConfig struct {
	AcceptConfig            ConnectionConfig
	AcceptConnectionCreator AcceptConnectionCreator
	ListenerHandler         ListenerHandler
	// ws或wss的http监听路径,如"/ws"或"/wss"
	Path string
	// 签名cert文件,wss专用
	CertFile string
	// 签名key文件,wss专用
	KeyFile string
}

type ListenerHandler 

type ListenerHandler interface {
	// accept a new connection
	OnConnectionConnected(listener Listener, acceptedConnection Connection)

	// a connection disconnect
	OnConnectionDisconnect(listener Listener, connection Connection)
}

handler for Listener

type Logger 

type Logger interface {
	Debug(format string, args ...interface{})
	Info(format string, args ...interface{})
	Warn(format string, args ...interface{})
	Error(format string, args ...interface{})
}

func GetLogger 

func GetLogger() Logger

func NewStdLogger 

func NewStdLogger(callDepth int) Logger

type NetMgr 

type NetMgr struct {
	// contains filtered or unexported fields
}

网络管理类,提供对外接口 

manager class

func GetNetMgr 

func GetNetMgr() *NetMgr

单例模式,在调用的时候才会执行初始化一次 

singleton mode, init once

func (*NetMgr) NewConnector 

func (m *NetMgr) NewConnector(ctx context.Context, address string, connectionConfig *ConnectionConfig,
	tag interface{}) Connection

create a new TcpConnection

func (*NetMgr) NewConnectorCustom 

func (m *NetMgr) NewConnectorCustom(ctx context.Context, address string, connectionConfig *ConnectionConfig,
	tag interface{}, connectionCreator ConnectionCreator) Connection

create a new Connection, with custom connectionCreator

func (*NetMgr) NewListener 

func (m *NetMgr) NewListener(ctx context.Context, address string, listenerConfig *ListenerConfig) Listener

func (*NetMgr) NewWsConnector 

func (m *NetMgr) NewWsConnector(ctx context.Context, address string, connectionConfig *ConnectionConfig,
	tag interface{}) Connection

func (*NetMgr) NewWsListener 

func (m *NetMgr) NewWsListener(ctx context.Context, address string, listenerConfig *ListenerConfig) Listener

func (*NetMgr) Shutdown 

func (m *NetMgr) Shutdown(waitForAllNetGoroutine bool)

waitForAllNetGoroutine:是否阻塞等待所有网络协程结束 

waitForAllNetGoroutine: wait blocks until all goroutine end

type Packet 

type Packet interface {
	// 消息号
	// 没有把消息号放在PacketHeader里,因为对TCP网络层来说,只需要知道每个数据包的分割长度就可以了,
	// 至于数据包具体的格式,不该是网络层关心的事情
	// 消息号也不是必须放在这里的,但是游戏项目一般都是用消息号,为了减少封装层次,就放这里了
	//  packet command number
	Command() PacketCommand

	RpcCallId() uint32

	SetRpcCallId(rpcCallId uint32)

	ErrorCode() uint32

	// default protobuf
	Message() proto.Message

	// 提供一个二进制数据的接口,支持外部直接传入序列化的字节流数据
	//  support stream data, outside can direct pass the serialized data
	GetStreamData() []byte

	// deep copy
	Clone() Packet
}

interface for packet

type PacketCommand 

type PacketCommand uint16

消息号[0,65535]

type PacketCreator 

type PacketCreator func() Packet

Packet ctor func

type PacketHandler 

type PacketHandler func(connection Connection, packet Packet)

handler for Packet

type PacketHandlerRegister 

type PacketHandlerRegister interface {
	Register(packetCommand PacketCommand, handler PacketHandler, protoMessage proto.Message)
}

type PacketHeader 

type PacketHeader interface {
	Len() uint32
	HasFlag(flag uint8) bool
	ReadFrom(packetHeaderData []byte)
	WriteTo(packetHeaderData []byte)
}

interface for PacketHeader

type ProtoCodec 

type ProtoCodec struct {
	RingBufferCodec

	// 在proto序列化后的数据,再做一层编码
	// encoder after proto.Message serialize
	ProtoPacketBytesEncoder func(protoPacketBytes [][]byte) [][]byte

	// 在proto反序列化之前,先做一层解码
	// decoder before proto.Message deserialize
	ProtoPacketBytesDecoder func(packetData []byte) []byte

	// 消息号和proto.Message type的映射表
	MessageCreatorMap map[PacketCommand]reflect.Type
}

codec for protobuf 

use DefaultPacketHeader,RingBufferCodec

func NewProtoCodec 

func NewProtoCodec(protoMessageTypeMap map[PacketCommand]reflect.Type) *ProtoCodec

func (*ProtoCodec) DecodePacket 

func (c *ProtoCodec) DecodePacket(connection Connection, packetHeader PacketHeader, packetData []byte) Packet

func (*ProtoCodec) EncodePacket 

func (c *ProtoCodec) EncodePacket(connection Connection, packet Packet) ([][]byte, uint8)

func (*ProtoCodec) Register 

func (c *ProtoCodec) Register(command PacketCommand, protoMessage proto.Message)

注册消息和proto.Message的映射 

protoMessage can be nil

type ProtoMessageCreator 

type ProtoMessageCreator func() proto.Message

proto.Message ctor func

type ProtoPacket 

type ProtoPacket struct {
	// contains filtered or unexported fields
}

packet for proto.Message

func NewProtoPacket 

func NewProtoPacket(command PacketCommand, message proto.Message) *ProtoPacket

func NewProtoPacketEx 

func NewProtoPacketEx(args ...any) *ProtoPacket

func NewProtoPacketWithData 

func NewProtoPacketWithData(command PacketCommand, data []byte) *ProtoPacket

func (*ProtoPacket) Clone 

func (p *ProtoPacket) Clone() Packet

deep copy

func (*ProtoPacket) Command 

func (p *ProtoPacket) Command() PacketCommand

func (*ProtoPacket) ErrorCode 

func (p *ProtoPacket) ErrorCode() uint32

func (*ProtoPacket) GetStreamData 

func (p *ProtoPacket) GetStreamData() []byte

某些特殊需求会直接使用序列化好的数据 

support stream data

func (*ProtoPacket) Message 

func (p *ProtoPacket) Message() proto.Message

func (*ProtoPacket) RpcCallId 

func (p *ProtoPacket) RpcCallId() uint32

func (*ProtoPacket) SetErrorCode 

func (p *ProtoPacket) SetErrorCode(code uint32) *ProtoPacket

func (*ProtoPacket) SetRpcCallId 

func (p *ProtoPacket) SetRpcCallId(rpcCallId uint32)

func (*ProtoPacket) WithRpc 

func (p *ProtoPacket) WithRpc(arg any) *ProtoPacket

type ProtoRegister 

type ProtoRegister interface {
	Register(command PacketCommand, protoMessage proto.Message)
}

type RingBuffer 

type RingBuffer struct {
	// contains filtered or unexported fields
}

环形buffer,专为TcpConnection定制,在收发包时,可以减少内存分配和拷贝 NOTE:不支持多线程,不具备通用性 

optimize for TcpConnection, reduce memory alloc and copy
not thread safe

func NewRingBuffer 

func NewRingBuffer(size int) *RingBuffer

指定大小的RingBuffer,不支持动态扩容 

fix size, not support dynamic expansion

func (*RingBuffer) GetBuffer 

func (b *RingBuffer) GetBuffer() []byte

Buffer

func (*RingBuffer) ReadBuffer 

func (b *RingBuffer) ReadBuffer() []byte

返回可读取的连续buffer(不产生copy) NOTE:调用ReadBuffer之前,需要先确保UnReadLength()>0 

continuous buffer can read

func (*RingBuffer) ReadFull 

func (b *RingBuffer) ReadFull(readLen int) []byte

读取指定长度的数据 

read data of a specified length

func (*RingBuffer) SetReaded 

func (b *RingBuffer) SetReaded(readedLength int)

设置已读取长度 

set readed length

func (*RingBuffer) SetWrote 

func (b *RingBuffer) SetWrote(wroteLength int)

设置已写入长度 

set wrote length

func (*RingBuffer) Size 

func (b *RingBuffer) Size() int

func (*RingBuffer) UnReadLength 

func (b *RingBuffer) UnReadLength() int

未被读取的长度

func (*RingBuffer) Write 

func (b *RingBuffer) Write(p []byte) (n int, err error)

func (*RingBuffer) WriteBuffer 

func (b *RingBuffer) WriteBuffer() []byte

返回可写入的连续buffer 

continuous buffer can write

type RingBufferCodec 

type RingBufferCodec struct {
	// 包头的编码接口,包头长度不能变
	// encoder for packer header
	HeaderEncoder func(connection Connection, packet Packet, headerData []byte)

	// 包体的编码接口
	// NOTE:返回值允许返回多个[]byte,如ProtoPacket在编码时,可以分别返回command和proto.Message的序列化[]byte
	// 如果只返回一个[]byte,就需要把command和proto.Message序列化的[]byte再合并成一个[]byte,造成性能损失
	// encoder for packer data
	// allow return multiple []byte,for example:when encode ProtoPacket, return two []byte(serialized data of command and proto.Message)
	DataEncoder func(connection Connection, packet Packet) ([][]byte, uint8)

	// 包头的解码接口,包头长度不能变
	// decoder for packer header
	HeaderDecoder func(connection Connection, headerData []byte)

	// 包体的解码接口
	// decoder for packer data
	DataDecoder func(connection Connection, packetHeader PacketHeader, packetData []byte) Packet
}

用了RingBuffer的连接的编解码接口 a codec with RingBuffer stream format: Length+Data process divided into two layers layer1: retrieve the original package data from RingBuffer layer2: encode or decode of the original package data NOTE: only support TcpConnection

func (*RingBufferCodec) CreatePacketHeader 

func (c *RingBufferCodec) CreatePacketHeader(connection Connection, packet Packet, packetData []byte) PacketHeader

func (*RingBufferCodec) Decode 

func (c *RingBufferCodec) Decode(connection Connection, data []byte) (newPacket Packet, err error)

func (*RingBufferCodec) Encode 

func (c *RingBufferCodec) Encode(connection Connection, packet Packet) []byte

TcpConnection做了优化,在Encode的过程中就直接写入sendBuffer 返回值:未能写入sendBuffer的数据(sendBuffer写满了的情况) encode packet and write encoded data to sendBuffer return the remain encoded data not wrote to sendBuffer(when sendBuffer is full)

func (*RingBufferCodec) PacketHeaderSize 

func (c *RingBufferCodec) PacketHeaderSize() uint32

type SendOption 

type SendOption interface {
	// contains filtered or unexported methods
}

option for Connection.Send

func Timeout 

func Timeout(timeout time.Duration) SendOption

func WithBlock 

func WithBlock() SendOption

阻塞模式(TODO)

func WithDiscard 

func WithDiscard() SendOption

消息满时丢弃

func WithInfiniteTimeout 

func WithInfiniteTimeout() SendOption

永不超时

type SimplePacketHeader 

type SimplePacketHeader struct {
	// (flags << 24) | len
	// flags [0,255)
	// len [0,16M)
	LenAndFlags uint32
	Command     uint16
}

a simple packet header for TcpConnectionSimple contains packet len and packet command

func NewSimplePacketHeader 

func NewSimplePacketHeader(len uint32, flags uint8, command PacketCommand) *SimplePacketHeader

func (*SimplePacketHeader) AddFlags 

func (h *SimplePacketHeader) AddFlags(flag uint8)

func (*SimplePacketHeader) Flags 

func (h *SimplePacketHeader) Flags() uint8

标记 [0,0xFF]

func (*SimplePacketHeader) HasFlag 

func (h *SimplePacketHeader) HasFlag(flag uint8) bool

func (*SimplePacketHeader) Len 

func (h *SimplePacketHeader) Len() uint32

包体长度,不包含包头的长度 

packet body length (without packet header's length)
[0,0x00FFFFFF]

func (*SimplePacketHeader) ReadFrom 

func (h *SimplePacketHeader) ReadFrom(packetHeaderData []byte)

从字节流读取数据,len(messageHeaderData)>=MessageHeaderSize 使用小端字节序 

parse LenAndFlags,Command from stream data

func (*SimplePacketHeader) SetFlags 

func (h *SimplePacketHeader) SetFlags(flags uint8)

func (*SimplePacketHeader) WriteTo 

func (h *SimplePacketHeader) WriteTo(packetHeaderData []byte)

写入字节流,使用小端字节序 

write LenAndFlags,Command to stream data

type SimpleProtoCodec 

type SimpleProtoCodec struct {
	// 消息号和proto.Message type的映射表
	MessageCreatorMap map[PacketCommand]reflect.Type
}

a simple protobuf codec for TcpConnectionSimple, without RingBuffer use SimplePacketHeader as PacketHeader

func NewSimpleProtoCodec 

func NewSimpleProtoCodec() *SimpleProtoCodec

func (*SimpleProtoCodec) CreatePacketHeader 

func (c *SimpleProtoCodec) CreatePacketHeader(connection Connection, packet Packet, packetData []byte) PacketHeader

func (*SimpleProtoCodec) Decode 

func (c *SimpleProtoCodec) Decode(connection Connection, data []byte) (newPacket Packet, err error)

func (*SimpleProtoCodec) Encode 

func (c *SimpleProtoCodec) Encode(connection Connection, packet Packet) []byte

func (*SimpleProtoCodec) PacketHeaderSize 

func (c *SimpleProtoCodec) PacketHeaderSize() uint32

func (*SimpleProtoCodec) Register 

func (c *SimpleProtoCodec) Register(command PacketCommand, protoMessage proto.Message)

注册消息和proto.Message的映射 

protoMessage can be nil

type StdLogger 

type StdLogger struct {
	// contains filtered or unexported fields
}

func (*StdLogger) Debug 

func (s *StdLogger) Debug(format string, args ...interface{})

func (*StdLogger) Error 

func (s *StdLogger) Error(format string, args ...interface{})

func (*StdLogger) Info 

func (s *StdLogger) Info(format string, args ...interface{})

func (*StdLogger) Warn 

func (s *StdLogger) Warn(format string, args ...interface{})

type TcpConnection 

type TcpConnection struct {
	// contains filtered or unexported fields
}

TcpConnection use RingBuffer to optimize

func NewTcpConnectionAccept 

func NewTcpConnectionAccept(conn net.Conn, config *ConnectionConfig) *TcpConnection

func NewTcpConnector 

func NewTcpConnector(config *ConnectionConfig) *TcpConnection

func (*TcpConnection) Close 

func (c *TcpConnection) Close()

func (*TcpConnection) Connect 

func (c *TcpConnection) Connect(address string) bool

func (*TcpConnection) GetCodec 

func (c *TcpConnection) GetCodec() Codec

func (*TcpConnection) GetConnectionId 

func (c *TcpConnection) GetConnectionId() uint32

unique id

func (*TcpConnection) GetHandler 

func (c *TcpConnection) GetHandler() ConnectionHandler

func (*TcpConnection) GetSendPacketChanLen 

func (c *TcpConnection) GetSendPacketChanLen() int

func (*TcpConnection) GetTag 

func (c *TcpConnection) GetTag() interface{}

获取关联数据 

get the associated tag

func (*TcpConnection) IsConnected 

func (c *TcpConnection) IsConnected() bool

func (*TcpConnection) IsConnector 

func (c *TcpConnection) IsConnector() bool

func (*TcpConnection) LocalAddr 

func (c *TcpConnection) LocalAddr() net.Addr

LocalAddr returns the local network address.

func (*TcpConnection) RemoteAddr 

func (c *TcpConnection) RemoteAddr() net.Addr

RemoteAddr returns the remote network address.

func (*TcpConnection) Rpc 

func (c *TcpConnection) Rpc(request Packet, reply proto.Message, opts ...SendOption) error

Rpc send a request to target and block wait reply

func (*TcpConnection) Send 

func (c *TcpConnection) Send(command PacketCommand, message proto.Message, opts ...SendOption) bool

发送proto包 

NOTE:如果是异步调用Send(command,message),调用之后,不要再对message进行读写!

func (*TcpConnection) SendPacket 

func (c *TcpConnection) SendPacket(packet Packet, opts ...SendOption) bool

发送数据 

NOTE:如果是异步调用SendPacket(command,message),调用之后,不要再对message进行读写!

func (*TcpConnection) SetCodec 

func (c *TcpConnection) SetCodec(codec Codec)

func (*TcpConnection) SetTag 

func (c *TcpConnection) SetTag(tag interface{})

设置关联数据 

set the associated tag

func (*TcpConnection) Start 

func (c *TcpConnection) Start(ctx context.Context, netMgrWg *sync.WaitGroup, onClose func(connection Connection))

start read&write goroutine

func (*TcpConnection) TrySendPacket 

func (c *TcpConnection) TrySendPacket(packet Packet, timeout time.Duration, opts ...SendOption) bool

超时发包,超时未发送则丢弃,适用于某些允许丢弃的数据包 可以防止某些"不重要的"数据包造成chan阻塞,比如游戏项目常见的聊天广播 

asynchronous send with Timeout (write to chan, not send immediately)
if return false, means not write to chan

type TcpConnectionSimple 

type TcpConnectionSimple struct {
	// contains filtered or unexported fields
}

不使用RingBuffer的TcpConnection 需要搭配对应的codec 

TcpConnection without RingBuffer

func NewTcpConnectionSimple 

func NewTcpConnectionSimple(config *ConnectionConfig) *TcpConnectionSimple

func NewTcpConnectionSimpleAccept 

func NewTcpConnectionSimpleAccept(conn net.Conn, config *ConnectionConfig) *TcpConnectionSimple

func (*TcpConnectionSimple) Close 

func (c *TcpConnectionSimple) Close()

func (*TcpConnectionSimple) Connect 

func (c *TcpConnectionSimple) Connect(address string) bool

func (*TcpConnectionSimple) GetCodec 

func (c *TcpConnectionSimple) GetCodec() Codec

func (*TcpConnectionSimple) GetConnectionId 

func (c *TcpConnectionSimple) GetConnectionId() uint32

unique id

func (*TcpConnectionSimple) GetHandler 

func (c *TcpConnectionSimple) GetHandler() ConnectionHandler

func (*TcpConnectionSimple) GetSendPacketChanLen 

func (c *TcpConnectionSimple) GetSendPacketChanLen() int

func (*TcpConnectionSimple) GetTag 

func (c *TcpConnectionSimple) GetTag() interface{}

获取关联数据 

get the associated tag

func (*TcpConnectionSimple) IsConnected 

func (c *TcpConnectionSimple) IsConnected() bool

func (*TcpConnectionSimple) IsConnector 

func (c *TcpConnectionSimple) IsConnector() bool

func (*TcpConnectionSimple) LocalAddr 

func (c *TcpConnectionSimple) LocalAddr() net.Addr

LocalAddr returns the local network address.

func (*TcpConnectionSimple) RemoteAddr 

func (c *TcpConnectionSimple) RemoteAddr() net.Addr

RemoteAddr returns the remote network address.

func (*TcpConnectionSimple) Rpc 

func (c *TcpConnectionSimple) Rpc(request Packet, reply proto.Message, opts ...SendOption) error

Rpc send a request to target and block wait reply

func (*TcpConnectionSimple) Send 

func (c *TcpConnectionSimple) Send(command PacketCommand, message proto.Message, opts ...SendOption) bool

发送proto包 

NOTE:如果是异步调用Send(command,message),调用之后,不要再对message进行读写!

func (*TcpConnectionSimple) SendPacket 

func (c *TcpConnectionSimple) SendPacket(packet Packet, opts ...SendOption) bool

发送数据 

NOTE:如果是异步调用SendPacket(command,message),调用之后,不要再对message进行读写!

func (*TcpConnectionSimple) SetCodec 

func (c *TcpConnectionSimple) SetCodec(codec Codec)

func (*TcpConnectionSimple) SetTag 

func (c *TcpConnectionSimple) SetTag(tag interface{})

设置关联数据 

set the associated tag

func (*TcpConnectionSimple) Start 

func (c *TcpConnectionSimple) Start(ctx context.Context, netMgrWg *sync.WaitGroup, onClose func(connection Connection))

start read&write goroutine

func (*TcpConnectionSimple) TrySendPacket 

func (c *TcpConnectionSimple) TrySendPacket(packet Packet, timeout time.Duration, opts ...SendOption) bool

超时发包,超时未发送则丢弃,适用于某些允许丢弃的数据包 可以防止某些"不重要的"数据包造成chan阻塞,比如游戏项目常见的聊天广播 

asynchronous send with Timeout (write to chan, not send immediately)
if return false, means not write to chan

type TcpListener 

type TcpListener struct {
	// contains filtered or unexported fields
}

tcp Listener

func NewTcpListener 

func NewTcpListener(listenerConfig *ListenerConfig) *TcpListener

func (*TcpListener) Addr 

func (l *TcpListener) Addr() net.Addr

Addr returns the listener's network address.

func (*TcpListener) Broadcast 

func (l *TcpListener) Broadcast(packet Packet)

广播消息 

broadcast packet to accepted connections

func (*TcpListener) Close 

func (l *TcpListener) Close()

关闭监听,并关闭管理的连接 

close listen, close the accepted connections

func (*TcpListener) GetConnection 

func (l *TcpListener) GetConnection(connectionId uint32) Connection

func (*TcpListener) GetListenerId 

func (l *TcpListener) GetListenerId() uint32

func (*TcpListener) IsRunning 

func (l *TcpListener) IsRunning() bool

func (*TcpListener) RangeConnections 

func (l *TcpListener) RangeConnections(f func(conn Connection) bool)

range for accepted connections

func (*TcpListener) Start 

func (l *TcpListener) Start(ctx context.Context, listenAddress string) bool

start goroutine

type TimeoutOption 

type TimeoutOption struct {
	Timeout time.Duration
}

type WsConnection 

type WsConnection struct {
	// contains filtered or unexported fields
}

WsConnection WebSocket

func NewWsConnection 

func NewWsConnection(config *ConnectionConfig) *WsConnection

func NewWsConnectionAccept 

func NewWsConnectionAccept(conn *websocket.Conn, config *ConnectionConfig, codec Codec, handler ConnectionHandler) *WsConnection

func (*WsConnection) Close 

func (c *WsConnection) Close()

func (*WsConnection) Connect 

func (c *WsConnection) Connect(address string) bool

func (*WsConnection) GetCodec 

func (c *WsConnection) GetCodec() Codec

func (*WsConnection) GetConn 

func (c *WsConnection) GetConn() *websocket.Conn

func (*WsConnection) GetConnectionId 

func (c *WsConnection) GetConnectionId() uint32

unique id

func (*WsConnection) GetHandler 

func (c *WsConnection) GetHandler() ConnectionHandler

func (*WsConnection) GetSendPacketChanLen 

func (c *WsConnection) GetSendPacketChanLen() int

func (*WsConnection) GetTag 

func (c *WsConnection) GetTag() interface{}

获取关联数据 

get the associated tag

func (*WsConnection) IsConnected 

func (c *WsConnection) IsConnected() bool

func (*WsConnection) IsConnector 

func (c *WsConnection) IsConnector() bool

func (*WsConnection) LocalAddr 

func (c *WsConnection) LocalAddr() net.Addr

func (*WsConnection) RemoteAddr 

func (c *WsConnection) RemoteAddr() net.Addr

func (*WsConnection) Rpc 

func (c *WsConnection) Rpc(request Packet, reply proto.Message, opts ...SendOption) error

Rpc send a request to target and block wait reply

func (*WsConnection) Send 

func (c *WsConnection) Send(command PacketCommand, message proto.Message, opts ...SendOption) bool

发送proto包 

NOTE:如果是异步调用Send(command,message),调用之后,不要再对message进行读写!

func (*WsConnection) SendPacket 

func (c *WsConnection) SendPacket(packet Packet, opts ...SendOption) bool

发送数据 

NOTE:如果是异步调用SendPacket(command,message),调用之后,不要再对message进行读写!

func (*WsConnection) SetCodec 

func (c *WsConnection) SetCodec(codec Codec)

func (*WsConnection) SetTag 

func (c *WsConnection) SetTag(tag interface{})

设置关联数据 

set the associated tag

func (*WsConnection) Start 

func (c *WsConnection) Start(ctx context.Context, netMgrWg *sync.WaitGroup, onClose func(connection Connection))

func (*WsConnection) TrySendPacket 

func (c *WsConnection) TrySendPacket(packet Packet, timeout time.Duration, opts ...SendOption) bool

超时发包,超时未发送则丢弃,适用于某些允许丢弃的数据包 可以防止某些"不重要的"数据包造成chan阻塞,比如游戏项目常见的聊天广播 

asynchronous send with Timeout (write to chan, not send immediately)
if return false, means not write to chan

type WsListener 

type WsListener struct {
	// contains filtered or unexported fields
}

func NewWsListener 

func NewWsListener(listenerConfig *ListenerConfig) *WsListener

func (*WsListener) Addr 

func (l *WsListener) Addr() net.Addr

func (*WsListener) Broadcast 

func (l *WsListener) Broadcast(packet Packet)

func (*WsListener) Close 

func (l *WsListener) Close()

func (*WsListener) GetConnection 

func (l *WsListener) GetConnection(connectionId uint32) Connection

func (*WsListener) GetListenerId 

func (l *WsListener) GetListenerId() uint32

func (*WsListener) IsRunning 

func (l *WsListener) IsRunning() bool

func (*WsListener) RangeConnections 

func (l *WsListener) RangeConnections(f func(conn Connection) bool)

range for accepted connections

func (*WsListener) Start 

func (l *WsListener) Start(ctx context.Context, listenAddress string) bool

type XorProtoCodec 

type XorProtoCodec struct {
	*ProtoCodec
	// contains filtered or unexported fields
}

codec for proto.Message and xor

func NewXorProtoCodec 

func NewXorProtoCodec(xorKey []byte, protoMessageTypeMap map[PacketCommand]reflect.Type) *XorProtoCodec

Source Files

View all Source files

Directories

Path Synopsis
example
custom_packet/client command
custom_packet/codec
custom_packet/server command
data_packet/client command
data_packet/server command
helloworld/client command
helloworld/server command
packet_size/client command
packet_size/server command
pb
rpc/client command
rpc/server command
simulate_game command
tcp_connection_simple/client command
tcp_connection_simple/server command
websocket/client command
websocket/server command

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.