commons/rpc-go
3
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

ing

Browse Source
This commit is contained in:
crusader 2018-04-12 10:55:29 +09:00
parent 903c790624
commit 4691a6923a
1 changed files with 14 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
client/client.go
View File

@ -30,7 +30,7 @@ type Client struct {
pendingRequests sync.Map pendingRequests sync.Map
} }
func (c *Client) Start(readChan <-chan []byte, writeChan chan<- []byte) error { func (c *Client) Start() error {
if c.stopChan != nil { if c.stopChan != nil {
return fmt.Errorf("%s already running. Stop it before starting it again", c.logHeader()) return fmt.Errorf("%s already running. Stop it before starting it again", c.logHeader())
} }
@ -47,6 +47,11 @@ func (c *Client) Start(readChan <-chan []byte, writeChan chan<- []byte) error {
return fmt.Errorf("%s Codec is not valid", c.logHeader()) return fmt.Errorf("%s Codec is not valid", c.logHeader())
} }
readChan, writeChan, err := c.Connector.Connect()
if nil != err {
return err
}
c.stopChan = make(chan struct{}) c.stopChan = make(chan struct{})
c.stopWg.Add(1) c.stopWg.Add(1)
@ -158,6 +163,12 @@ func (c *Client) internalSend(hasResponse bool, result interface{}, method strin
func (c *Client) handleClient(readChan <-chan []byte, writeChan chan<- []byte) { func (c *Client) handleClient(readChan <-chan []byte, writeChan chan<- []byte) {
defer func() { defer func() {
if err := c.Connector.Disconnect(); nil != err {
logging.Logger().Warn(err)
}
logging.Logger().Infof("%s Stopped", c.logHeader())
c.stopWg.Done() c.stopWg.Done()
}() }()
@ -168,6 +179,8 @@ func (c *Client) handleClient(readChan <-chan []byte, writeChan chan<- []byte) {
go c.handleSend(stopChan, sendDoneChan, writeChan) go c.handleSend(stopChan, sendDoneChan, writeChan)
go c.handleReceive(stopChan, receiveDoneChan, readChan) go c.handleReceive(stopChan, receiveDoneChan, readChan)
logging.Logger().Infof("%s Started", c.logHeader())
select { select {
case <-sendDoneChan: case <-sendDoneChan:
close(stopChan) close(stopChan)