README
¶
znet
golang powerful network framework
Features
- High-performance Event-Loop under multi-threads model
- Supporting multiple protocols: TCP,Websocket
- Supporting reactor event-notification mechanisms: epoll in Linux/Windows and kqueue in FreeBSD
- Supporting safe goroutines worker pool
- Supporting two contentType: JSON/Protobuf
- Supporting router service for different operate and handle functions
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():
return
default:
}
packet := codec.Factory().NewWithHeader(codec.Header{Operate: OperateFoo, ContentType: codec.ContentTypeJSON})
bytes, err := packet.Pack(map[string]any{"key": "foo"})
if err != nil {
return
}
conn.Write(bytes)
time.Sleep(time.Second)
}
}()
<-ctx.Done()
}
Architecture
- Framework
- Engine Start Sequence Diagram
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
goos: linux
goarch: amd64
pkg: github.com/ebar-go/znet
cpu: Intel(R) Core(TM) i7-9700 CPU @ 3.00GHz
|-----------------------------------|
| connections | memory | cpu |
|-----------------------------------|
| 10000 | 50MB | |
|-----------------------------------|
Documentation
¶
Index ¶
- type Action
- type Callback
- type Connection
- func (conn *Connection) AddBeforeCloseHook(hooks ...func(conn *Connection))
- func (conn *Connection) Close()
- func (conn *Connection) ID() string
- func (conn *Connection) IP() 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) Write(p []byte) (int, error)
- type ConnectionHandler
- type Context
- type Engine
- type HandleFunc
- type Handler
- type Instance
- type Option
- type Options
- type Reactor
- 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{}, callback func(int))
- func (shard *ShardSubReactor) RegisterConnection(conn *Connection)
- func (shard *ShardSubReactor) UnregisterConnection(conn *Connection)
- type SingleSubReactor
- func (sub *SingleSubReactor) GetConnection(fd int) *Connection
- func (sub *SingleSubReactor) Offer(fds ...int)
- func (sub *SingleSubReactor) Polling(stopCh <-chan struct{}, callback func(int))
- func (sub *SingleSubReactor) RegisterConnection(conn *Connection)
- func (sub *SingleSubReactor) UnregisterConnection(conn *Connection)
- type SubReactor
- type Thread
- type ThreadOptions
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) IP ¶ added in v0.1.4
func (conn *Connection) IP() string
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
type ConnectionHandler ¶
type ConnectionHandler func(conn *Connection)
ConnectionHandler represents a connection handler
type Engine ¶
type Engine struct {
// contains filtered or unexported fields
}
func (*Engine) ListenWebsocket ¶ added in v0.1.4
ListenWebsocket listens for websocket connections
type HandleFunc ¶
type HandleFunc func(ctx *Context)
HandleFunc represents a handler function for Context
type Handler ¶
Handler is a handler for operation
func StandardHandler ¶
StandardHandler is a function to convert standard handler.
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 eng interface
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
Thread ThreadOptions
}
Options represents app options
func (*Options) NewReactorOrDie ¶ added in v0.1.4
type Reactor ¶
type Reactor struct {
// contains filtered or unexported fields
}
Reactor represents the epoll model for listen connections.
func NewReactor ¶ added in v0.1.4
func NewReactor(options ReactorOptions) (*Reactor, error)
NewReactor return a new main reactor instance
func (*Reactor) Run ¶
func (reactor *Reactor) Run(stopCh <-chan struct{}, onRequest ConnectionHandler)
Run runs the Reactor with the given signal.
type ReactorOptions ¶
type ReactorOptions struct {
// EpollBufferSize is the size of the active connections in every duration,default is 100
EpollBufferSize int
// ThreadQueueCapacity is the cap of the thread queue, default is 100
ThreadQueueCapacity 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
}
ReactorOptions represents the options for the reactor
type Router ¶
type Router struct {
// contains filtered or unexported fields
}
Router represents router instance
func (*Router) OnNotFound ¶
func (router *Router) OnNotFound(handler HandleFunc) *Router
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{}, callback func(int))
func (*ShardSubReactor) RegisterConnection ¶
func (shard *ShardSubReactor) RegisterConnection(conn *Connection)
func (*ShardSubReactor) UnregisterConnection ¶
func (shard *ShardSubReactor) UnregisterConnection(conn *Connection)
type SingleSubReactor ¶ added in v0.1.3
type SingleSubReactor struct {
// contains filtered or unexported fields
}
SingleSubReactor represents sub reactor
func NewSingleSubReactor ¶ added in v0.1.3
func NewSingleSubReactor(bufferSize int) *SingleSubReactor
NewSingleSubReactor creates an instance of a SingleSubReactor
func (*SingleSubReactor) GetConnection ¶ added in v0.1.3
func (sub *SingleSubReactor) GetConnection(fd int) *Connection
GetConnection returns a connection by fd
func (*SingleSubReactor) Offer ¶ added in v0.1.3
func (sub *SingleSubReactor) Offer(fds ...int)
Offer push the active connections fd to the queue
func (*SingleSubReactor) Polling ¶ added in v0.1.3
func (sub *SingleSubReactor) Polling(stopCh <-chan struct{}, callback func(int))
Polling poll with callback function
func (*SingleSubReactor) RegisterConnection ¶ added in v0.1.3
func (sub *SingleSubReactor) RegisterConnection(conn *Connection)
RegisterConnection registers a new connection to the epoll listener
func (*SingleSubReactor) UnregisterConnection ¶ added in v0.1.3
func (sub *SingleSubReactor) UnregisterConnection(conn *Connection)
UnregisterConnection removes the connection from the epoll listener
type SubReactor ¶
type SubReactor interface {
RegisterConnection(conn *Connection)
UnregisterConnection(conn *Connection)
GetConnection(fd int) *Connection
Offer(fds ...int)
Polling(stopCh <-chan struct{}, callback func(int))
}
type Thread ¶ added in v0.1.3
type Thread struct {
// contains filtered or unexported fields
}
Thread represents context manager
func NewThread ¶ added in v0.1.3
func NewThread(options ThreadOptions) *Thread
NewThread returns a new Thread instance
func (*Thread) HandleRequest ¶ added in v0.1.3
func (e *Thread) HandleRequest(conn *Connection)
HandleRequest handle new request for connection
func (*Thread) Use ¶ added in v0.1.3
func (e *Thread) Use(handler ...HandleFunc)
Use registers middleware
type ThreadOptions ¶ added in v0.1.3
Source Files
¶
Directories
Click to show internal directories.
Click to hide internal directories.