diff --git a/client.go b/client.go index c89c177..9ca25b7 100644 --- a/client.go +++ b/client.go @@ -2,12 +2,13 @@ package srpc import ( //"bytes" - //"errors" + "errors" //"fmt" "io" "net" "sync" - //"time" + "sync/atomic" + "time" ) //func (client *Client) Call(funcName string, args ...interface{}) (ret []byte, err error) { @@ -137,3 +138,42 @@ func (c *RPCClient) Stop() { c.stopWg.Wait() c.stopChan = nil } + +func clientHandler(c *RPCClient) { + defer c.stopWg.Done() + + var conn io.ReadWriteCloser + var err error + var stopping atomic.Value + + for { + dialChan := make(chan struct{}) + go func() { + if conn, err = c.DialHandler(c.Addr); err != nil { + if stopping.Load() == nil { + c.LogError("srpc - '%s' cannoc estable rpc connection: '%s'\n", c.Addr, err) + } + } + close(dialChan) + }() + + select { + case <-c.stopChan: + stopping.Store(true) + <-dialChan + return + case <-dialChan: + } + + if err != nil { + select { + case <-c.stopChan: + return + case <-time.After(time.Second): + } + continue + } + + c.stopWg.Add(1) + } +}