arpc

package module
v1.1.0 Latest Latest
Warning

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

 Go to latest Published: Dec 15, 2020 License: MIT Imports: 14 Imported by: 16

README

ARPC - More Effective Network Communication

Mentioned in Awesome Go

GoDoc 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

	// MaxBodyLen limits Message body length.
	MaxBodyLen 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 (
	// 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 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 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 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 RecvBufferSize 

func RecvBufferSize() int

RecvBufferSize returns default client's read buffer size.

func SendQueueSize 

func SendQueueSize() int

SendQueueSize returns default client's send queue channel capacity.

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 SetBufferFactory 

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

SetBufferFactory registers default buffer maker.

func SetHandler 

func SetHandler(h Handler)

SetHandler sets default Handler.

func SetLogTag 

func SetLogTag(tag string)

SetLogTag sets DefaultHandler's log tag.

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 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) (*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{}) *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) (*ClientPool, error)

NewClientPool creates a ClientPool.

func NewClientPoolFromDialers 

func NewClientPoolFromDialers(dialers []DialerFunc) (*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) 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)

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

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

	// 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 read buffer size.
	RecvBufferSize() int
	// SetRecvBufferSize sets client's read buffer size.
	SetRecvBufferSize(size int)

	// SendQueueSize returns client's send queue channel capacity.
	SendQueueSize() int
	// SetSendQueueSize sets client's send queue channel capacity.
	SetSendQueueSize(size 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)

	// GetBuffer makes a buffer by size.
	GetBuffer(size int) []byte

	// SetBufferFactory registers buffer maker.
	SetBufferFactory(f func(int) []byte)
}

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 [4]byte

Header defines Message head

func (Header) BodyLen 

func (h Header) BodyLen() int

BodyLen returns Message body length.

type Message 

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

Message represents an arpc 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) 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 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) NewMessage 

func (s *Server) NewMessage(cmd byte, method string, v 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
broadcast/client
broadcast/server
graceful/client
graceful/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
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
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.