rpcx

package module
v2.0.0-beta.7.0...-56585fc Latest Latest
Warning

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

Go to latest
Published: Dec 19, 2019 License: MIT Imports: 12 Imported by: 1

README

rpcx

Easy to use and developer friendly RPC library

go report license Maintenance PRs Welcome Ask Me Anything !

Difference Between Other RPC Libraries

  • Once Callable established, local and remote are both in Full duplex mode.
  • No more client or server roles, Callables can call/called each others
  • Lots of Middlewares and developer can Custom their own middleware!
  • It works without any proto files , just define functions as usual.
  • Based on asynchronous I/O (liblpc) and provides synchronous call semantics
  • High Performace: event drivend, msgpack(for serialize),reuse context,goroutine pool...

Usage

# install rpcx v2
go get -u github.com/gen-iot/rpcx/v2@latest
// import 
import "github.com/gen-iot/rpcx/v2"

Overview

  • Developer friendly
  • Middlewares: dump,mq_proxy,recover,validate...

Middlewares

Name Desciption
dump dump request/response content
mq_proxy transfer requst/response
over custom message queue
recover recover panic
validate validate request/response data
req_not_nill discard request which req param is nil

Getting Started

RPC Functions Formal

Remember :param ctx,err always required

  1. Both have in&out
func Function(ctx rpcx.Context, in InType)(out OutType, err error)
  1. Only err
func Function(ctx rpcx.Context)(err error)
  1. out and err
func Function(ctx rpcx.Context) (out OutType,err error)
  1. in and err
func Function(ctx rpcx.Context, in InType)(err error)
Create Core
core, err := rpcx.New()
Close Core
err := core.Close()
Register RPC Function
core, err := rpcx.New()
std.AssertError(err, "new rpc")
core.RegFuncWithName(
    "hello", // manual specified func name
    func(ctx rpcx.Context, msg string) (string, error) {
        return "hello from server", nil
    })
Connect To Remote Core
sockAddr, err := liblpc.ResolveTcpAddr("127.0.0.1")
std.AssertError(err,"resolve addr")
callable := rpcx.NewClientStreamCallable(core, sockAddr, nil)
callable.Start()
Add Exist Conn To Core
// `fd` must be a valid file descriptor
callable := rpcx.NewConnStreamCallable(core,fd, nil)
callable.Start()
Invoke RPC Functions

Suppose there is a remote function:

func hello(ctx rpcx.Context)(string,error)

Now call it

out := new(string)
err := callable.Call3(time.Second*5, "hello", out)
std.AssertError(err, "call 'hello'")
fmt.Println("result:",*out)

Callable Calls

All Call[0-6] support Timeout And Middlewares

Function Header In Out
Call
Call0 ✅
Call1 ✅
Call2 ✅ ✅
Call3 ✅
Call4 ✅ ✅
Call5 ✅ ✅
Call6 ✅ ✅ ✅

Middleware

middlewares works like AOP concept as we know in Java.

📌Import middlewares before use

import "github.com/gen-iot/rpcx/v2/middleware"
RPC Core

📌middlewares apply on rpc core will affect whole rpc context

core, err := rpcx.New()
std.AssertError(err, "new rpc core")
core.Use(
    middleware.Recover(true), // recover 
    middleware.ValidateStruct( // validator
        middleware.ValidateInOut,
        std.NewValidator(std.LANG_EN)),
)
RPC Functions
core, err := rpcx.New()
std.AssertError(err, "new rpc")
core.RegFuncWithName("hello",
    func(ctx rpcx.Context, msg string) (string, error) {
        return "hello from server", nil
    },
    middleware.LoginRequred(),  // require logined
    middleware.MustAdmin(),       // require admin role
)

Callable

simple call

err := callable.Call("hello",
    middleware.Recover(true), // recover 
)

simple call with headers

ackHeader, err := callable.Call1("helloWithHeader",&RpcMsgHeader{
                    "key1":"value1",
                    "key2":"value2",
                },
    middleware.Recover(true), // recover 
)
fmt.Println("ack header->",ackHeader)

More Example

try examples

License

Released under the MIT License

Documentation

Index

Constants

View Source
const RpcLoopDefaultBufferSize = 1024 * 1024 * 4

Variables

View Source
var Debug = true
View Source
var ErrNeedMore = errors.New("codec want read more bytes")

Functions

This section is empty.

Types

type BaseCallable

type BaseCallable struct {
	CallableCallbacks

	liblpc.BaseUserData
	TimeWheelEntryImpl
	// contains filtered or unexported fields
}

func NewBaseCallable

func NewBaseCallable(core Core, writer WriterCloser, driven Callable) *BaseCallable

func (*BaseCallable) Call

func (this *BaseCallable) Call(timeout time.Duration, name string, mids ...MiddlewareFunc) error

func (*BaseCallable) Call0

func (this *BaseCallable) Call0(timeout time.Duration, name string, headers RpcMsgHeader, mids ...MiddlewareFunc) (ackHeader RpcMsgHeader, err error)

func (*BaseCallable) Call1

func (this *BaseCallable) Call1(timeout time.Duration, name string, in interface{}, mids ...MiddlewareFunc) error

func (*BaseCallable) Call2

func (this *BaseCallable) Call2(timeout time.Duration, name string, headers RpcMsgHeader, in interface{}, mids ...MiddlewareFunc) (ackHeader RpcMsgHeader, err error)

func (*BaseCallable) Call3

func (this *BaseCallable) Call3(timeout time.Duration, name string, out interface{}, mids ...MiddlewareFunc) error

func (*BaseCallable) Call4

func (this *BaseCallable) Call4(timeout time.Duration, name string, headers RpcMsgHeader, out interface{}, mids ...MiddlewareFunc) (ackHeader RpcMsgHeader, err error)

func (*BaseCallable) Call5

func (this *BaseCallable) Call5(timeout time.Duration, name string, in, out interface{}, mids ...MiddlewareFunc) error

func (*BaseCallable) Call6

func (this *BaseCallable) Call6(timeout time.Duration, name string, headers RpcMsgHeader, in, out interface{}, mids ...MiddlewareFunc) (ackHeader RpcMsgHeader, err error)

func (*BaseCallable) Close

func (this *BaseCallable) Close() error

func (*BaseCallable) Len

func (this *BaseCallable) Len() int

func (*BaseCallable) Perform

func (this *BaseCallable) Perform(timeout time.Duration, c Context)

func (*BaseCallable) Start

func (this *BaseCallable) Start()

func (*BaseCallable) Use

func (this *BaseCallable) Use(m ...MiddlewareFunc)

func (*BaseCallable) Writer

func (this *BaseCallable) Writer() Writer

type Callable

type Callable interface {
	liblpc.UserDataStorage

	Start()

	Call(timeout time.Duration, name string, mids ...MiddlewareFunc) error
	Call0(timeout time.Duration, name string, headers RpcMsgHeader, mids ...MiddlewareFunc) (ackHeader RpcMsgHeader, err error)

	Call1(timeout time.Duration, name string, in interface{}, mids ...MiddlewareFunc) error
	Call2(timeout time.Duration, name string, headers RpcMsgHeader, in interface{}, mids ...MiddlewareFunc) (ackHeader RpcMsgHeader, err error)

	Call3(timeout time.Duration, name string, out interface{}, mids ...MiddlewareFunc) error
	Call4(timeout time.Duration, name string, headers RpcMsgHeader, out interface{}, mids ...MiddlewareFunc) (ackHeader RpcMsgHeader, err error)

	Call5(timeout time.Duration, name string, in, out interface{}, mids ...MiddlewareFunc) error
	Call6(timeout time.Duration, name string, headers RpcMsgHeader, in, out interface{}, mids ...MiddlewareFunc) (ackHeader RpcMsgHeader, err error)

	Perform(timeout time.Duration, ctx Context)

	SetOnReady(cb CallableCallback)
	SetOnClose(cb CallableCallback)

	liblpc.BucketEntry
	BindTimeWheel(timeWheel *liblpc.TimeWheel)
	NotifyTimeWheel()

	Writer() Writer
}

func NewClientStreamCallable

func NewClientStreamCallable(core Core, addr *liblpc.SyscallSockAddr,
	userData interface{},
	m ...MiddlewareFunc) (Callable, error)

func NewConnStreamCallable

func NewConnStreamCallable(core Core, fd int, userData interface{}, m ...MiddlewareFunc) Callable

type CallableCallback

type CallableCallback func(callable Callable, err error)

type CallableCallbacks

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

func (*CallableCallbacks) DoClosed

func (this *CallableCallbacks) DoClosed(err error)

func (*CallableCallbacks) DoReady

func (this *CallableCallbacks) DoReady(err error)

func (*CallableCallbacks) SetOnClose

func (this *CallableCallbacks) SetOnClose(cb CallableCallback)

func (*CallableCallbacks) SetOnReady

func (this *CallableCallbacks) SetOnReady(cb CallableCallback)

type Context

type Context interface {
	Callable() Callable

	Id() string

	SetMethod(string)
	Method() string

	LocalFuncDesc() FuncDesc

	SetRequestHeader(header RpcMsgHeader)
	RequestHeader() RpcMsgHeader

	SetResponseHeader(header RpcMsgHeader)
	ResponseHeader() RpcMsgHeader

	SetRequest(in interface{})
	Request() (in interface{})

	SetResponse(out interface{})
	Response() (out interface{})

	RequestType() reflect.Type
	SetRequestType(t reflect.Type)

	ResponseType() reflect.Type
	SetResponseType(t reflect.Type)

	ReqMsg() *RawMsg
	SetReqMsg(msg *RawMsg)

	AckMsg() *RawMsg
	SetAckMsg(msg *RawMsg)

	FuncDesc() FuncDesc
	SetFuncDesc(desc FuncDesc)

	SetError(err error)
	Error() error

	SetWriter(w Writer)
	Writer() Writer

	AddDefer(deferFunc func())

	Reset()

	Init(call Callable, inMsg *RawMsg)

	BuildOutMsg() (*RawMsg, error)

	liblpc.UserDataStorage
}

type Core

type Core interface {
	Loop() *liblpc.IOEvtLoop
	RegFunc(f interface{}, m ...MiddlewareFunc)
	RegFuncWithName(fname string, f interface{}, m ...MiddlewareFunc)
	PreUse(m ...MiddlewareFunc)
	Use(m ...MiddlewareFunc)
	BuildChain(h HandleFunc) HandleFunc
	BuildPreUsedChain(h HandleFunc) HandleFunc
	Run(ctx context.Context)
	Start(ctx context.Context)
	GrabContext() Context
	ReleaseContext(c Context)
	PromiseGroup() *std.PromiseGroup
	NotifyCallableRead(call Callable, buf std.ReadableBuffer)
	io.Closer
}

func New

func New() (core Core, err error)

type FuncDesc

type FuncDesc uint8
const (
	ReqHasData FuncDesc = 0x01
	RspHasData FuncDesc = 0x02
)

type HandleFunc

type HandleFunc func(ctx Context)

type MiddlewareFunc

type MiddlewareFunc func(next HandleFunc) HandleFunc

type MsgType

type MsgType int
const (
	ReqMsg MsgType = iota
	AckMsg
)

type RawMsg

type RawMsg struct {
	Id         string            `json:"msgId"`
	MethodName string            `json:"methodName"`
	Headers    map[string]string `json:"headers"`
	Type       MsgType           `json:"type"` // req or ack
	Err        *string           `json:"err"`  // fast path for ack error
	Data       []byte            `json:"data"` // req = param
}

func (*RawMsg) BindData

func (this *RawMsg) BindData(v interface{}) error

func (*RawMsg) GetError

func (this *RawMsg) GetError() error

func (*RawMsg) SetData

func (this *RawMsg) SetData(v interface{}) error

func (*RawMsg) SetError

func (this *RawMsg) SetError(err error)

func (*RawMsg) SetErrorString

func (this *RawMsg) SetErrorString(es string)

type RpcMsgHeader

type RpcMsgHeader = map[string]string

type SignalCallable

type SignalCallable struct {
	Callable
	// contains filtered or unexported fields
}

func NewSignalCallable

func NewSignalCallable(call Callable) *SignalCallable

func (*SignalCallable) Close

func (this *SignalCallable) Close() error

func (*SignalCallable) CloseSignal

func (this *SignalCallable) CloseSignal() <-chan error

usage : err,ok := CloseSignal(); !ok -> return , ok -> check error

func (*SignalCallable) ReadySignal

func (this *SignalCallable) ReadySignal() <-chan error

usage : err,ok := ReadySignal(); !ok -> return , ok -> check error

type TimeWheelEntryImpl

type TimeWheelEntryImpl struct {
	io.Closer
	// contains filtered or unexported fields
}

func (*TimeWheelEntryImpl) BindTimeWheel

func (this *TimeWheelEntryImpl) BindTimeWheel(timeWheel *liblpc.TimeWheel)

not thread safe

func (*TimeWheelEntryImpl) GetRefCounter

func (this *TimeWheelEntryImpl) GetRefCounter() *int32

func (*TimeWheelEntryImpl) NotifyTimeWheel

func (this *TimeWheelEntryImpl) NotifyTimeWheel()

type VirtualCallable

type VirtualCallable struct {
	*BaseCallable
	// contains filtered or unexported fields
}

func NewVirtualCallable

func NewVirtualCallable(core Core, writer WriterCloser) *VirtualCallable

func (*VirtualCallable) Close

func (this *VirtualCallable) Close() error

func (VirtualCallable) Len

func (this VirtualCallable) Len() int

func (*VirtualCallable) MockReadData

func (this *VirtualCallable) MockReadData(data []byte)

not guarantee data seq, don't use in multi goroutine

func (*VirtualCallable) Start

func (this *VirtualCallable) Start()

func (VirtualCallable) Use

func (this VirtualCallable) Use(m ...MiddlewareFunc)

type Writer

type Writer interface {
	Write(ctx Context, data []byte, inLoop bool)
}

type WriterCloser

type WriterCloser interface {
	io.Closer
	Writer
}

Directories

Path Synopsis

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
⌂ 换网址