znet

package module
v0.1.7 Latest Latest
Warning

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

 Go to latest Published: Dec 5, 2022 License: GPL-3.0 Imports: 15 Imported by: 1

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(
	ActionFoo = 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(ActionFoo, func(ctx *znet.Context) (any, error) {
		// return response to the client
		return map[string]any{"foo": "bar"}, nil
	})
	
	// run the instance, graceful stop by ctrl-c.
	instance.Run(signal.SetupSignalHandler())
}

Architecture

  • Framework Framework
  • Engine Start Sequence Diagram
    Sequence Diagram

Protocol

  • TCP : We design the protocol for sticky packet problem
|-------------- header --------------|-------- body --------|
|packetLength| action |      seq     |-------- body --------|
|     4      |    2   |       2      |          n           |
  • WebSocket : don't need the packet length
|-------------- header --------------|-------- body --------|
|        action       |      seq     |-------- body --------|
|           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

Constants

View Source 
const (
	BeforeServerStart    = "beforeServerStart"
	AfterServerStart     = "afterServerStart"
	BeforeServerShutdown = "beforeServerShutdown"
	AfterServerShutdown  = "afterServerShutdown"
)
View Source 
const (
	ContentTypeJson  = "json"
	ContentTypeProto = "protobuf"
)

Variables

This section is empty.

Functions

func RegisterEvent added in v0.1.7 

func RegisterEvent(event string, callback func())

RegisterEvent registers an event handler

Types

type Action added in v0.1.3 

type Action[Request, Response any] func(ctx *Context, request *Request) (*Response, error)

Action isa generic function that is friendly to user

type Callback 

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

Callback manage connection callback handlers.

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) Close 

func (conn *Connection) Close()

Close closes the connection

func (*Connection) ID 

func (conn *Connection) ID() string

ID returns the uuid of the connection

func (*Connection) IP added in v0.1.4 

func (conn *Connection) IP() string

func (*Connection) Property 

func (conn *Connection) Property() *structure.ConcurrentMap[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) Write 

func (conn *Connection) Write(p []byte) (int, error)

Write writes message to the connection

type ConnectionHandler 

type ConnectionHandler func(conn *Connection)

ConnectionHandler represents a connection handler

type Context 

type Context struct {
	context.Context
	// contains filtered or unexported fields
}

Context represents a context for request

func (*Context) Abort 

func (ctx *Context) Abort()

Abort stop invoke handler

func (*Context) Bind added in v0.1.7 

func (ctx *Context) Bind(container any) error

func (*Context) Conn 

func (ctx *Context) Conn() *Connection

Conn return instance of Connection

func (*Context) Next 

func (ctx *Context) Next()

Next invoke next handler

func (*Context) Packet added in v0.1.4 

func (ctx *Context) Packet() *codec.Packet

type Engine 

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

Engine provide context/handler management

func NewEngine 

func NewEngine() *Engine

func (*Engine) Use 

func (e *Engine) Use(handlers ...HandleFunc)

type HandleFunc 

type HandleFunc func(ctx *Context)

HandleFunc represents a handler function for Context

type Handler 

type Handler func(ctx *Context) (any, error)

Handler is a handler for operation

func StandardHandler 

func StandardHandler[Request, Response any](action Action[Request, Response]) Handler

StandardHandler is a function to convert standard handler.

type Network added in v0.1.7 

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

Network socket server master

func New 

func New(setters ...Option) *Network

New returns a new instance

func (*Network) ListenTCP added in v0.1.7 

func (instance *Network) ListenTCP(addr string)

ListenTCP listens for tcp connections

func (*Network) ListenWebsocket added in v0.1.7 

func (instance *Network) ListenWebsocket(addr string)

ListenWebsocket listens for websocket connections

func (*Network) Router added in v0.1.7 

func (instance *Network) Router() *Router

Router return instance of Router

func (*Network) Run added in v0.1.7 

func (instance *Network) Run(stopCh <-chan struct{}) error

Run starts the event-loop

type Option 

type Option func(options *Options)

type Options 

type Options struct {
	// Debug enables debug logging
	Debug bool
	// OnOpen is a callback function that is called when the connection is established
	OnOpen ConnectionHandler

	// OnClose is a callback function that is called when the connection is closed
	OnClose ConnectionHandler

	// OnError is a callback function that is called when process error
	OnError func(ctx *Context, err error)

	// Middlewares is a lot of callback functions that are called when the connection send new message
	Middlewares []HandleFunc

	Reactor ReactorOptions

	Thread ThreadOptions

	Acceptor acceptor.Options
}

Options represents app options

func (*Options) NewCallback added in v0.1.6 

func (options *Options) NewCallback() *Callback

func (*Options) NewReactorOrDie added in v0.1.4 

func (options *Options) NewReactorOrDie() *Reactor

func (*Options) NewRouter added in v0.1.3 

func (options *Options) NewRouter() *Router

func (*Options) NewThread added in v0.1.3 

func (options *Options) NewThread() *Thread

func (*Options) Validate added in v0.1.1 

func (options *Options) Validate() error

Validate validates the options parameter

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 *Reactor, err 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

func (ReactorOptions) NewSubReactor added in v0.1.5 

func (options ReactorOptions) NewSubReactor() SubReactor

type Router 

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

Router represents router instance

func NewRouter 

func NewRouter() *Router

func (*Router) OnNotFound 

func (router *Router) OnNotFound(handler HandleFunc) *Router

OnNotFound is called when operation is not found

func (*Router) Route 

func (router *Router) Route(action int16, handler Handler) *Router

Route register handler for action

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 (thread *Thread) HandleRequest(conn *Connection)

HandleRequest handle new request for connection

func (*Thread) Use added in v0.1.3 

func (thread *Thread) Use(handlers ...HandleFunc)

Use registers middleware

type ThreadOptions added in v0.1.3 

type ThreadOptions struct {
	// MaxReadBufferSize is the size of the max read buffer, default is 512
	MaxReadBufferSize int

	ContentType string

	WorkerPool *pool.Options
	// contains filtered or unexported fields
}

func (ThreadOptions) NewCodec added in v0.1.5 

func (options ThreadOptions) NewCodec() (cc codec.Codec)

func (ThreadOptions) NewWorkerPool added in v0.1.6 

func (options ThreadOptions) NewWorkerPool() pool.GoroutinePool

Source Files

View all Source files

Directories

Path Synopsis
examples
chat Module
wepoll

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.