vex

README

⛓ Vex

Vex 是一个使用 tcp 通信和传输数据的框架。

Read me in English

🍃 功能特性

• 基于 tcp 传输数据，直接使用或二次开发都很简单
• 极简设计的 API，内置连接池，可以对性能进行调优
• 支持客户端、服务器引入拦截器，方便接入监控和告警
• 支持信号量监控机制和平滑下线
• 支持连接数限制，并支持超时中断
• 自带 pack 数据传输协议，用于简单的数据传输场景

历史版本的特性请查看 HISTORY.md。未来版本的新特性和计划请查看 FUTURE.md。

📃 协议描述

自带的 pack 数据传输协议抽象出了一个数据包的概念，不管是请求还是响应都视为一种数据包。

ABNF：

PACKET = MAGIC TYPE DATASIZE DATA ; 数据包
MAGIC = 3OCTET ; 魔数，3 个字节表示，目前是 0xC638B，也就是 811915
TYPE = OCTET ; 数据包类型，0x00-0xFF，从 0 开始，最多 255 种数据包类型
DATASIZE = 4OCTET ; 数据包的数据大小，4 个字节表示，最大是 4GB
DATA = *OCTET ; 数据包的数据，大小未知，需要靠 DATASIZE 来确认

人类语言描述：

数据包：
magic     type    data_size    {data}
3byte     1byte     4byte      unknown

🔦 使用案例

$ go get -u github.com/FishGoddess/vex

我们提供了原生和 pack 两种使用方式，其中原生可以自定义协议，随意读写操作数据，用于二次开发，而 pack 则是自带的数据包传输协议，用于简单的数据传输场景。

原生客户端：

package main

import (
	"fmt"

	"github.com/FishGoddess/vex"
)

func main() {
	client, err := vex.NewClient("127.0.0.1:6789")
	if err != nil {
		panic(err)
	}

	defer client.Close()

	msg := []byte("hello")
	if _, err := client.Write(msg); err != nil {
		panic(err)
	}

	var buf [1024]byte
	n, err := client.Read(buf[:])
	if err != nil {
		panic(err)
	}

	fmt.Println("Received:", string(buf[:n]))
}

原生服务端：

package main

import (
	"fmt"
	"io"

	"github.com/FishGoddess/vex"
)

func handle(ctx *vex.Context) {
	var buf [1024]byte
	for {
		n, err := ctx.Read(buf[:])
		if err == io.EOF {
			break
		}

		if err != nil {
			panic(err)
		}

		fmt.Println("Received:", string(buf[:n]))

		if _, err = ctx.Write(buf[:n]); err != nil {
			panic(err)
		}
	}
}

func main() {
	// Create a server listening on 127.0.0.1:6789 and set a handle function to it.
	// Also, we can give it a name like "echo" so we can see it in logs.
	server := vex.NewServer("127.0.0.1:6789", handle, vex.WithName("echo"))

	// Use Serve() to begin serving.
	// Press ctrl+c/control+c to close the server.
	if err := server.Serve(); err != nil {
		panic(err)
	}
}

Pack 客户端：

package main

import (
	"fmt"

	"github.com/FishGoddess/vex"
	"github.com/FishGoddess/vex/pack"
)

func main() {
	client, err := vex.NewClient("127.0.0.1:6789")
	if err != nil {
		panic(err)
	}

	defer client.Close()

	// Use Send method to send a packet to server and receive a packet from server.
	// Try to change 'hello' to 'error' and see what happens.
	packet, err := pack.Send(client, 1, []byte("hello"))
	if err != nil {
		panic(err)
	}

	fmt.Println(string(packet))
}

Pack 服务端：

package main

import (
	"context"
	"errors"
	"fmt"

	"github.com/FishGoddess/vex"
	"github.com/FishGoddess/vex/pack"
)

func newRouter() *pack.Router {
	router := pack.NewRouter()

	// Use Register method to register your handler for some packets.
	router.Register(1, func(ctx context.Context, packetType pack.PacketType, requestPacket []byte) (responsePacket []byte, err error) {
		msg := string(requestPacket)
		fmt.Println(msg)

		if msg == "error" {
			return nil, errors.New(msg)
		} else {
			return requestPacket, nil
		}
	})

	return router
}

func main() {
	// Create a router for packets.
	router := newRouter()

	// Create a server listening on 127.0.0.1:6789 and set a handle function to it.
	server := vex.NewServer("127.0.0.1:6789", router.Handle, vex.WithName("pack"))

	// Use Serve() to begin serving.
	// Press ctrl+c/control+c to close the server.
	if err := server.Serve(); err != nil {
		panic(err)
	}
}

所有的使用案例都在 _examples 目录。

🛠 性能测试

$ make bench
BenchmarkReadWrite-2      140317              8356 ns/op               0 B/op          0 allocs/op

$ make benchpack
BenchmarkPackReadWrite-2   61564             19650 ns/op            2080 B/op          6 allocs/op

协议	连接数	rps
-	1	50231
-	2	116790
Pack	1	30852
Pack	2	67453

数据包大小为 1KB。

测试环境：AMD EPYC 7K62, 2 Cores, 8GB RAM, linux。

Documentation

Index

• type Client
  • func NewClient(address string, opts ...Option) (Client, error)
• type Config
  • func (c *Config) ApplyOptions(opts []Option) *Config
• type Context
  • func (c *Context) Cancel()
  • func (c *Context) Deadline() (deadline time.Time, ok bool)
  • func (c *Context) Done() <-chan struct{}
  • func (c *Context) Err() error
  • func (c *Context) LocalAddr() net.Addr
  • func (c *Context) Read(p []byte) (n int, err error)
  • func (c *Context) RemoteAddr() net.Addr
  • func (c *Context) Value(key any) any
  • func (c *Context) Write(p []byte) (n int, err error)
• type HandleFunc
• type Option
  • func WithAfterClosing(afterClosing func(address string)) Option
  • func WithAfterHandling(afterHandling func(ctx *Context)) Option
  • func WithAfterServing(afterServing func(address string)) Option
  • func WithBeforeClosing(beforeClosing func(address string)) Option
  • func WithBeforeHandling(beforeHandling func(ctx *Context)) Option
  • func WithBeforeServing(beforeServing func(address string)) Option
  • func WithCloseTimeout(timeout time.Duration) Option
  • func WithConnectTimeout(timeout time.Duration) Option
  • func WithMaxConnections(maxConnections uint32) Option
  • func WithName(name string) Option
  • func WithOnConnected(onConnected func(clientAddress string, serverAddress string)) Option
  • func WithOnDisconnected(onDisconnected func(clientAddress string, serverAddress string)) Option
  • func WithReadBufferSize(bufferSize uint32) Option
  • func WithReadTimeout(timeout time.Duration) Option
  • func WithWriteBufferSize(bufferSize uint32) Option
  • func WithWriteTimeout(timeout time.Duration) Option
  • func (o Option) ApplyTo(conf *Config)
• type Server
  • func NewServer(address string, handle HandleFunc, opts ...Option) Server
• type Status

Types

type Client

type Client interface {
	io.ReadWriteCloser
}

func NewClient

func NewClient(address string, opts ...Option) (Client, error)

NewClient creates a new client connecting to address. Return an error if connect failed.

type Config

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

func (*Config) ApplyOptions

func (c *Config) ApplyOptions(opts []Option) *Config

type Context

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

Context connects client and server which can be read and written.

func (*Context) Cancel

func (c *Context) Cancel()

Cancel cancels the context.

func (*Context) Deadline

func (c *Context) Deadline() (deadline time.Time, ok bool)

Deadline returns the time when context has done. See context.Context.

func (*Context) Done

func (c *Context) Done() <-chan struct{}

Done returns a channel that's closed when context has done. See context.Context.

func (*Context) Err

func (c *Context) Err() error

Err returns the underlying error. See context.Context.

func (*Context) LocalAddr

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

LocalAddr returns the local network address.

func (*Context) Read

func (c *Context) Read(p []byte) (n int, err error)

Read reads data to p. See io.Reader.

func (*Context) RemoteAddr

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

RemoteAddr returns the remote network address.

func (*Context) Value

func (c *Context) Value(key any) any

Value returns the value associated with this context for key, or nil if no value is associated with key. See context.Context.

func (*Context) Write

func (c *Context) Write(p []byte) (n int, err error)

Write writes p to data. See io.Writer.

type HandleFunc

type HandleFunc func(ctx *Context)

HandleFunc is a function for handling connected context. You should design your own handler function for your server.

type Option

type Option func(conf *Config)

func WithAfterClosing

func WithAfterClosing(afterClosing func(address string)) Option

WithAfterClosing sets after closing function to config.

func WithAfterHandling

func WithAfterHandling(afterHandling func(ctx *Context)) Option

WithAfterHandling sets after handling function to config.

func WithAfterServing

func WithAfterServing(afterServing func(address string)) Option

WithAfterServing sets after serving function to config.

func WithBeforeClosing

func WithBeforeClosing(beforeClosing func(address string)) Option

WithBeforeClosing sets before closing function to config.

func WithBeforeHandling

func WithBeforeHandling(beforeHandling func(ctx *Context)) Option

WithBeforeHandling sets before handling function to config.

func WithBeforeServing

func WithBeforeServing(beforeServing func(address string)) Option

WithBeforeServing sets before serving function to config.

func WithCloseTimeout

func WithCloseTimeout(timeout time.Duration) Option

WithCloseTimeout sets close timeout to config.

func WithConnectTimeout

func WithConnectTimeout(timeout time.Duration) Option

WithConnectTimeout sets connect timeout to config.

func WithMaxConnections

func WithMaxConnections(maxConnections uint32) Option

WithMaxConnections sets max connections to config.

func WithName

func WithName(name string) Option

WithName sets name to config.

func WithOnConnected

func WithOnConnected(onConnected func(clientAddress string, serverAddress string)) Option

WithOnConnected sets on connected function to config.

func WithOnDisconnected

func WithOnDisconnected(onDisconnected func(clientAddress string, serverAddress string)) Option

WithOnDisconnected sets on disconnected function to config.

func WithReadBufferSize

func WithReadBufferSize(bufferSize uint32) Option

WithReadBufferSize sets read buffer size to config.

func WithReadTimeout

func WithReadTimeout(timeout time.Duration) Option

WithReadTimeout sets read timeout to config.

func WithWriteBufferSize

func WithWriteBufferSize(bufferSize uint32) Option

WithWriteBufferSize sets write buffer size to config.

func WithWriteTimeout

func WithWriteTimeout(timeout time.Duration) Option

WithWriteTimeout sets write timeout to config.

func (Option) ApplyTo

func (o Option) ApplyTo(conf *Config)

type Server

type Server interface {
	io.Closer

	Serve() error
	Status() Status
}

func NewServer

func NewServer(address string, handle HandleFunc, opts ...Option) Server

NewServer creates a new server serving on address. Handler is an interface of handling a connection.

type Status

type Status struct {
	// Connected is the count of connected connections.
	Connected uint64 `json:"connected"`
}