warden

README

net/rpc/warden

项目简介

来自 bilibili 主站技术部的 RPC 框架，融合主站技术部的核心科技，带来如飞一般的体验。

编译环境

• 请只用 Golang v1.9.x 以上版本编译执行

依赖包

• grpc

Documentation

Index

• Constants
• func GrpcAuthMiddleWare() grpc.UnaryServerInterceptor
• func WithDialLogFlag(flag int8) grpc.DialOption
• func WithLogFlag(flag int8) grpc.CallOption
• type Server
  • func NewServer(conf *ServerConfig, opt ...grpc.ServerOption) (s *Server)
  • func (s *Server) GetValidate() *validator.Validate
  • func (s *Server) RegisterValidation(key string, fn validator.Func) error
  • func (s *Server) Run(addr string) error
  • func (s *Server) RunUnix(file string) error
  • func (s *Server) Serve(lis net.Listener) error
  • func (s *Server) Server() *grpc.Server
  • func (s *Server) SetConfig(conf *ServerConfig) (err error)
  • func (s *Server) Shutdown(ctx context.Context) (err error)
  • func (s *Server) Start() (*Server, error)
  • func (s *Server) StartWithAddr() (*Server, net.Addr, error)
  • func (s *Server) Use(handlers ...grpc.UnaryServerInterceptor) *Server
• type ServerConfig

Examples

• Server

Constants

const (
	// disable all log.
	LogFlagDisable = 1 << iota
	// disable print args on log.
	LogFlagDisableArgs
	// disable info level log.
	LogFlagDisableInfo
)

Warden Log Flag

Functions

func GrpcAuthMiddleWare

func GrpcAuthMiddleWare() grpc.UnaryServerInterceptor

func WithDialLogFlag

func WithDialLogFlag(flag int8) grpc.DialOption

WithDialLogFlag set client level log behaviour.

func WithLogFlag

func WithLogFlag(flag int8) grpc.CallOption

WithLogFlag disable client access log.

Types

type Server

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

Server is the framework's server side instance, it contains the GrpcServer, interceptor and interceptors. Create an instance of Server, by using NewServer().

Example

package main

import (
	"context"
	"io"
	"time"

	"github.com/zhangjinglei/wahaha/pkg/net/rpc/warden"

	pb "github.com/zhangjinglei/wahaha/pkg/net/rpc/warden/internal/proto/testproto"

	xtime "github.com/zhangjinglei/wahaha/pkg/time"

	"google.golang.org/grpc"
)

type helloServer struct {
}

func (s *helloServer) SayHello(ctx context.Context, in *pb.HelloRequest) (*pb.HelloReply, error) {
	return &pb.HelloReply{Message: "Hello " + in.Name, Success: true}, nil
}

func (s *helloServer) StreamHello(ss pb.Greeter_StreamHelloServer) error {
	for i := 0; i < 3; i++ {
		in, err := ss.Recv()
		if err == io.EOF {
			return nil
		}
		if err != nil {
			return err
		}
		ret := &pb.HelloReply{Message: "Hello " + in.Name, Success: true}
		err = ss.Send(ret)
		if err != nil {
			return err
		}
	}
	return nil

}

func main() {
	s := warden.NewServer(&warden.ServerConfig{Timeout: xtime.Duration(time.Second), Addr: ":8080"})
	// apply server interceptor middleware
	s.Use(func(ctx context.Context, req interface{}, args *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
		newctx, cancel := context.WithTimeout(ctx, time.Second*10)
		defer cancel()
		resp, err := handler(newctx, req)
		return resp, err
	})
	pb.RegisterGreeterServer(s.Server(), &helloServer{})
	s.Start()
}

func NewServer

func NewServer(conf *ServerConfig, opt ...grpc.ServerOption) (s *Server)

NewServer returns a new blank Server instance with a default server interceptor.

func (*Server) GetValidate

func (s *Server) GetValidate() *validator.Validate

GetValidate return the default validate

func (*Server) RegisterValidation

func (s *Server) RegisterValidation(key string, fn validator.Func) error

RegisterValidation adds a validation Func to a Validate's map of validators denoted by the key NOTE: if the key already exists, the previous validation function will be replaced. NOTE: this method is not thread-safe it is intended that these all be registered prior to any validation

func (*Server) Run

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

Run create a tcp listener and start goroutine for serving each incoming request. Run will return a non-nil error unless Stop or GracefulStop is called.

func (*Server) RunUnix

func (s *Server) RunUnix(file string) error

RunUnix create a unix listener and start goroutine for serving each incoming request. RunUnix will return a non-nil error unless Stop or GracefulStop is called.

func (*Server) Serve

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

Serve accepts incoming connections on the listener lis, creating a new ServerTransport and service goroutine for each. Serve will return a non-nil error unless Stop or GracefulStop is called.

func (*Server) Server

func (s *Server) Server() *grpc.Server

Server return the grpc server for registering service.

func (*Server) SetConfig

func (s *Server) SetConfig(conf *ServerConfig) (err error)

SetConfig hot reloads server config

func (*Server) Shutdown

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

Shutdown stops the server gracefully. It stops the server from accepting new connections and RPCs and blocks until all the pending RPCs are finished or the context deadline is reached.

func (*Server) Start

func (s *Server) Start() (*Server, error)

Start create a new goroutine run server with configured listen addr will panic if any error happend return server itself

func (*Server) StartWithAddr

func (s *Server) StartWithAddr() (*Server, net.Addr, error)

StartWithAddr create a new goroutine run server with configured listen addr will panic if any error happend return server itself and the actually listened address (if configured listen port is zero, the os will allocate an unused port)

func (*Server) Use

func (s *Server) Use(handlers ...grpc.UnaryServerInterceptor) *Server

Use attachs a global inteceptor to the server. For example, this is the right place for a rate limiter or error management inteceptor.

type ServerConfig

type ServerConfig struct {
	// Network is grpc listen network,default value is tcp
	Network string `dsn:"network"`
	// Addr is grpc listen addr,default value is 0.0.0.0:9000
	Addr string `dsn:"address"`
	// Timeout is context timeout for per rpc call.
	Timeout xtime.Duration `dsn:"query.timeout"`
	// IdleTimeout is a duration for the amount of time after which an idle connection would be closed by sending a GoAway.
	// Idleness duration is defined since the most recent time the number of outstanding RPCs became zero or the connection establishment.
	IdleTimeout xtime.Duration `dsn:"query.idleTimeout"`
	// MaxLifeTime is a duration for the maximum amount of time a connection may exist before it will be closed by sending a GoAway.
	// A random jitter of +/-10% will be added to MaxConnectionAge to spread out connection storms.
	MaxLifeTime xtime.Duration `dsn:"query.maxLife"`
	// ForceCloseWait is an additive period after MaxLifeTime after which the connection will be forcibly closed.
	ForceCloseWait xtime.Duration `dsn:"query.closeWait"`
	// KeepAliveInterval is after a duration of this time if the server doesn't see any activity it pings the client to see if the transport is still alive.
	KeepAliveInterval xtime.Duration `dsn:"query.keepaliveInterval"`
	// KeepAliveTimeout  is After having pinged for keepalive check, the server waits for a duration of Timeout and if no activity is seen even after that
	// the connection is closed.
	KeepAliveTimeout xtime.Duration `dsn:"query.keepaliveTimeout"`
	// LogFlag to control log behaviour. e.g. LogFlag: warden.LogFlagDisableLog.
	// Disable: 1 DisableArgs: 2 DisableInfo: 4
	LogFlag int8 `dsn:"query.logFlag"`
}

ServerConfig is rpc server conf.