Skip to content

G
getty
Commit 8eec657e authored by AlexStocks's avatar AlexStocks
Browse files
Options

Add: RPC

parent 3f3fa13a
Hide whitespace changes
Inline Side-by-side
Showing with 227 additions and 291 deletions
change_log.md
View file @ 8eec657e
......@@ -14,9 +14,13 @@
## develop history ##
---
- 2018/07/01
> Feature
* Add RPC
- 2018/06/25
> buf fix
* delete juju/errors on read/write in case of network i/o timeout
* Using juju/errors.Cause on read/write in case of network i/o timeout
- 2018/03/29
> improvement
......
client.go
View file @ 8eec657e
......@@ -185,7 +185,7 @@ func (c *client) dialUDP() Session {
}
conn.SetReadDeadline(wheel.Now().Add(1e9))
length, err = conn.Read(buf)
if netErr, ok := err.(net.Error); ok && netErr.Timeout() {
if netErr, ok := jerrors.Cause(err).(net.Error); ok && netErr.Timeout() {
err = nil
}
if err != nil {
......
conn.go
View file @ 8eec657e
......@@ -249,8 +249,8 @@ func (t *gettyTCPConn) read(p []byte) (int, error) {
length, err = t.reader.Read(p)
log.Debug("now:%s, length:%d, err:%s", currentTime, length, err)
atomic.AddUint32(&t.readBytes, uint32(length))
//return length, jerrors.Trace(err)
return length, err
return length, jerrors.Trace(err)
//return length, err
}
// tcp connection write
......@@ -283,8 +283,8 @@ func (t *gettyTCPConn) Write(pkg interface{}) (int, error) {
atomic.AddUint32(&t.writeBytes, (uint32)(len(p)))
}
log.Debug("now:%s, length:%d, err:%s", currentTime, length, err)
//return length, jerrors.Trace(err)
return length, err
return length, jerrors.Trace(err)
//return length, err
}
// close tcp connection
......@@ -402,8 +402,8 @@ func (u *gettyUDPConn) read(p []byte) (int, *net.UDPAddr, error) {
atomic.AddUint32(&u.readBytes, uint32(length))
}
return length, addr, err
// return length, addr, jerrors.Trace(err)
//return length, addr, err
return length, addr, jerrors.Trace(err)
}
// write udp packet, @ctx should be of type UDPContext
......@@ -449,8 +449,8 @@ func (u *gettyUDPConn) Write(udpCtx interface{}) (int, error) {
}
log.Debug("WriteMsgUDP(peerAddr:%s) = {length:%d, error:%s}", peerAddr, length, err)
// return length, jerrors.Trace(err)
return length, err
return length, jerrors.Trace(err)
//return length, err
}
// close udp connection
......@@ -547,8 +547,8 @@ func (w *gettyWSConn) read() ([]byte, error) {
}
}
// return b, jerrors.Trace(e)
return b, e
return b, jerrors.Trace(e)
//return b, e
}
func (w *gettyWSConn) updateWriteDeadline() error {
......@@ -589,8 +589,8 @@ func (w *gettyWSConn) Write(pkg interface{}) (int, error) {
if err = w.conn.WriteMessage(websocket.BinaryMessage, p); err == nil {
atomic.AddUint32(&w.writeBytes, (uint32)(len(p)))
}
// return len(p), jerrors.Trace(err)
return len(p), err
return len(p), jerrors.Trace(err)
//return len(p), err
}
func (w *gettyWSConn) writePing() error {
......
rpc/client.go
View file @ 8eec657e
package rpc
import (
"errors"
"fmt"
"math/rand"
"net"
"sync"
"sync/atomic"
"time"
)
import (
jerrors "github.com/juju/errors"
)
import (
"github.com/AlexStocks/getty"
"github.com/AlexStocks/goext/net"
log "github.com/AlexStocks/log4go"
)
var (
errInvalidAddress = errors.New("remote address invalid or empty")
errSessionNotExist = errors.New("session not exist")
errInvalidAddress = jerrors.New("remote address invalid or empty")
errSessionNotExist = jerrors.New("session not exist")
errClientClosed = jerrors.New("client closed")
)
func init() {
......@@ -24,6 +30,7 @@ func init() {
}
type Client struct {
conf *Config
lock sync.RWMutex
sessions []*rpcSession
gettyClient getty.Client
......@@ -36,22 +43,15 @@ type Client struct {
sendLock sync.Mutex
}
func NewClient() *Client {
func NewClient(conf *Config) *Client {
c := &Client{
pendingResponses: make(map[uint64]*PendingResponse),
conf: conf,
gettyClient: getty.NewTCPClient(
getty.WithServerAddress(gxnet.HostAddress(conf.ServerHost, conf.ServerPort)),
getty.WithConnectionNumber((int)(conf.ConnectionNum)),
),
}
c.Init()
return c
}
func (c *Client) Init() {
initConf(defaultClientConfFile)
initLog(defaultClientLogConfFile)
initProfiling()
c.gettyClient = getty.NewTCPClient(
getty.WithServerAddress(gxnet.HostAddress(conf.ServerHost, conf.ServerPort)),
getty.WithConnectionNumber((int)(conf.ConnectionNum)),
)
c.gettyClient.RunEventLoop(c.newSession)
for {
if c.isAvailable() {
......@@ -60,6 +60,8 @@ func (c *Client) Init() {
time.Sleep(1e6)
}
log.Info("client init ok")
return c
}
func (c *Client) newSession(session getty.Session) error {
......@@ -86,8 +88,8 @@ func (c *Client) newSession(session getty.Session) error {
session.SetName(conf.GettySessionParam.SessionName)
session.SetMaxMsgLen(conf.GettySessionParam.MaxMsgLen)
session.SetPkgHandler(NewRpcClientPacketHandler()) //
session.SetEventListener(NewRpcClientHandler(c)) //
session.SetPkgHandler(NewRpcClientPacketHandler())
session.SetEventListener(NewRpcClientHandler(c))
session.SetRQLen(conf.GettySessionParam.PkgRQSize)
session.SetWQLen(conf.GettySessionParam.PkgWQSize)
session.SetReadTimeout(conf.GettySessionParam.tcpReadTimeout)
......@@ -123,6 +125,7 @@ func (c *Client) Call(service, method string, args interface{}, reply interface{
<-resp.done
return resp.err
}
return errSessionNotExist
}
......@@ -130,27 +133,39 @@ func (c *Client) isAvailable() bool {
if c.selectSession() == nil {
return false
}
return true
}
func (c *Client) Close() {
var sessions *[]*rpcSession
c.lock.Lock()
if c.gettyClient != nil {
for _, s := range c.sessions {
log.Info("close client session{%s, last active:%s, request number:%d}",
s.session.Stat(), s.session.GetActive().String(), s.reqNum)
s.session.Close()
}
sessions = &(c.sessions)
c.sessions = nil
c.gettyClient.Close()
c.gettyClient = nil
c.sessions = c.sessions[:0]
}
c.lock.Unlock()
if sessions != nil {
for _, s := range *sessions {
log.Info("close client session{%s, last active:%s, request number:%d}",
s.session.Stat(), s.session.GetActive().String(), s.reqNum)
s.session.Close()
}
}
}
func (c *Client) selectSession() getty.Session {
c.lock.RLock()
defer c.lock.RUnlock()
if c.sessions == nil {
return nil
}
count := len(c.sessions)
if count == 0 {
return nil
......@@ -165,15 +180,25 @@ func (c *Client) addSession(session getty.Session) {
}
c.lock.Lock()
defer c.lock.Unlock()
if c.sessions == nil {
return
}
c.sessions = append(c.sessions, &rpcSession{session: session})
c.lock.Unlock()
}
func (c *Client) removeSession(session getty.Session) {
if session == nil {
return
}
c.lock.Lock()
defer c.lock.Unlock()
if c.sessions == nil {
return
}
for i, s := range c.sessions {
if s.session == session {
c.sessions = append(c.sessions[:i], c.sessions[i+1:]...)
......@@ -182,7 +207,6 @@ func (c *Client) removeSession(session getty.Session) {
}
}
log.Info("after remove session{%s}, left session number:%d", session.Stat(), len(c.sessions))
c.lock.Unlock()
}
func (c *Client) updateSession(session getty.Session) {
......@@ -190,13 +214,17 @@ func (c *Client) updateSession(session getty.Session) {
return
}
c.lock.Lock()
defer c.lock.Unlock()
if c.sessions == nil {
return
}
for i, s := range c.sessions {
if s.session == session {
c.sessions[i].reqNum++
break
}
}
c.lock.Unlock()
}
func (c *Client) getClientRpcSession(session getty.Session) (rpcSession, error) {
......@@ -205,6 +233,11 @@ func (c *Client) getClientRpcSession(session getty.Session) (rpcSession, error)
rpcSession rpcSession
)
c.lock.Lock()
defer c.lock.Unlock()
if c.sessions == nil {
return rpcSession, errClientClosed
}
err = errSessionNotExist
for _, s := range c.sessions {
if s.session == session {
......@@ -213,7 +246,6 @@ func (c *Client) getClientRpcSession(session getty.Session) (rpcSession, error)
break
}
}
c.lock.Unlock()
return rpcSession, err
}
......
rpc/config.go
View file @ 8eec657e
package rpc
import (
"fmt"
"path"
"time"
)
import (
log "github.com/AlexStocks/log4go"
config "github.com/koding/multiconfig"
)
const (
defaultClientConfFile string = "client_config.toml"
defaultClientLogConfFile string = "client_log.xml"
defaultServerConfFile string = "server_config.toml"
defaultServerLogConfFile string = "server_log.xml"
)
var (
conf *Config
)
type (
GettySessionParam struct {
CompressEncoding bool `default:"false"`
......@@ -75,51 +57,3 @@ type (
GettySessionParam GettySessionParam `required:"true"`
}
)
func initConf(confFile string) {
var err error
if path.Ext(confFile) != ".toml" {
panic(fmt.Sprintf("application configure file name{%v} suffix must be .toml", confFile))
}
conf = new(Config)
config.MustLoadWithPath(confFile, conf)
conf.heartbeatPeriod, err = time.ParseDuration(conf.HeartbeatPeriod)
if err != nil {
panic(fmt.Sprintf("time.ParseDuration(HeartbeatPeroid{%#v}) = error{%v}", conf.HeartbeatPeriod, err))
}
conf.sessionTimeout, err = time.ParseDuration(conf.SessionTimeout)
if err != nil {
panic(fmt.Sprintf("time.ParseDuration(SessionTimeout{%#v}) = error{%v}", conf.SessionTimeout, err))
}
conf.failFastTimeout, err = time.ParseDuration(conf.FailFastTimeout)
if err != nil {
panic(fmt.Sprintf("time.ParseDuration(FailFastTimeout{%#v}) = error{%v}", conf.FailFastTimeout, err))
}
conf.GettySessionParam.keepAlivePeriod, err = time.ParseDuration(conf.GettySessionParam.KeepAlivePeriod)
if err != nil {
panic(fmt.Sprintf("time.ParseDuration(KeepAlivePeriod{%#v}) = error{%v}", conf.GettySessionParam.KeepAlivePeriod, err))
}
conf.GettySessionParam.tcpReadTimeout, err = time.ParseDuration(conf.GettySessionParam.TcpReadTimeout)
if err != nil {
panic(fmt.Sprintf("time.ParseDuration(TcpReadTimeout{%#v}) = error{%v}", conf.GettySessionParam.TcpReadTimeout, err))
}
conf.GettySessionParam.tcpWriteTimeout, err = time.ParseDuration(conf.GettySessionParam.TcpWriteTimeout)
if err != nil {
panic(fmt.Sprintf("time.ParseDuration(TcpWriteTimeout{%#v}) = error{%v}", conf.GettySessionParam.TcpWriteTimeout, err))
}
conf.GettySessionParam.waitTimeout, err = time.ParseDuration(conf.GettySessionParam.WaitTimeout)
if err != nil {
panic(fmt.Sprintf("time.ParseDuration(WaitTimeout{%#v}) = error{%v}", conf.GettySessionParam.WaitTimeout, err))
}
return
}
func initLog(logFile string) {
if path.Ext(logFile) != ".xml" {
panic(fmt.Sprintf("log configure file name{%v} suffix must be .xml", logFile))
}
log.LoadConfiguration(logFile)
log.Info("config{%#v}", conf)
}
rpc/handler.go
View file @ 8eec657e
......@@ -2,12 +2,14 @@ package rpc
import (
"encoding/json"
"errors"
"reflect"
"sync"
"time"
)
import (
"github.com/AlexStocks/getty"
jerrors "github.com/juju/errors"
log "github.com/AlexStocks/log4go"
)
......@@ -19,7 +21,7 @@ const (
)
var (
errTooManySessions = errors.New("too many echo sessions")
errTooManySessions = jerrors.New("too many echo sessions")
)
type rpcSession struct {
......@@ -29,23 +31,26 @@ type rpcSession struct {
}
type RpcServerHandler struct {
sessionMap map[getty.Session]*rpcSession
rwlock sync.RWMutex
maxSessionNum int
sessionTimeout time.Duration
sessionMap map[getty.Session]*rpcSession
rwlock sync.RWMutex
sendLock sync.Mutex
}
func NewRpcServerHandler() *RpcServerHandler {
r := &RpcServerHandler{
sessionMap: make(map[getty.Session]*rpcSession),
func NewRpcServerHandler(maxSessionNum int, sessionTimeout time.Duration) *RpcServerHandler {
return &RpcServerHandler{
maxSessionNum: maxSessionNum,
sessionTimeout: sessionTimeout,
sessionMap: make(map[getty.Session]*rpcSession),
}
return r
}
func (h *RpcServerHandler) OnOpen(session getty.Session) error {
var err error
h.rwlock.RLock()
if conf.SessionNumber < len(h.sessionMap) {
if h.maxSessionNum < len(h.sessionMap) {
err = errTooManySessions
}
h.rwlock.RUnlock()
......@@ -101,7 +106,7 @@ func (h *RpcServerHandler) OnCron(session getty.Session) {
h.rwlock.RLock()
if _, ok := h.sessionMap[session]; ok {
active = session.GetActive()
if conf.sessionTimeout.Nanoseconds() < time.Since(active).Nanoseconds() {
if h.sessionTimeout.Nanoseconds() < time.Since(active).Nanoseconds() {
flag = true
log.Warn("session{%s} timeout{%s}, reqNum{%d}",
session.Stat(), time.Since(active).String(), h.sessionMap[session].reqNum)
......@@ -198,13 +203,13 @@ func (h *RpcClientHandler) OnMessage(session getty.Session, pkg interface{}) {
return
}
if len(p.header.Error) > 0 {
pendingResponse.err = errors.New(p.header.Error)
pendingResponse.err = jerrors.New(p.header.Error)
}
err := json.Unmarshal(p.body.([]byte), pendingResponse.reply)
if err != nil {
pendingResponse.err = err
}
pendingResponse.done <- true
pendingResponse.done <- struct{}{}
}
func (h *RpcClientHandler) OnCron(session getty.Session) {
......
rpc/packet.go
View file @ 8eec657e
......@@ -4,10 +4,13 @@ import (
"bytes"
"encoding/binary"
"encoding/json"
"errors"
"reflect"
)
import (
jerrors "github.com/juju/errors"
)
const (
MaxPacketLen = 16 * 1024
RequestSendOnly int16 = 1
......@@ -18,9 +21,9 @@ const (
)
var (
ErrNotEnoughStream = errors.New("packet stream is not enough")
ErrTooLargePackage = errors.New("package length is exceed the echo package's legal maximum length.")
ErrNotFoundServiceOrMethod = errors.New("server invalid service or method")
ErrNotEnoughStream = jerrors.New("packet stream is not enough")
ErrTooLargePackage = jerrors.New("package length is exceed the echo package's legal maximum length.")
ErrNotFoundServiceOrMethod = jerrors.New("server invalid service or method")
)
type RequestHeader struct {
......@@ -279,9 +282,9 @@ type PendingResponse struct {
seq uint64
err error
reply interface{}
done chan bool
done chan struct{}
}
func NewPendingResponse() *PendingResponse {
return &PendingResponse{done: make(chan bool)}
return &PendingResponse{done: make(chan struct{})}
}
rpc/readwriter.go
View file @ 8eec657e
......@@ -2,12 +2,12 @@ package rpc
import (
"bytes"
"errors"
)
import (
"github.com/AlexStocks/getty"
log "github.com/AlexStocks/log4go"
jerrors "github.com/juju/errors"
)
type RpcServerPacketHandler struct {
......@@ -49,7 +49,7 @@ func (p *RpcServerPacketHandler) Write(ss getty.Session, pkg interface{}) error
if resp, ok = pkg.(*RpcResponse); !ok {
log.Error("illegal pkg:%+v\n", pkg)
return errors.New("invalid rpc response")
return jerrors.New("invalid rpc response")
}
buf, err = resp.Marshal()
......@@ -99,7 +99,7 @@ func (p *RpcClientPacketHandler) Write(ss getty.Session, pkg interface{}) error
if req, ok = pkg.(*RpcRequest); !ok {
log.Error("illegal pkg:%+v\n", pkg)
return errors.New("invalid rpc request")
return jerrors.New("invalid rpc request")
}
buf, err = req.Marshal()
......
rpc/rpc.go
View file @ 8eec657e
......@@ -2,17 +2,52 @@ package rpc
import (
"reflect"
"sync"
"unicode"
"unicode/utf8"
)
import (
log "github.com/AlexStocks/log4go"
)
var typeOfError = reflect.TypeOf((*error)(nil)).Elem()
var (
typeOfError = reflect.TypeOf((*error)(nil)).Elem()
)
type methodType struct {
sync.Mutex
method reflect.Method
ArgType reflect.Type
ReplyType reflect.Type
numCalls uint
}
type service struct {
name string
rcvr reflect.Value
typ reflect.Type
method map[string]*methodType
}
// Is this an exported - upper case - name
func isExported(name string) bool {
rune, _ := utf8.DecodeRuneInString(name)
return unicode.IsUpper(rune)
}
// suitableMethods returns suitable Rpc methods of typ, it will report
// error using log if reportErr is true.
func suitableMethods(typ reflect.Type, reportErr bool) map[string]*methodType {
// Is this type exported or a builtin?
func isExportedOrBuiltinType(t reflect.Type) bool {
for t.Kind() == reflect.Ptr {
t = t.Elem()
}
// PkgPath will be non-empty even for an exported type,
// so we need to check the type name as well.
return isExported(t.Name()) || t.PkgPath() == ""
}
// prepareMethods returns suitable Rpc methods of typ
func prepareMethods(typ reflect.Type) map[string]*methodType {
methods := make(map[string]*methodType)
for m := 0; m < typ.NumMethod(); m++ {
method := typ.Method(m)
......@@ -24,65 +59,37 @@ func suitableMethods(typ reflect.Type, reportErr bool) map[string]*methodType {
}
// Method needs three ins: receiver, *args, *reply.
if mtype.NumIn() != 3 {
if reportErr {
log.Warn("method", mname, "has wrong number of ins:", mtype.NumIn())
}
log.Warn("method %s has wrong number of ins %d which should be 3", mname, mtype.NumIn())
continue
}
// First arg need not be a pointer.
argType := mtype.In(1)
if !isExportedOrBuiltinType(argType) {
if reportErr {
log.Warn(mname, "argument type not exported:", argType)
}
log.Error("method{%s} argument type not exported{%v}", mname, argType)
continue
}
// Second arg must be a pointer.
replyType := mtype.In(2)
if replyType.Kind() != reflect.Ptr {
if reportErr {
log.Warn("method", mname, "reply type not a pointer:", replyType)
}
log.Error("method{%s} reply type not a pointer{%v}", mname, replyType)
continue
}
// Reply type must be exported.
if !isExportedOrBuiltinType(replyType) {
if reportErr {
log.Warn("method", mname, "reply type not exported:", replyType)
}
log.Error("method{%s} reply type not exported{%v}", mname, replyType)
continue
}
// Method needs one out.
if mtype.NumOut() != 1 {
if reportErr {
log.Warn("method", mname, "has wrong number of outs:", mtype.NumOut())
}
log.Error("method{%s} has wrong number of out parameters{%d}", mname, mtype.NumOut())
continue
}
// The return type of the method must be error.
if returnType := mtype.Out(0); returnType != typeOfError {
if reportErr {
log.Warn("method", mname, "returns", returnType.String(), "not error")
}
log.Error("method{%s}'s return type{%s} is not error", mname, returnType.String())
continue
}
methods[mname] = &methodType{method: method, ArgType: argType, ReplyType: replyType}
}
return methods
}
// Is this an exported - upper case - name
func isExported(name string) bool {
rune, _ := utf8.DecodeRuneInString(name)
return unicode.IsUpper(rune)
}
// Is this type exported or a builtin?
func isExportedOrBuiltinType(t reflect.Type) bool {
for t.Kind() == reflect.Ptr {
t = t.Elem()
}
// PkgPath will be non-empty even for an exported type,
// so we need to check the type name as well.
return isExported(t.Name()) || t.PkgPath() == ""
}
rpc/server.go
View file @ 8eec657e
package rpc
import (
"errors"
"fmt"
"net"
"os"
......@@ -9,10 +8,12 @@ import (
"reflect"
"syscall"
"time"
)
import (
"github.com/AlexStocks/getty"
"github.com/AlexStocks/goext/log"
"github.com/AlexStocks/goext/net"
jerrors "github.com/juju/errors"
log "github.com/AlexStocks/log4go"
)
......@@ -20,72 +21,72 @@ import (
type Server struct {
tcpServerList []getty.Server
serviceMap map[string]*service
conf *Config
}
func NewServer() *Server {
func NewServer(conf *Config) *Server {
s := &Server{
serviceMap: make(map[string]*service),
conf: conf,
}
return s
}
func (server *Server) Run() {
initConf(defaultServerConfFile)
initLog(defaultServerLogConfFile)
initProfiling()
server.Init()
gxlog.CInfo("%s starts successfull! its version=%s, its listen ends=%s:%s\n",
conf.AppName, Version, conf.Host, conf.Ports)
func (s *Server) Run() {
s.Init()
log.Info("%s starts successfull! its version=%s, its listen ends=%s:%s\n",
conf.AppName, Version, conf.Host, conf.Ports)
server.initSignal()
s.conf.AppName, getty.Version, s.conf.Host, s.conf.Ports)
s.initSignal()
}
func (server *Server) Register(rcvr interface{}) error {
s := new(service)
s.typ = reflect.TypeOf(rcvr)
s.rcvr = reflect.ValueOf(rcvr)
sname := reflect.Indirect(s.rcvr).Type().Name()
if sname == "" {
s := "rpc.Register: no service name for type " + s.typ.String()
func (s *Server) Register(rcvr interface{}) error {
svc := &service{
typ: reflect.TypeOf(rcvr),
rcvr: reflect.ValueOf(rcvr),
name: reflect.Indirect(reflect.ValueOf(rcvr)).Type().Name(),
// Install the methods
method: prepareMethods(reflect.TypeOf(rcvr)),
}
if svc.name == "" {
s := "rpc.Register: no service name for type " + svc.typ.String()
log.Error(s)
return errors.New(s)
return jerrors.New(s)
}
if !isExported(sname) {
s := "rpc.Register: type " + sname + " is not exported"
if !isExported(svc.name) {
s := "rpc.Register: type " + svc.name + " is not exported"
log.Error(s)
return errors.New(s)
return jerrors.New(s)
}
if _, present := server.serviceMap[sname]; present {
return errors.New("rpc: service already defined: " + sname)
if _, present := s.serviceMap[svc.name]; present {
return jerrors.New("rpc: service already defined: " + svc.name)
}
s.name = sname
// Install the methods
s.method = suitableMethods(s.typ, true)
if len(s.method) == 0 {
str := ""
if len(svc.method) == 0 {
// To help the user, see if a pointer receiver would work.
method := suitableMethods(reflect.PtrTo(s.typ), false)
method := prepareMethods(reflect.PtrTo(svc.typ))
str := "rpc.Register: type " + svc.name + " has no exported methods of suitable type"
if len(method) != 0 {
str = "rpc.Register: type " + sname + " has no exported methods of suitable type (hint: pass a pointer to value of that type)"
} else {
str = "rpc.Register: type " + sname + " has no exported methods of suitable type"
str = "rpc.Register: type " + svc.name + " has no exported methods of suitable type (" +
"hint: pass a pointer to value of that type)"
}
log.Error(s)
return errors.New(str)
log.Error(str)
return jerrors.New(str)
}
server.serviceMap[s.name] = s
s.serviceMap[svc.name] = svc
return nil
}
func (server *Server) newSession(session getty.Session) error {
func (s *Server) newSession(session getty.Session) error {
var (
ok bool
tcpConn *net.TCPConn
)
if conf.GettySessionParam.CompressEncoding {
if s.conf.GettySessionParam.CompressEncoding {
session.SetCompressType(getty.CompressZip)
}
......@@ -93,59 +94,59 @@ func (server *Server) newSession(session getty.Session) error {
panic(fmt.Sprintf("%s, session.conn{%#v} is not tcp connection\n", session.Stat(), session.Conn()))
}
tcpConn.SetNoDelay(conf.GettySessionParam.TcpNoDelay)
tcpConn.SetKeepAlive(conf.GettySessionParam.TcpKeepAlive)
if conf.GettySessionParam.TcpKeepAlive {
tcpConn.SetKeepAlivePeriod(conf.GettySessionParam.keepAlivePeriod)
tcpConn.SetNoDelay(s.conf.GettySessionParam.TcpNoDelay)
tcpConn.SetKeepAlive(s.conf.GettySessionParam.TcpKeepAlive)
if s.conf.GettySessionParam.TcpKeepAlive {
tcpConn.SetKeepAlivePeriod(s.conf.GettySessionParam.keepAlivePeriod)
}
tcpConn.SetReadBuffer(conf.GettySessionParam.TcpRBufSize)
tcpConn.SetWriteBuffer(conf.GettySessionParam.TcpWBufSize)
session.SetName(conf.GettySessionParam.SessionName)
session.SetMaxMsgLen(conf.GettySessionParam.MaxMsgLen)
session.SetPkgHandler(NewRpcServerPacketHandler(server)) //
session.SetEventListener(NewRpcServerHandler()) //
session.SetRQLen(conf.GettySessionParam.PkgRQSize)
session.SetWQLen(conf.GettySessionParam.PkgWQSize)
session.SetReadTimeout(conf.GettySessionParam.tcpReadTimeout)
session.SetWriteTimeout(conf.GettySessionParam.tcpWriteTimeout)
session.SetCronPeriod((int)(conf.sessionTimeout.Nanoseconds() / 1e6))
session.SetWaitTime(conf.GettySessionParam.waitTimeout)
tcpConn.SetReadBuffer(s.conf.GettySessionParam.TcpRBufSize)
tcpConn.SetWriteBuffer(s.conf.GettySessionParam.TcpWBufSize)
session.SetName(s.conf.GettySessionParam.SessionName)
session.SetMaxMsgLen(s.conf.GettySessionParam.MaxMsgLen)
session.SetPkgHandler(NewRpcServerPacketHandler(s)) //
session.SetEventListener(NewRpcServerHandler()) //
session.SetRQLen(s.conf.GettySessionParam.PkgRQSize)
session.SetWQLen(s.conf.GettySessionParam.PkgWQSize)
session.SetReadTimeout(s.conf.GettySessionParam.tcpReadTimeout)
session.SetWriteTimeout(s.conf.GettySessionParam.tcpWriteTimeout)
session.SetCronPeriod((int)(s.conf.sessionTimeout.Nanoseconds() / 1e6))
session.SetWaitTime(s.conf.GettySessionParam.waitTimeout)
log.Debug("app accepts new session:%s\n", session.Stat())
return nil
}
func (server *Server) Init() {
func (s *Server) Init() {
var (
addr string
portList []string
tcpServer getty.Server
)
portList = conf.Ports
portList = s.conf.Ports
if len(portList) == 0 {
panic("portList is nil")
}
for _, port := range portList {
addr = gxnet.HostAddress2(conf.Host, port)
addr = gxnet.HostAddress2(s.conf.Host, port)
tcpServer = getty.NewTCPServer(
getty.WithLocalAddress(addr),
)
// run server
tcpServer.RunEventLoop(server.newSession)
log.Debug("server bind addr{%s} ok!", addr)
server.tcpServerList = append(server.tcpServerList, tcpServer)
// run s
tcpServer.RunEventLoop(s.newSession)
log.Debug("s bind addr{%s} ok!", addr)
s.tcpServerList = append(s.tcpServerList, tcpServer)
}
}
func (server *Server) Stop() {
for _, tcpServer := range server.tcpServerList {
func (s *Server) Stop() {
for _, tcpServer := range s.tcpServerList {
tcpServer.Close()
}
}
func (server *Server) initSignal() {
func (s *Server) initSignal() {
// signal.Notify的ch信道是阻塞的(signal.Notify不会阻塞发送信号), 需要设置缓冲
signals := make(chan os.Signal, 1)
// It is not possible to block SIGKILL or syscall.SIGSTOP
......@@ -157,7 +158,7 @@ func (server *Server) initSignal() {
case syscall.SIGHUP:
// reload()
default:
go time.AfterFunc(conf.failFastTimeout, func() {
go time.AfterFunc(s.conf.failFastTimeout, func() {
// log.Warn("app exit now by force...")
// os.Exit(1)
log.Exit("app exit now by force...")
......@@ -165,7 +166,7 @@ func (server *Server) initSignal() {
})
// 要么survialTimeout时间内执行完毕下面的逻辑然后程序退出，要么执行上面的超时函数程序强行退出
server.Stop()
s.Stop()
// fmt.Println("app exit now...")
log.Exit("app exit now...")
log.Close()
......
rpc/service.go deleted 100644 → 0
View file @ 3f3fa13a
package rpc
import (
"reflect"
"sync"
)
type methodType struct {
sync.Mutex
method reflect.Method
ArgType reflect.Type
ReplyType reflect.Type
numCalls uint
}
type service struct {
name string
rcvr reflect.Value
typ reflect.Type
method map[string]*methodType
}
rpc/util.go deleted 100644 → 0
View file @ 3f3fa13a
package rpc
import (
"net/http"
"github.com/AlexStocks/goext/net"
log "github.com/AlexStocks/log4go"
)
const (
pprofPath = "/debug/pprof/"
)
func initProfiling() {
var (
addr string
)
// addr = *host + ":" + "10000"
addr = gxnet.HostAddress(conf.Host, conf.ProfilePort)
log.Info("App Profiling startup on address{%v}", addr+pprofPath)
go func() {
log.Info(http.ListenAndServe(addr, nil))
}()
}
rpc/version.go deleted 100644 → 0
View file @ 3f3fa13a
package rpc
var (
Version = "0.8.2"
)
session.go
View file @ 8eec657e
......@@ -594,7 +594,7 @@ func (s *session) handleTCPPackage() error {
// s.conn.SetReadTimeout(time.Now().Add(s.rTimeout))
bufLen, err = conn.read(buf)
if err != nil {
if netError, ok = err.(net.Error); ok && netError.Timeout() {
if netError, ok = jerrors.Cause(err).(net.Error); ok && netError.Timeout() {
break
}
log.Error("%s, [session.conn.read] = error{%s}", s.sessionToken(), jerrors.ErrorStack(err))
......@@ -670,7 +670,7 @@ func (s *session) handleUDPPackage() error {
bufLen, addr, err = conn.read(buf)
log.Debug("conn.read() = bufLen:%d, addr:%#v, err:%s", bufLen, addr, jerrors.ErrorStack(err))
if netError, ok = err.(net.Error); ok && netError.Timeout() {
if netError, ok = jerrors.Cause(err).(net.Error); ok && netError.Timeout() {
continue
}
if err != nil {
......@@ -730,7 +730,7 @@ func (s *session) handleWSPackage() error {
break
}
pkg, err = conn.read()
if netError, ok = err.(net.Error); ok && netError.Timeout() {
if netError, ok = jerrors.Cause(err).(net.Error); ok && netError.Timeout() {
continue
}
if err != nil {
......
version.go
View file @ 8eec657e
......@@ -10,9 +10,9 @@
package getty
const (
Version = "0.8.4"
DATE = "2018/06/25"
Version = "0.9.1"
DATE = "2018/06/30"
GETTY_MAJOR = 0
GETTY_MINOR = 8
GETTY_BUILD = 5
GETTY_MINOR = 9
GETTY_BUILD = 1
)
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment