commons_go/rpc
3
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

ing

Browse Source
This commit is contained in:
crusader 2018-03-22 18:08:37 +09:00
parent 945c47c19a
commit fccf18a26a
1 changed files with 23 additions and 36 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

59
servlet.go
View File

@ -34,7 +34,6 @@ type rpcServlet struct {
responseQueueChan chan *responseState
doneChan chan<- error
conn interface{}
serverCodec protocol.ServerCodec
@ -48,17 +47,17 @@ func (s *rpcServlet) Context() ServletContext {
func (s *rpcServlet) Start(parentCTX cuc.Context, conn interface{}, doneChan chan<- error) error {
if nil == s.sh {
panic("Servlet: servlet handler must be specified.")
return fmt.Errorf("RPC Servlet: servlet handler must be specified")
}
s.sh.Validate()
if nil == s.rwcSH {
panic("Servlet: servlet RWC handler must be specified.")
return fmt.Errorf("RPC Servlet: servlet RWC handler must be specified")
}
s.rwcSH.Validate()
if s.stopChan != nil {
return fmt.Errorf("Servlet: servlet is already running. Stop it before starting it again")
return fmt.Errorf("RPC Servlet: servlet is already running. Stop it before starting it again")
}
s.ctx = s.sh.ServletContext(parentCTX)
@ -67,19 +66,18 @@ func (s *rpcServlet) Start(parentCTX cuc.Context, conn interface{}, doneChan cha
return err
}
s.doneChan = doneChan
s.conn = conn
s.serverCodec = sc
if err := s.sh.Init(s.ctx); nil != err {
logging.Logger().Panicf("Servlet: Initialization of servlet has been failed %v", err)
return fmt.Errorf("RPC Servlet: Initialization of servlet has been failed %v", err)
}
s.responseQueueChan = make(chan *responseState, s.sh.GetPendingResponses())
s.stopChan = make(chan struct{})
s.stopWg.Add(1)
go handleServlet(s)
go handleServlet(s, doneChan)
return nil
}
@ -100,7 +98,7 @@ func (s *rpcServlet) Send(method string, args ...interface{}) (err error) {
func (s *rpcServlet) Stop() {
if s.stopChan == nil {
logging.Logger().Warnf("Server: server must be started before stopping it")
logging.Logger().Warnf("RPC Servlet: RPC Servlet must be started before stopping it")
return
}
close(s.stopChan)
@ -113,17 +111,17 @@ func (s *rpcServlet) Stop() {
s.conn = nil
s.serverCodec = nil
logging.Logger().Infof("Servlet is stopped")
logging.Logger().Infof("RPC Servlet: RPC Servlet is stopped")
}
func handleServlet(s *rpcServlet) {
func handleServlet(s *rpcServlet, doneChan chan<- error) {
var err error
logging.Logger().Infof("Servlet is started")
logging.Logger().Infof("RPC Servlet: RPC Servlet is started")
defer func() {
s.stopWg.Done()
s.doneChan <- err
doneChan <- err
}()
subStopChan := make(chan struct{})
@ -148,38 +146,32 @@ func handleServlet(s *rpcServlet) {
}
if err != nil {
logging.Logger().Errorf("RPC Server: servlet error %v", err)
logging.Logger().Errorf("RPC Servlet: servlet error %v", err)
}
}
func handleReader(s *rpcServlet, stopChan chan struct{}, doneChan chan error) {
logging.Logger().Debugf("reader of Servlet is started")
logging.Logger().Debugf("RPC Servlet: Reader of Servlet is started")
var err error
defer func() {
logging.Logger().Debugf("reader of Servlet is stopped")
if r := recover(); r != nil {
if err == nil {
err = fmt.Errorf("RPC Server: Panic when reading request from client: %v", r)
}
}
logging.Logger().Debugf("RPC Servlet: Reader of Servlet is stopped")
doneChan <- err
}()
for {
if nil == s.conn {
err = fmt.Errorf("RPC Server: disconnected from client")
err = fmt.Errorf("RPC Servlet: Disconnected from client")
return
}
requestCodec, err := s.rwcSH.ReadRequest(s.ctx, s.serverCodec, s.conn)
if nil != err {
if err == io.ErrUnexpectedEOF || err == io.EOF {
err = fmt.Errorf("RPC Server: disconnected from client")
err = fmt.Errorf("RPC Server: Disconnected from client")
return
}
logging.Logger().Errorf("RPC Server: Cannot read request: [%s]", err)
logging.Logger().Errorf("RPC Servlet: Cannot read request: [%s]", err)
continue
}
@ -188,7 +180,7 @@ func handleReader(s *rpcServlet, stopChan chan struct{}, doneChan chan error) {
select {
case <-stopChan:
err = fmt.Errorf("RPC Server: reading request stopped because get stop channel")
err = fmt.Errorf("RPC Servlet: Reading request stopped because get stop channel")
return
default:
}
@ -196,16 +188,11 @@ func handleReader(s *rpcServlet, stopChan chan struct{}, doneChan chan error) {
}
func handleWriter(s *rpcServlet, stopChan chan struct{}, doneChan chan error) {
logging.Logger().Debugf("writer of Servlet is started")
logging.Logger().Debugf("RPC Servlet: Writer of Servlet is started")
var err error
defer func() {
logging.Logger().Debugf("writer of Servlet is stopped")
if r := recover(); r != nil {
if err == nil {
err = fmt.Errorf("RPC Server: Panic when writing response to client: %v", r)
}
}
logging.Logger().Debugf("RPC Servlet: Writer of Servlet is stopped")
doneChan <- err
}()
@ -220,24 +207,24 @@ func handleWriter(s *rpcServlet, stopChan chan struct{}, doneChan chan error) {
select {
case <-stopChan:
err = fmt.Errorf("RPC Server: writing message stopped because get stop channel")
err = fmt.Errorf("RPC Servlet: Writing message stopped because get stop channel")
return
case rs = <-s.responseQueueChan:
}
}
if nil == s.conn {
err = fmt.Errorf("RPC Server: disconnected from client")
err = fmt.Errorf("RPC Servlet: Disconnected from client")
return
}
if nil != rs.requestCodec {
if err := s.rwcSH.WriteResponse(s.ctx, s.conn, rs.requestCodec, rs.result, rs.err); nil != err {
logging.Logger().Errorf("RPC Server: response error %v", err)
logging.Logger().Errorf("RPC Servlet: response error %v", err)
}
} else {
if err := s.rwcSH.WriteNotification(s.ctx, s.conn, s.serverCodec, rs.noti.method, rs.noti.args); nil != err {
logging.Logger().Errorf("RPC Server: notification error %v", err)
logging.Logger().Errorf("RPC Servlet: notification error %v", err)
}
}