arpc

package module
v1.2.2 Latest Latest
Warning

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

 Go to latest Published: Sep 17, 2021 License: MIT Imports: 15 Imported by: 10

README

ARPC - More Effective Network Communication

Slack

Mentioned in Awesome Go MIT licensed Build Status Go Report Card Coverage Statusd

Contents

Features

  • Two-Way Calling
  • Two-Way Notify
  • Sync and Async Calling
  • Sync and Async Response
  • Batch Write | Writev | net.Buffers
  • Broadcast
  • Middleware
  • Pub/Sub
  • Opentracing
Pattern Interactive Directions Description
call two-way:
c -> s
s -> c		 request and response
notify two-way:
c -> s
s -> c		 request without response

Performance

  • simple echo load testing
Framework Protocol Codec Configuration Connection Num Number of Goroutines Per Connection Qps
arpc tcp/localhost encoding/json os: VMWare Ubuntu 18.04
cpu: AMD 3500U 4c8t
mem: 2G		 8 10 80-100k
grpc http2/localhost protobuf os: VMWare Ubuntu 18.04
cpu: AMD 3500U 4c8t
mem: 2G		 8 10 20-30k

Header Layout

  • LittleEndian
bodyLen reserved cmd flag methodLen sequence method body
4 bytes 1 byte 1 byte 1 bytes 1 bytes 8 bytes methodLen bytes bodyLen-methodLen bytes

Installation

  1. Get and install arpc
$ go get -u github.com/lesismal/arpc
  1. Import in your code:
import "github.com/lesismal/arpc"

Quick Start

package main

import "github.com/lesismal/arpc"

func main() {
	server := arpc.NewServer()

	// register router
	server.Handler.Handle("/echo", func(ctx *arpc.Context) {
		str := ""
		if err := ctx.Bind(&str); err == nil {
			ctx.Write(str)
		}
	})

	server.Run("localhost:8888")
}

package main

import (
	"log"
	"net"
	"time"

	"github.com/lesismal/arpc"
)

func main() {
	client, err := arpc.NewClient(func() (net.Conn, error) {
		return net.DialTimeout("tcp", "localhost:8888", time.Second*3)
	})
	if err != nil {
		panic(err)
	}
	defer client.Stop()

	req := "hello"
	rsp := ""
	err = client.Call("/echo", &req, &rsp, time.Second*5)
	if err != nil {
		log.Fatalf("Call failed: %v", err)
	} else {
		log.Printf("Call Response: \"%v\"", rsp)
	}
}

API Examples

Register Routers
var handler arpc.Handler

// package
handler = arpc.DefaultHandler
// server
handler = server.Handler
// client
handler = client.Handler

// message would be default handled one by one  in the same conn reader goroutine
handler.Handle("/route", func(ctx *arpc.Context) { ... })
handler.Handle("/route2", func(ctx *arpc.Context) { ... })

// this make message handled by a new goroutine
async := true
handler.Handle("/asyncResponse", func(ctx *arpc.Context) { ... }, async)
Router Middleware

See router middleware, it's easy to implement middlewares yourself

import "github.com/lesismal/arpc/extension/middleware/router"

var handler arpc.Handler

// package
handler = arpc.DefaultHandler
// server
handler = server.Handler
// client
handler = client.Handler

handler.Use(router.Recover())
handler.Use(router.Logger())
handler.Use(func(ctx *arpc.Context) { ... })
handler.Handle("/echo", func(ctx *arpc.Context) { ... })
handler.Use(func(ctx *arpc.Context) { ... })
Coder Middleware
  • Coder Middleware is used for converting a message data to your designed format, e.g encrypt/decrypt and compress/uncompress
import "github.com/lesismal/arpc/extension/middleware/coder/gzip"

var handler arpc.Handler

// package
handler = arpc.DefaultHandler
// server
handler = server.Handler
// client
handler = client.Handler

handler.UseCoder(gzip.New())
handler.Handle("/echo", func(ctx *arpc.Context) { ... })
Client Call, CallAsync, Notify
  1. Call (Block, with timeout/context)
request := &Echo{...}
response := &Echo{}
timeout := time.Second*5
err := client.Call("/call/echo", request, response, timeout)
// ctx, cancel := context.WithTimeout(context.Background(), time.Second)
// defer cancel()
// err := client.CallWith(ctx, "/call/echo", request, response)
  1. CallAsync (Nonblock, with callback and timeout/context)
request := &Echo{...}

timeout := time.Second*5
err := client.CallAsync("/call/echo", request, func(ctx *arpc.Context) {
	response := &Echo{}
	ctx.Bind(response)
	...	
}, timeout)
  1. Notify (same as CallAsync with timeout/context, without callback)
data := &Notify{...}
client.Notify("/notify", data, time.Second)
// ctx, cancel := context.WithTimeout(context.Background(), time.Second)
// defer cancel()
// client.NotifyWith(ctx, "/notify", data)
Server Call, CallAsync, Notify
  1. Get client and keep it in your application
var client *arpc.Client
server.Handler.Handle("/route", func(ctx *arpc.Context) {
	client = ctx.Client
	// release client
	client.OnDisconnected(func(c *arpc.Client){
		client = nil
	})
})

go func() {
	for {
		time.Sleep(time.Second)
		if client != nil {
			client.Call(...)
			client.CallAsync(...)
			client.Notify(...)
		}
	}
}()
  1. Then Call/CallAsync/Notify
Broadcast - Notify
var mux = sync.RWMutex{}
var clientMap = make(map[*arpc.Client]struct{})

func broadcast() {
	var svr *arpc.Server = ... 
	msg := svr.NewMessage(arpc.CmdNotify, "/broadcast", fmt.Sprintf("broadcast msg %d", i))
	mux.RLock()
	for client := range clientMap {
		client.PushMsg(msg, arpc.TimeZero)
	}
	mux.RUnlock()
}
Async Response
var handler arpc.Handler

// package
handler = arpc.DefaultHandler
// server
handler = server.Handler
// client
handler = client.Handler

asyncResponse := true // default is true, or set false
handler.Handle("/echo", func(ctx *arpc.Context) {
	req := ...
	err := ctx.Bind(req)
	if err == nil {
		ctx.Write(data)
	}
}, asyncResponse)
Handle New Connection
// package
arpc.DefaultHandler.HandleConnected(func(c *arpc.Client) {
	...
})

// server
svr := arpc.NewServer()
svr.Handler.HandleConnected(func(c *arpc.Client) {
	...
})

// client
client, err := arpc.NewClient(...)
client.Handler.HandleConnected(func(c *arpc.Client) {
	...
})
Handle Disconnected
// package
arpc.DefaultHandler.HandleDisconnected(func(c *arpc.Client) {
	...
})

// server
svr := arpc.NewServer()
svr.Handler.HandleDisconnected(func(c *arpc.Client) {
	...
})

// client
client, err := arpc.NewClient(...)
client.Handler.HandleDisconnected(func(c *arpc.Client) {
	...
})
Handle Client's send queue overstock
// package
arpc.DefaultHandler.HandleOverstock(func(c *arpc.Client) {
	...
})

// server
svr := arpc.NewServer()
svr.Handler.HandleOverstock(func(c *arpc.Client) {
	...
})

// client
client, err := arpc.NewClient(...)
client.Handler.HandleOverstock(func(c *arpc.Client) {
	...
})
Custom Net Protocol
// server
var ln net.Listener = ...
svr := arpc.NewServer()
svr.Serve(ln)

// client
dialer := func() (net.Conn, error) { 
	return ... 
}
client, err := arpc.NewClient(dialer)
Custom Codec
import "github.com/lesismal/arpc/codec"

var codec arpc.Codec = ...

// package
codec.Defaultcodec = codec

// server
svr := arpc.NewServer()
svr.Codec = codec

// client
client, err := arpc.NewClient(...)
client.Codec = codec
Custom Logger
import "github.com/lesismal/arpc/log"

var logger arpc.Logger = ...
log.SetLogger(logger) // log.DefaultLogger = logger
Custom operations before conn's recv and send
arpc.DefaultHandler.BeforeRecv(func(conn net.Conn) error) {
	// ...
})

arpc.DefaultHandler.BeforeSend(func(conn net.Conn) error) {
	// ...
})
Custom arpc.Client's Reader by wrapping net.Conn
arpc.DefaultHandler.SetReaderWrapper(func(conn net.Conn) io.Reader) {
	// ...
})
Custom arpc.Client's send queue capacity
arpc.DefaultHandler.SetSendQueueSize(4096)

JS Client

Web Chat Examples

Pub/Sub Examples

  • start a server
import "github.com/lesismal/arpc/extension/pubsub"

var (
	address = "localhost:8888"

	password = "123qwe"

	topicName = "Broadcast"
)

func main() {
	s := pubsub.NewServer()
	s.Password = password

	// server publish to all clients
	go func() {
		for i := 0; true; i++ {
			time.Sleep(time.Second)
			s.Publish(topicName, fmt.Sprintf("message from server %v", i))
		}
	}()

	s.Run(address)
}
  • start a subscribe client
import "github.com/lesismal/arpc/log"
import "github.com/lesismal/arpc/extension/pubsub"

var (
	address = "localhost:8888"

	password = "123qwe"

	topicName = "Broadcast"
)

func onTopic(topic *pubsub.Topic) {
	log.Info("[OnTopic] [%v] \"%v\", [%v]",
		topic.Name,
		string(topic.Data),
		time.Unix(topic.Timestamp/1000000000, topic.Timestamp%1000000000).Format("2006-01-02 15:04:05.000"))
}

func main() {
	client, err := pubsub.NewClient(func() (net.Conn, error) {
		return net.DialTimeout("tcp", address, time.Second*3)
	})
	if err != nil {
		panic(err)
	}
	client.Password = password

	// authentication
	err = client.Authenticate()
	if err != nil {
		panic(err)
	}

	// subscribe topic
	if err := client.Subscribe(topicName, onTopic, time.Second); err != nil {
		panic(err)
	}

	<-make(chan int)
}
  • start a publish client
import "github.com/lesismal/arpc/extension/pubsub"

var (
	address = "localhost:8888"

	password = "123qwe"

	topicName = "Broadcast"
)

func main() {
	client, err := pubsub.NewClient(func() (net.Conn, error) {
		return net.DialTimeout("tcp", address, time.Second*3)
	})
	if err != nil {
		panic(err)
	}
	client.Password = password

	// authentication
	err = client.Authenticate()
	if err != nil {
		panic(err)
	}

	for i := 0; true; i++ {
		if i%5 == 0 {
			// publish msg to all clients
			client.Publish(topicName, fmt.Sprintf("message from client %d", i), time.Second)
		} else {
			// publish msg to only one client
			client.PublishToOne(topicName, fmt.Sprintf("message from client %d", i), time.Second)
		}
		time.Sleep(time.Second)
	}
}

More Examples

Documentation

Index

Constants

View Source 
const (
	// TimeZero represents zero time.
	TimeZero time.Duration = 0
	// TimeForever represents forever time.
	TimeForever time.Duration = 1<<63 - 1
)
View Source 
const (
	// CmdNone is invalid
	CmdNone byte = 0

	// CmdRequest the other side should response to a request message
	CmdRequest byte = 1

	// CmdResponse the other side should not response to a request message
	CmdResponse byte = 2

	// CmdNotify the other side should not response to a request message
	CmdNotify byte = 3
)
View Source 
const (
	// HeaderIndexBodyLenBegin .
	HeaderIndexBodyLenBegin = 0
	// HeaderIndexBodyLenEnd .
	HeaderIndexBodyLenEnd = 4
	// HeaderIndexReserved .
	HeaderIndexReserved = 4
	// HeaderIndexCmd .
	HeaderIndexCmd = 5
	// HeaderIndexFlag .
	HeaderIndexFlag = 6
	// HeaderIndexMethodLen .
	HeaderIndexMethodLen = 7
	// HeaderIndexSeqBegin .
	HeaderIndexSeqBegin = 8
	// HeaderIndexSeqEnd .
	HeaderIndexSeqEnd = 16
	// HeaderFlagMaskError .
	HeaderFlagMaskError byte = 0x01
	// HeaderFlagMaskAsync .
	HeaderFlagMaskAsync byte = 0x02
)
View Source 
const (
	// HeadLen represents Message head length.
	HeadLen int = 16

	// MaxMethodLen limits Message method length.
	MaxMethodLen int = 127

	// DefaultMaxBodyLen limits Message body length.
	DefaultMaxBodyLen int = 1024*1024*64 - 16
)

Variables

View Source 
var (
	// ErrClientTimeout represents a timeout error because of timer or context.
	ErrClientTimeout = errors.New("timeout")

	// ErrClientInvalidTimeoutZero represents an error of 0 time parameter.
	ErrClientInvalidTimeoutZero = errors.New("invalid timeout, should not be 0")

	// ErrClientInvalidTimeoutLessThanZero represents an error of less than 0 time parameter.
	ErrClientInvalidTimeoutLessThanZero = errors.New("invalid timeout, should not be < 0")

	// ErrClientInvalidTimeoutZeroWithNonNilCallback represents an error with 0 time parameter but with non-nil callback.
	ErrClientInvalidTimeoutZeroWithNonNilCallback = errors.New("invalid timeout 0 with non-nil callback")

	// ErrClientOverstock represents an error of Client's send queue is full.
	ErrClientOverstock = errors.New("timeout: rpc Client's send queue is full")

	// ErrClientReconnecting represents an error that Client is reconnecting.
	ErrClientReconnecting = errors.New("client reconnecting")

	// ErrClientStopped represents an error that Client is stopped.
	ErrClientStopped = errors.New("client stopped")

	// ErrClientInvalidPoolDialers represents an error of empty dialer array.
	ErrClientInvalidPoolDialers = errors.New("invalid dialers: empty array")
)

client error

View Source 
var (
	// ErrInvalidRspMessage represents an error of invalid message CMD.
	ErrInvalidRspMessage = errors.New("invalid response message cmd")

	// ErrMethodNotFound represents an error of method not found.
	ErrMethodNotFound = errors.New("method not found")

	// ErrInvalidFlagBitIndex represents an error of invlaid flag bit index.
	ErrInvalidFlagBitIndex = errors.New("invalid index, should be 0-7")
)

message error

View Source 
var BufferPool interface {
	Malloc(size int) []byte
	Realloc(buf []byte, size int) []byte
	Free(buf []byte)
} = New(64)

BufferPool is the default buffer pool instance.

View Source 
var (
	// ErrContextResponseToNotify represents an error that response to a notify message.
	ErrContextResponseToNotify = errors.New("should not response to a context with notify message")
)

context error

View Source 
var (
	// ErrTimeout represents an error of timeout.
	ErrTimeout = errors.New("timeout")
)

general errors

Functions

func AsyncExecute added in v1.1.10 

func AsyncExecute(f func())

AsyncExecute executes a func.

func AsyncResponse 

func AsyncResponse() bool

AsyncResponse returns default AsyncResponse flag.

func BatchRecv 

func BatchRecv() bool

BatchRecv returns default BatchRecv flag.

func BatchSend 

func BatchSend() bool

BatchSend returns default BatchSend flag.

func BeforeRecv 

func BeforeRecv(h func(net.Conn) error)

BeforeRecv registers default handler which will be called before Recv.

func BeforeSend 

func BeforeSend(h func(net.Conn) error)

BeforeSend registers default handler which will be called before Send.

func EnablePool added in v1.1.9 

func EnablePool(enable bool)

EnablePool registers handlers for pool operation for Context and Message and Message.Buffer

func Free added in v1.1.9 

func Free(buf []byte)

Free exports default package method

func Handle 

func Handle(m string, h HandlerFunc, args ...interface{})

Handle registers default method/router handler.

If pass a Boolean value of "true", the handler will be called asynchronously in a new goroutine, Else the handler will be called synchronously in the client's reading goroutine one by one.

func HandleConnected 

func HandleConnected(onConnected func(*Client))

HandleConnected registers default handler which will be called when client connected.

func HandleDisconnected 

func HandleDisconnected(onDisConnected func(*Client))

HandleDisconnected registers default handler which will be called when client disconnected.

func HandleMalloc added in v1.1.9 

func HandleMalloc(f func(int) []byte)

HandleMalloc registers default buffer maker.

func HandleMessageDropped 

func HandleMessageDropped(onOverstock func(c *Client, m *Message))

HandleMessageDropped registers default handler which will be called when message dropped.

func HandleNotFound 

func HandleNotFound(h HandlerFunc)

HandleNotFound registers default "" method/router handler, It will be called when mothod/router is not found.

func HandleOverstock 

func HandleOverstock(onOverstock func(c *Client, m *Message))

HandleOverstock registers default handler which will be called when client send queue is overstock.

func HandleSessionMiss 

func HandleSessionMiss(onSessionMiss func(c *Client, m *Message))

HandleSessionMiss registers default handler which will be called when async message seq not found.

func Malloc added in v1.1.9 

func Malloc(size int) []byte

Malloc exports default package method

func MaxBodyLen 

func MaxBodyLen() int

func Realloc added in v1.1.9 

func Realloc(buf []byte, size int) []byte

Realloc exports default package method

func RecvBufferSize 

func RecvBufferSize() int

RecvBufferSize returns default client's read buffer size.

func SendBufferSize added in v1.1.9 

func SendBufferSize() int

SendBufferSize returns default client's read buffer size.

func SendQueueSize 

func SendQueueSize() int

SendQueueSize returns default client's send queue channel capacity.

func SetAsyncExecutor added in v1.1.10 

func SetAsyncExecutor(executor func(f func()))

SetAsyncExecutor sets executor. AsyncExecute executes a func

func SetAsyncResponse 

func SetAsyncResponse(async bool)

SetAsyncResponse sets default AsyncResponse flag.

func SetBatchRecv 

func SetBatchRecv(batch bool)

SetBatchRecv sets default BatchRecv flag.

func SetBatchSend 

func SetBatchSend(batch bool)

SetBatchSend sets default BatchSend flag.

func SetHandler 

func SetHandler(h Handler)

SetHandler sets default Handler.

func SetLogTag 

func SetLogTag(tag string)

SetLogTag sets DefaultHandler's log tag.

func SetMaxBodyLen added in v1.1.11 

func SetMaxBodyLen(l int)

func SetReaderWrapper 

func SetReaderWrapper(wrapper func(conn net.Conn) io.Reader)

SetReaderWrapper registers default reader wrapper for net.Conn.

func SetRecvBufferSize 

func SetRecvBufferSize(size int)

SetRecvBufferSize sets default client's read buffer size.

func SetSendBufferSize added in v1.1.9 

func SetSendBufferSize(size int)

SetSendBufferSize sets default client's read buffer size.

func SetSendQueueSize 

func SetSendQueueSize(size int)

SetSendQueueSize sets default client's send queue channel capacity.

func Use 

func Use(h HandlerFunc)

Use registers default method/router handler middleware.

func UseCoder 

func UseCoder(coder MessageCoder)

UseCoder registers default message coding middleware, coder.Encode will be called before message send, coder.Decode will be called after message recv.

Types

type Client 

type Client struct {
	Conn    net.Conn
	Codec   codec.Codec
	Handler Handler
	Reader  io.Reader
	Dialer  DialerFunc
	Head    Header
	// contains filtered or unexported fields
}

Client represents an arpc Client. There may be multiple outstanding Calls or Notifys associated with a single Client, and a Client may be used by multiple goroutines simultaneously.

func NewClient 

func NewClient(dialer DialerFunc, args ...interface{}) (*Client, error)

NewClient creates a Client.

func (*Client) Call 

func (c *Client) Call(method string, req interface{}, rsp interface{}, timeout time.Duration, args ...interface{}) error

Call makes an rpc call with a timeout. Call will block waiting for the server's response until timeout.

func (*Client) CallAsync 

func (c *Client) CallAsync(method string, req interface{}, handler HandlerFunc, timeout time.Duration, args ...interface{}) error

CallAsync makes an asynchronous rpc call with timeout. CallAsync will not block waiting for the server's response, But the handler will be called if the response arrives before the timeout.

func (*Client) CallWith 

func (c *Client) CallWith(ctx context.Context, method string, req interface{}, rsp interface{}, args ...interface{}) error

CallWith uses context to make rpc call. CallWith blocks to wait for a response from the server until it times out.

func (*Client) CheckState 

func (c *Client) CheckState() error

CheckState checks Client's state.

func (*Client) Delete 

func (c *Client) Delete(key interface{})

Delete deletes key-value pair

func (*Client) Get 

func (c *Client) Get(key interface{}) (interface{}, bool)

Get returns value for key.

func (*Client) NewMessage 

func (c *Client) NewMessage(cmd byte, method string, v interface{}, args ...interface{}) *Message

NewMessage creates a Message by client's seq, handler and codec.

func (*Client) Notify 

func (c *Client) Notify(method string, data interface{}, timeout time.Duration, args ...interface{}) error

Notify makes a notify with timeout. A notify does not need a response from the server.

func (*Client) NotifyWith 

func (c *Client) NotifyWith(ctx context.Context, method string, data interface{}, args ...interface{}) error

NotifyWith use context to make rpc notify. A notify does not need a response from the server.

func (*Client) PushMsg 

func (c *Client) PushMsg(msg *Message, timeout time.Duration) error

PushMsg pushes a msg to Client's send queue with timeout.

func (*Client) Restart 

func (c *Client) Restart() error

Restart stops and restarts a Client.

func (*Client) Set 

func (c *Client) Set(key interface{}, value interface{})

Set sets key-value pair.

func (*Client) Stop 

func (c *Client) Stop()

Stop stops a Client.

type ClientPool 

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

ClientPool represents an arpc Client Pool.

func NewClientPool 

func NewClientPool(dialer DialerFunc, size int, args ...interface{}) (*ClientPool, error)

NewClientPool creates a ClientPool.

func NewClientPoolFromDialers 

func NewClientPoolFromDialers(dialers []DialerFunc, args ...interface{}) (*ClientPool, error)

NewClientPoolFromDialers creates a ClientPool by multiple dialers.

func (*ClientPool) Get 

func (pool *ClientPool) Get(index int) *Client

Get returns a Client by index.

func (*ClientPool) Handler 

func (pool *ClientPool) Handler() Handler

Handler returns Handler.

func (*ClientPool) Next 

func (pool *ClientPool) Next() *Client

Next returns a Client by round robin.

func (*ClientPool) Size 

func (pool *ClientPool) Size() int

Size returns Client number.

func (*ClientPool) Stop 

func (pool *ClientPool) Stop()

Stop stops all clients.

type Context 

type Context struct {
	Client  *Client
	Message *Message
	// contains filtered or unexported fields
}

Context represents an arpc Call's context.

func (*Context) Abort added in v1.1.0 

func (ctx *Context) Abort()

Abort stops the one-by-one-calling of middlewares and method/router handler.

func (*Context) Bind 

func (ctx *Context) Bind(v interface{}) error

Bind parses the body data and stores the result in the value pointed to by v.

func (*Context) Body 

func (ctx *Context) Body() []byte

Body returns body.

func (*Context) Deadline added in v1.1.0 

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

Deadline implements stdlib's Context.

func (*Context) Done 

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

Done implements stdlib's Context.

func (*Context) Err added in v1.1.0 

func (ctx *Context) Err() error

Err implements stdlib's Context.

func (*Context) Error 

func (ctx *Context) Error(v interface{}) error

Error responses an error Message to the Client.

func (*Context) Get 

func (ctx *Context) Get(key interface{}) (interface{}, bool)

Get returns value for key.

func (*Context) Next 

func (ctx *Context) Next()

Next calls next middleware or method/router handler.

func (*Context) Release added in v1.1.9 

func (ctx *Context) Release()

func (*Context) Set 

func (ctx *Context) Set(key interface{}, value interface{})

Set sets key-value pair.

func (*Context) Value added in v1.1.0 

func (ctx *Context) Value(key interface{}) interface{}

Value returns the value associated with this context for key, implements stdlib's Context.

func (*Context) Values 

func (ctx *Context) Values() map[interface{}]interface{}

Values returns values.

func (*Context) Write 

func (ctx *Context) Write(v interface{}) error

Write responses a Message to the Client.

func (*Context) WriteWithTimeout 

func (ctx *Context) WriteWithTimeout(v interface{}, timeout time.Duration) error

WriteWithTimeout responses a Message to the Client with timeout.

type DialerFunc 

type DialerFunc func() (net.Conn, error)

DialerFunc defines the dialer used by arpc Client to connect to the server.

type Handler 

type Handler interface {
	// Clone returns a copy of Handler.
	Clone() Handler

	// LogTag returns log tag value.
	LogTag() string
	// SetLogTag sets log tag.
	SetLogTag(tag string)

	// HandleConnected registers handler which will be called when client connected.
	HandleConnected(onConnected func(*Client))
	// OnConnected will be called when client is connected.
	OnConnected(c *Client)

	// HandleDisconnected registers handler which will be called when client is disconnected.
	HandleDisconnected(onDisConnected func(*Client))
	// OnDisconnected will be called when client is disconnected.
	OnDisconnected(c *Client)

	// HandleOverstock registers handler which will be called when client send queue is overstock.
	HandleOverstock(onOverstock func(c *Client, m *Message))
	// OnOverstock will be called when client chSend is full.
	OnOverstock(c *Client, m *Message)

	// HandleMessageDone registers handler which will be called when message dropped.
	HandleMessageDone(onMessageDone func(c *Client, m *Message))
	// OnMessageDone will be called when message is dropped.
	OnMessageDone(c *Client, m *Message)

	// HandleMessageDropped registers handler which will be called when message dropped.
	HandleMessageDropped(onOverstock func(c *Client, m *Message))
	// OnOverstock will be called when message is dropped.
	OnMessageDropped(c *Client, m *Message)

	// HandleSessionMiss registers handler which will be called when async message seq not found.
	HandleSessionMiss(onSessionMiss func(c *Client, m *Message))
	// OnSessionMiss will be called when async message seq not found.
	OnSessionMiss(c *Client, m *Message)

	// HandleContextDone registers handler which will be called when message dropped.
	HandleContextDone(onContextDone func(ctx *Context))
	// OnContextDone will be called when message is dropped.
	OnContextDone(ctx *Context)

	// BeforeRecv registers handler which will be called before Recv.
	BeforeRecv(h func(net.Conn) error)
	// BeforeSend registers handler which will be called before Send.
	BeforeSend(h func(net.Conn) error)

	// BatchRecv returns BatchRecv flag.
	BatchRecv() bool
	// SetBatchRecv sets BatchRecv flag.
	SetBatchRecv(batch bool)
	// BatchSend returns BatchSend flag.
	BatchSend() bool
	// SetBatchSend sets BatchSend flag.
	SetBatchSend(batch bool)

	// AsyncWrite returns AsyncWrite flag.
	AsyncWrite() bool
	// SetAsyncWrite sets AsyncWrite flag.
	SetAsyncWrite(async bool)

	// AsyncResponse returns AsyncResponse flag.
	AsyncResponse() bool
	// SetAsyncResponse sets AsyncResponse flag.
	SetAsyncResponse(async bool)

	// WrapReader wraps net.Conn to Read data with io.Reader.
	WrapReader(conn net.Conn) io.Reader
	// SetReaderWrapper registers reader wrapper for net.Conn.
	SetReaderWrapper(wrapper func(conn net.Conn) io.Reader)

	// Recv reads a message from a client.
	Recv(c *Client) (*Message, error)
	// Send writes buffer data to a connection.
	Send(c net.Conn, buffer []byte) (int, error)
	// SendN writes multiple buffer data to a connection.
	SendN(conn net.Conn, buffers net.Buffers) (int, error)

	// RecvBufferSize returns client's recv buffer size.
	RecvBufferSize() int
	// SetRecvBufferSize sets client's recv buffer size.
	SetRecvBufferSize(size int)

	// SendBufferSize returns client's send buffer size.
	SendBufferSize() int
	// SetSendBufferSize sets client's send buffer size.
	SetSendBufferSize(size int)

	// SendQueueSize returns client's send queue channel capacity.
	SendQueueSize() int
	// SetSendQueueSize sets client's send queue channel capacity.
	SetSendQueueSize(size int)

	// MaxBodyLen returns max body length of a message.
	MaxBodyLen() int
	// SetMaxBodyLen sets max body length of a message.
	SetMaxBodyLen(l int)

	// Use registers method/router handler middleware.
	Use(h HandlerFunc)

	// UseCoder registers message coding middleware,
	// coder.Encode will be called before message send,
	// coder.Decode will be called after message recv.
	UseCoder(coder MessageCoder)

	// Coders returns coding middlewares.
	Coders() []MessageCoder

	// Handle registers method/router handler.
	//
	// If pass a Boolean value of "true", the handler will be called asynchronously in a new goroutine,
	// Else the handler will be called synchronously in the client's reading goroutine one by one.
	Handle(m string, h HandlerFunc, args ...interface{})

	// HandleNotFound registers "" method/router handler,
	// It will be called when mothod/router is not found.
	HandleNotFound(h HandlerFunc)

	// OnMessage finds method/router middlewares and handler, then call them one by one.
	OnMessage(c *Client, m *Message)

	// Malloc makes a buffer by size.
	Malloc(size int) []byte
	// HandleMalloc registers buffer maker.
	HandleMalloc(f func(size int) []byte)

	// Free release a buffer.
	Free([]byte)
	// HandleFree registers buffer releaser.
	HandleFree(f func(buf []byte))

	// EnablePool registers handlers for pool operation for Context and Message and Message.Buffer
	EnablePool(enable bool)

	Context() (context.Context, context.CancelFunc)
	SetContext(ctx context.Context, cancel context.CancelFunc)
	Cancel()

	// NewMessage creates a Message.
	NewMessage(cmd byte, method string, v interface{}, isError bool, isAsync bool, seq uint64, codec codec.Codec, values map[interface{}]interface{}) *Message

	// NewMessageWithBuffer creates a message with the buffer and manage the message by the pool.
	// The buffer arg should be managed by a pool if EnablePool(true) .
	NewMessageWithBuffer(buffer []byte) *Message

	// SetAsyncExecutor sets executor.
	SetAsyncExecutor(executor func(f func()))
	// AsyncExecute executes a func
	AsyncExecute(f func())
}

Handler defines net message handler interface. 

var DefaultHandler Handler = NewHandler()

DefaultHandler is the default Handler used by arpc

func NewHandler 

func NewHandler() Handler

NewHandler returns a default Handler implementation.

type HandlerFunc 

type HandlerFunc func(*Context)

HandlerFunc defines message handler of arpc middleware and method/router. 

type Header []byte

Header defines Message head

func (Header) BodyLen 

func (h Header) BodyLen() int

BodyLen returns Message body length.

type MemPool added in v1.1.9 

type MemPool struct {
	Debug bool
	// contains filtered or unexported fields
}

MemPool

func New added in v1.1.9 

func New(minSize int) *MemPool

func (*MemPool) Free added in v1.1.9 

func (mp *MemPool) Free(buf []byte)

Free .

func (*MemPool) Malloc added in v1.1.9 

func (mp *MemPool) Malloc(size int) []byte

func (*MemPool) Realloc added in v1.1.9 

func (mp *MemPool) Realloc(buf []byte, size int) []byte

Realloc .

type Message 

type Message struct {
	Buffer []byte
	// contains filtered or unexported fields
}

Message represents an arpc Message.

func NewMessage added in v1.1.9 

func NewMessage(cmd byte, method string, v interface{}, isError bool, isAsync bool, seq uint64, h Handler, codec codec.Codec, values map[interface{}]interface{}) *Message

NewMessage creates a Message.

func (*Message) BodyLen 

func (m *Message) BodyLen() int

BodyLen returns body length.

func (*Message) Cmd 

func (m *Message) Cmd() byte

Cmd returns cmd.

func (*Message) Data 

func (m *Message) Data() []byte

Data returns payload data after method.

func (*Message) Error 

func (m *Message) Error() error

Error returns error.

func (*Message) Get 

func (m *Message) Get(key interface{}) (interface{}, bool)

Get returns value for key.

func (*Message) IsAsync 

func (m *Message) IsAsync() bool

IsAsync returns async flag.

func (*Message) IsError 

func (m *Message) IsError() bool

IsError returns error flag.

func (*Message) IsFlagBitSet 

func (m *Message) IsFlagBitSet(index int) bool

IsFlagBitSet returns flag bit value.

func (*Message) Len 

func (m *Message) Len() int

Len returns total length of buffer.

func (*Message) Method 

func (m *Message) Method() string

Method returns method.

func (*Message) MethodLen 

func (m *Message) MethodLen() int

MethodLen returns method length.

func (*Message) Payback added in v1.1.9 

func (m *Message) Payback()

Payback put Message to the pool.

func (*Message) Release added in v1.1.9 

func (m *Message) Release() int32

Release decrement the reference count and returns the current value.

func (*Message) ResetAttrs added in v1.1.9 

func (m *Message) ResetAttrs()

ResetAttrs resets reserved/cmd/flag/methodLen to 0.

func (*Message) Retain added in v1.1.9 

func (m *Message) Retain() int32

Retain increment the reference count and returns the current value.

func (*Message) Seq 

func (m *Message) Seq() uint64

Seq returns sequence number.

func (*Message) Set 

func (m *Message) Set(key interface{}, value interface{})

Set sets key-value pair.

func (*Message) SetAsync 

func (m *Message) SetAsync(isAsync bool)

SetAsync sets async flag.

func (*Message) SetBodyLen 

func (m *Message) SetBodyLen(l int)

SetBodyLen sets body length.

func (*Message) SetCmd 

func (m *Message) SetCmd(cmd byte)

SetCmd sets cmd.

func (*Message) SetError 

func (m *Message) SetError(isError bool)

SetError sets error flag.

func (*Message) SetFlagBit 

func (m *Message) SetFlagBit(index int, value bool) error

SetFlagBit sets flag bit value by index.

func (*Message) SetMethodLen 

func (m *Message) SetMethodLen(l int)

SetMethodLen sets method length.

func (*Message) SetSeq 

func (m *Message) SetSeq(seq uint64)

SetSeq sets sequence number.

func (*Message) Values 

func (m *Message) Values() map[interface{}]interface{}

Values returns values.

type MessageCoder 

type MessageCoder interface {
	// Encode wrap message before send to client
	Encode(*Client, *Message) *Message
	// Decode unwrap message between recv and handle
	Decode(*Client, *Message) *Message
}

MessageCoder defines Message coding middleware interface.

type NativeAllocator added in v1.1.9 

type NativeAllocator struct{}

NativeAllocator definition

func (*NativeAllocator) Free added in v1.1.9 

func (a *NativeAllocator) Free(buf []byte)

Free .

func (*NativeAllocator) Malloc added in v1.1.9 

func (a *NativeAllocator) Malloc(size int) []byte

Malloc .

func (*NativeAllocator) Realloc added in v1.1.9 

func (a *NativeAllocator) Realloc(buf []byte, size int) []byte

Realloc .

type Server 

type Server struct {
	Accepted int64
	CurrLoad int64
	MaxLoad  int64

	Codec   codec.Codec
	Handler Handler

	Listener net.Listener
	// contains filtered or unexported fields
}

Server represents an arpc Server.

func NewServer 

func NewServer() *Server

NewServer creates an arpc Server.

func (*Server) Broadcast added in v1.2.2 

func (s *Server) Broadcast(method string, v interface{}, args ...interface{})

func (*Server) BroadcastWithFilter added in v1.2.2 

func (s *Server) BroadcastWithFilter(method string, v interface{}, filter func(*Client) bool, args ...interface{})

func (*Server) ForEach added in v1.2.2 

func (s *Server) ForEach(h func(*Client))

func (*Server) ForEachWithFilter added in v1.2.2 

func (s *Server) ForEachWithFilter(h func(*Client), filter func(*Client) bool)

func (*Server) NewMessage 

func (s *Server) NewMessage(cmd byte, method string, v interface{}, args ...interface{}) *Message

NewMessage creates a Message.

func (*Server) Run 

func (s *Server) Run(addr string) error

Run starts tcp service on addr.

func (*Server) Serve 

func (s *Server) Serve(ln net.Listener) error

Serve starts service with listener.

func (*Server) Shutdown 

func (s *Server) Shutdown(ctx context.Context) error

Shutdown shutdown service.

func (*Server) Stop 

func (s *Server) Stop() error

Stop stops service.

type WebsocketConn 

type WebsocketConn interface {
	HandleWebsocket(func())
}

WebsocketConn defines websocket-conn interface.

Source Files

View all Source files

Directories

Path Synopsis
examples
bench/client
bench/server
bench_nbio/client
bench_nbio/server
bench_pool/client
bench_pool/server
broadcast/client
broadcast/server
broadcast_pool/client
broadcast_pool/server
graceful/client
graceful/server
mempool_fuzzy/client
mempool_fuzzy/server
micro/client
micro/server
middleware/coder/gzip/client
middleware/coder/gzip/server
middleware/coder/tracing/client
middleware/coder/tracing/server
middleware/router/client
middleware/router/server
nbio/client
nbio/server
notify/client
notify/server
pool/client
pool/server
protocols/kcp/client
protocols/kcp/server
protocols/quic/client
protocols/quic/server
protocols/tls/client
protocols/tls/server
protocols/unixsocket/client
protocols/unixsocket/server
protocols/utp/client
protocols/utp/server
protocols/websocket/client
protocols/websocket/server
pubsub/client
pubsub/server
rpc/client
rpc/server
webchat
extension
listener
micro
micro/etcd
middleware/coder
middleware/coder/gzip
middleware/coder/tracing
middleware/router
protocol/quic
protocol/websocket
pubsub

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.