README
¶
znet
golang powerful network framework
Features
- High-performance Event-Loop under multi-threads model
- Supporting multiple protocols: TCP,Websocket
- Supporting two event-notification mechanisms: epoll in Linux/Windows and kqueue in FreeBSD
- Supporting safe goroutines worker pool
- Supporting two contentType: JSON/Protobuf
Quick start
- install
go get -u github.com/ebar-go/znet
- go run server.go
// server.go
package main
import (
"github.com/ebar-go/ego/utils/runtime/signal"
"github.com/ebar-go/znet"
)
const(
OperateFoo = 1 // define a foo operate
)
func main() {
// new instance
instance := znet.New()
// listen tcp and websocket
instance.ListenTCP(":8081")
instance.ListenWebsocket(":8082")
// register a router for some operate
instance.Router().Route(OperateFoo, func(ctx *znet.Context) (any, error) {
// return response to the client
return map[string]any{"val": "bar"}, nil
})
// run the instance, graceful stop by ctrl-c.
instance.Run(signal.SetupSignalHandler())
}
- go run client.go
// client.go
package main
import (
"context"
"github.com/ebar-go/znet/codec"
"log"
"net"
"time"
)
func main() {
conn, err := net.Dial("tcp", "localhost:8081")
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
go func() {
for {
bytes := make([]byte, 512)
n, err := conn.Read(bytes)
if err != nil {
return
}
log.Println("receive response: ", string(bytes[:n]))
}
}()
ctx, cancel := context.WithTimeout(context.Background(), time.Second*10)
defer cancel()
go func() {
for {
select {
case <-ctx.Done():
default:
packet := &codec.Packet{Operate: OperateFoo, ContentType: codec.ContentTypeJSON}
bytes, err := codec.Default().Pack(packet, map[string]any{"key": "foo"})
if err != nil {
return
}
conn.Write(bytes)
time.Sleep(time.Second)
}
}
}()
<-ctx.Done()
}
Protocol
- We design the protocol for solve TCP sticky packet problem
- ByteOrder: big endian
|-------------- header ------------- |-------- body --------|
|packetLength|operate|contentType|seq|-------- body --------|
| 4 | 2 | 2 | 2 | n |
Benchmark
Documentation
¶
Index ¶
- type Callback
- type Connection
- func (conn *Connection) AddBeforeCloseHook(hooks ...func(conn *Connection))
- func (conn *Connection) Close()
- func (conn *Connection) ID() string
- func (conn *Connection) Property() *internal.Container[string, any]
- func (conn *Connection) Push(p []byte)
- func (conn *Connection) Read(p []byte) (int, error)
- func (conn *Connection) ReadPacket(buf []byte, packetLengthSize int) (n int, err error)
- func (conn *Connection) Write(p []byte) (int, error)
- type ConnectionHandler
- type Context
- type Engine
- type EventLoop
- type HandleFunc
- type Handler
- type Instance
- type MainReactor
- type Option
- type Options
- type ReactorOptions
- type Router
- type ShardSubReactor
- func (shard *ShardSubReactor) GetConnection(fd int) *Connection
- func (shard *ShardSubReactor) Offer(fds ...int)
- func (shard *ShardSubReactor) Polling(stopCh <-chan struct{}, handler func(active int))
- func (shard *ShardSubReactor) RegisterConnection(conn *Connection)
- func (shard *ShardSubReactor) UnregisterConnection(conn *Connection)
- type SubReactor
- func (sub *SubReactor) GetConnection(fd int) *Connection
- func (sub *SubReactor) Offer(fds ...int)
- func (sub *SubReactor) Polling(stopCh <-chan struct{}, handler func(active int))
- func (sub *SubReactor) RegisterConnection(conn *Connection)
- func (sub *SubReactor) UnregisterConnection(conn *Connection)
- type SubReactorInstance
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Callback ¶
type Callback struct {
// contains filtered or unexported fields
}
Callback manage connection callback handlers.
func (*Callback) OnConnect ¶
func (callback *Callback) OnConnect(conn *Connection)
OnConnect is called when the connection is established
func (*Callback) OnDisconnect ¶
func (callback *Callback) OnDisconnect(conn *Connection)
OnDisconnect is called when the connection is closed
type Connection ¶
type Connection struct {
// contains filtered or unexported fields
}
Connection represents client connection
func NewConnection ¶
func NewConnection(conn net.Conn, fd int) *Connection
NewConnection returns a new Connection instance
func (*Connection) AddBeforeCloseHook ¶
func (conn *Connection) AddBeforeCloseHook(hooks ...func(conn *Connection))
AddBeforeCloseHook adds a hook to the connection before closed
func (*Connection) ID ¶
func (conn *Connection) ID() string
UIID returns the uuid associated with the connection
func (*Connection) Property ¶
func (conn *Connection) Property() *internal.Container[string, any]
Property return properties container
func (*Connection) Push ¶
func (conn *Connection) Push(p []byte)
Push send message to the connection
func (*Connection) Read ¶
func (conn *Connection) Read(p []byte) (int, error)
Read reads message from the connection
func (*Connection) ReadPacket ¶
func (conn *Connection) ReadPacket(buf []byte, packetLengthSize int) (n int, err error)
ReadPacket reads a line message from the connection
type Engine ¶
type Engine struct {
// contains filtered or unexported fields
}
Engine represents context manager
func (*Engine) HandleContext ¶
HandleContext handles context
func (*Engine) NewContext ¶
func (e *Engine) NewContext(conn *Connection, bytes []byte) *Context
NewContext return a new Context instance
func (*Engine) ReleaseContext ¶
type EventLoop ¶ added in v0.1.1
type EventLoop struct {
// contains filtered or unexported fields
}
EventLoop represents im framework public access api.
func (*EventLoop) ListenWebsocket ¶ added in v0.1.1
ListenWebsocket listens for websocket connections
type HandleFunc ¶
type HandleFunc func(ctx *Context)
HandlerFunc represents a handler function for Context
type Instance ¶
type Instance interface {
// Router return the router instance
Router() *Router
// ListenTCP listen tcp server
ListenTCP(addr string)
// ListenWebsocket listen websocket server
ListenWebsocket(addr string)
// Run runs the instance with the given signal handler
Run(stopCh <-chan struct{}) error
}
Instance represents an el interface
type MainReactor ¶
type MainReactor struct {
// contains filtered or unexported fields
}
MainReactor represents the epoll model for processing action connections.
func NewMainReactor ¶
func NewMainReactor(options ReactorOptions) (*MainReactor, error)
NewMainReactor return a new main reactor instance
func (*MainReactor) Run ¶
func (reactor *MainReactor) Run(stopCh <-chan struct{})
Run runs the MainReactor with the given signal.
type Options ¶
type Options struct {
// Debug enables debug logging
Debug bool
// OnConnect is a callback function that is called when the connection is established
OnConnect ConnectionHandler
// OnDisconnect is a callback function that is called when the connection is closed
OnDisconnect ConnectionHandler
// Middlewares is a lot of callback functions that are called when the connection send new message
Middlewares []HandleFunc
Reactor ReactorOptions
}
Options represents app options
func (*Options) NewMainReactor ¶
func (options *Options) NewMainReactor() *MainReactor
type ReactorOptions ¶
type ReactorOptions struct {
// EpollBufferSize is the size of the active connections in every duration,default is 100
EpollBufferSize int
// WorkerPollSize is the size of the worker pool, default is 1000
WorkerPoolSize int
// ThreadQueueCapacity is the cap of the thread queue, default is 100
ThreadQueueCapacity int
// MaxReadBufferSize is the size of the max read buffer, default is 512
MaxReadBufferSize int
// SubReactorShardCount is the number of sub-reactor shards, default is 32
// if the parameter is zero, the number of sub-reactor will be 1
SubReactorShardCount int
// contains filtered or unexported fields
}
ReactorOptions represents the options for the reactor
type Router ¶
type Router struct {
// contains filtered or unexported fields
}
Router
func (*Router) OnNotFound ¶
OnNotFound is called when operation is not found
type ShardSubReactor ¶
type ShardSubReactor struct {
// contains filtered or unexported fields
}
func NewShardSubReactor ¶
func NewShardSubReactor(shardCount, bufferSize int) *ShardSubReactor
func (*ShardSubReactor) GetConnection ¶
func (shard *ShardSubReactor) GetConnection(fd int) *Connection
func (*ShardSubReactor) Offer ¶
func (shard *ShardSubReactor) Offer(fds ...int)
func (*ShardSubReactor) Polling ¶
func (shard *ShardSubReactor) Polling(stopCh <-chan struct{}, handler func(active int))
func (*ShardSubReactor) RegisterConnection ¶
func (shard *ShardSubReactor) RegisterConnection(conn *Connection)
func (*ShardSubReactor) UnregisterConnection ¶
func (shard *ShardSubReactor) UnregisterConnection(conn *Connection)
type SubReactor ¶
type SubReactor struct {
// contains filtered or unexported fields
}
SubReactor represents sub reactor
func NewSubReactor ¶
func NewSubReactor(bufferSize int) *SubReactor
NewSubReactor creates a instance of a SubReactor
func (*SubReactor) GetConnection ¶
func (sub *SubReactor) GetConnection(fd int) *Connection
GetConnection returns a connection by fd
func (*SubReactor) Offer ¶
func (sub *SubReactor) Offer(fds ...int)
Offer push the active connections fd to the queue
func (*SubReactor) Polling ¶
func (sub *SubReactor) Polling(stopCh <-chan struct{}, handler func(active int))
Polling poll with callback function
func (*SubReactor) RegisterConnection ¶
func (sub *SubReactor) RegisterConnection(conn *Connection)
RegisterConnection registers a new connection to the epoll listener
func (*SubReactor) UnregisterConnection ¶
func (sub *SubReactor) UnregisterConnection(conn *Connection)
UnregisterConnection removes the connection from the epoll listener
type SubReactorInstance ¶
type SubReactorInstance interface {
RegisterConnection(conn *Connection)
UnregisterConnection(conn *Connection)
GetConnection(fd int) *Connection
Offer(fds ...int)
Polling(stopCh <-chan struct{}, handler func(active int))
}
Source Files
¶
Directories
Click to show internal directories.
Click to hide internal directories.