improve code quality

This commit is contained in:
Astone 2019-12-03 15:25:58 +08:00
parent b5d65305db
commit 29fb58db39
20 changed files with 67 additions and 89 deletions

View File

@ -44,11 +44,9 @@ func (tc *telegramConn) run() {
tc.recv = updates tc.recv = updates
tc.syncCond.Signal() tc.syncCond.Signal()
for { select {
select { case <-tc.exit:
case <-tc.exit: return
return
}
} }
} }

View File

@ -339,7 +339,9 @@ func (g *grpcClient) Call(ctx context.Context, req client.Request, rsp interface
d, ok := ctx.Deadline() d, ok := ctx.Deadline()
if !ok { if !ok {
// no deadline so we create a new one // no deadline so we create a new one
ctx, _ = context.WithTimeout(ctx, callOpts.RequestTimeout) var cancel context.CancelFunc
ctx, cancel = context.WithTimeout(ctx, callOpts.RequestTimeout)
defer cancel()
} else { } else {
// got a deadline so no need to setup context // got a deadline so no need to setup context
// but we need to set the timeout we pass along // but we need to set the timeout we pass along

View File

@ -162,7 +162,6 @@ func (r *rpcClient) call(ctx context.Context, node *registry.Node, req Request,
select { select {
case err := <-ch: case err := <-ch:
grr = err
return err return err
case <-ctx.Done(): case <-ctx.Done():
grr = errors.Timeout("go.micro.client", fmt.Sprintf("%v", ctx.Err())) grr = errors.Timeout("go.micro.client", fmt.Sprintf("%v", ctx.Err()))
@ -378,7 +377,9 @@ func (r *rpcClient) Call(ctx context.Context, request Request, response interfac
d, ok := ctx.Deadline() d, ok := ctx.Deadline()
if !ok { if !ok {
// no deadline so we create a new one // no deadline so we create a new one
ctx, _ = context.WithTimeout(ctx, callOpts.RequestTimeout) var cancel context.CancelFunc
ctx, cancel = context.WithTimeout(ctx, callOpts.RequestTimeout)
defer cancel()
} else { } else {
// got a deadline so no need to setup context // got a deadline so no need to setup context
// but we need to set the timeout we pass along // but we need to set the timeout we pass along

View File

@ -29,12 +29,10 @@ func (c *Codec) ReadBody(b interface{}) error {
return err return err
} }
switch b.(type) { switch v := b.(type) {
case *[]byte: case *[]byte:
v := b.(*[]byte)
*v = buf *v = buf
case *Frame: case *Frame:
v := b.(*Frame)
v.Data = buf v.Data = buf
default: default:
return fmt.Errorf("failed to read body: %v is not type of *[]byte", b) return fmt.Errorf("failed to read body: %v is not type of *[]byte", b)
@ -45,14 +43,13 @@ func (c *Codec) ReadBody(b interface{}) error {
func (c *Codec) Write(m *codec.Message, b interface{}) error { func (c *Codec) Write(m *codec.Message, b interface{}) error {
var v []byte var v []byte
switch b.(type) { switch vb := b.(type) {
case *Frame: case *Frame:
v = b.(*Frame).Data v = vb.Data
case *[]byte: case *[]byte:
ve := b.(*[]byte) v = *vb
v = *ve
case []byte: case []byte:
v = b.([]byte) v = vb
default: default:
return fmt.Errorf("failed to write: %v is not type of *[]byte or []byte", b) return fmt.Errorf("failed to write: %v is not type of *[]byte or []byte", b)
} }

View File

@ -12,25 +12,22 @@ type Message struct {
} }
func (n Marshaler) Marshal(v interface{}) ([]byte, error) { func (n Marshaler) Marshal(v interface{}) ([]byte, error) {
switch v.(type) { switch ve := v.(type) {
case *[]byte: case *[]byte:
ve := v.(*[]byte)
return *ve, nil return *ve, nil
case []byte: case []byte:
return v.([]byte), nil return ve, nil
case *Message: case *Message:
return v.(*Message).Body, nil return ve.Body, nil
} }
return nil, errors.New("invalid message") return nil, errors.New("invalid message")
} }
func (n Marshaler) Unmarshal(d []byte, v interface{}) error { func (n Marshaler) Unmarshal(d []byte, v interface{}) error {
switch v.(type) { switch ve := v.(type) {
case *[]byte: case *[]byte:
ve := v.(*[]byte)
*ve = d *ve = d
case *Message: case *Message:
ve := v.(*Message)
ve.Body = d ve.Body = d
} }
return errors.New("invalid message") return errors.New("invalid message")

View File

@ -29,15 +29,12 @@ func (c *Codec) ReadBody(b interface{}) error {
return err return err
} }
switch b.(type) { switch v := b.(type) {
case *string: case *string:
v := b.(*string)
*v = string(buf) *v = string(buf)
case *[]byte: case *[]byte:
v := b.(*[]byte)
*v = buf *v = buf
case *Frame: case *Frame:
v := b.(*Frame)
v.Data = buf v.Data = buf
default: default:
return fmt.Errorf("failed to read body: %v is not type of *[]byte", b) return fmt.Errorf("failed to read body: %v is not type of *[]byte", b)
@ -48,20 +45,17 @@ func (c *Codec) ReadBody(b interface{}) error {
func (c *Codec) Write(m *codec.Message, b interface{}) error { func (c *Codec) Write(m *codec.Message, b interface{}) error {
var v []byte var v []byte
switch b.(type) { switch ve := b.(type) {
case *Frame: case *Frame:
v = b.(*Frame).Data v = ve.Data
case *[]byte: case *[]byte:
ve := b.(*[]byte)
v = *ve v = *ve
case *string: case *string:
ve := b.(*string)
v = []byte(*ve) v = []byte(*ve)
case string: case string:
ve := b.(string)
v = []byte(ve) v = []byte(ve)
case []byte: case []byte:
v = b.([]byte) v = ve
default: default:
return fmt.Errorf("failed to write: %v is not type of *[]byte or []byte", b) return fmt.Errorf("failed to write: %v is not type of *[]byte or []byte", b)
} }

View File

@ -823,7 +823,7 @@ func (n *network) processCtrlChan(listener tunnel.Listener) {
log.Tracef("Network metric for router %s and gateway %s: %v", event.Route.Router, event.Route.Gateway, metric) log.Tracef("Network metric for router %s and gateway %s: %v", event.Route.Router, event.Route.Gateway, metric)
// check we don't overflow max int 64 // check we don't overflow max int 64
if d := route.Metric + metric; d > math.MaxInt64 || d <= 0 { if d := route.Metric + metric; d <= 0 {
// set to max int64 if we overflow // set to max int64 if we overflow
route.Metric = math.MaxInt64 route.Metric = math.MaxInt64
} else { } else {

View File

@ -94,11 +94,6 @@ func (n *Network) Connect(ctx context.Context, req *pbNet.ConnectRequest, resp *
// Nodes returns the list of nodes // Nodes returns the list of nodes
func (n *Network) Nodes(ctx context.Context, req *pbNet.NodesRequest, resp *pbNet.NodesResponse) error { func (n *Network) Nodes(ctx context.Context, req *pbNet.NodesRequest, resp *pbNet.NodesResponse) error {
depth := uint(req.Depth)
if depth <= 0 || depth > network.MaxDepth {
depth = network.MaxDepth
}
// root node // root node
nodes := map[string]network.Node{} nodes := map[string]network.Node{}

View File

@ -274,7 +274,7 @@ func (p *Proxy) watchRoutes() {
return return
} }
if err := p.manageRoutes(event.Route, fmt.Sprintf("%s", event.Type)); err != nil { if err := p.manageRoutes(event.Route, event.Type.String()); err != nil {
// TODO: should we bail here? // TODO: should we bail here?
continue continue
} }
@ -562,12 +562,9 @@ func NewProxy(opts ...options.Option) proxy.Proxy {
defer t.Stop() defer t.Stop()
// we must refresh route metrics since they do not trigger new events // we must refresh route metrics since they do not trigger new events
for { for range t.C {
select { // refresh route metrics
case <-t.C: p.refreshMetrics()
// refresh route metrics
p.refreshMetrics()
}
} }
}() }()

View File

@ -230,7 +230,9 @@ func (m *mdnsRegistry) GetService(service string) ([]*Service, error) {
p := mdns.DefaultParams(service) p := mdns.DefaultParams(service)
// set context with timeout // set context with timeout
p.Context, _ = context.WithTimeout(context.Background(), m.opts.Timeout) var cancel context.CancelFunc
p.Context, cancel = context.WithTimeout(context.Background(), m.opts.Timeout)
defer cancel()
// set entries channel // set entries channel
p.Entries = entries p.Entries = entries
// set the domain // set the domain
@ -308,7 +310,9 @@ func (m *mdnsRegistry) ListServices() ([]*Service, error) {
p := mdns.DefaultParams("_services") p := mdns.DefaultParams("_services")
// set context with timeout // set context with timeout
p.Context, _ = context.WithTimeout(context.Background(), m.opts.Timeout) var cancel context.CancelFunc
p.Context, cancel = context.WithTimeout(context.Background(), m.opts.Timeout)
defer cancel()
// set entries channel // set entries channel
p.Entries = entries p.Entries = entries
// set domain // set domain

View File

@ -11,24 +11,22 @@ type serviceWatcher struct {
} }
func (s *serviceWatcher) Next() (*registry.Result, error) { func (s *serviceWatcher) Next() (*registry.Result, error) {
for { // check if closed
// check if closed select {
select { case <-s.closed:
case <-s.closed: return nil, registry.ErrWatcherStopped
return nil, registry.ErrWatcherStopped default:
default:
}
r, err := s.stream.Recv()
if err != nil {
return nil, err
}
return &registry.Result{
Action: r.Action,
Service: ToService(r.Service),
}, nil
} }
r, err := s.stream.Recv()
if err != nil {
return nil, err
}
return &registry.Result{
Action: r.Action,
Service: ToService(r.Service),
}, nil
} }
func (s *serviceWatcher) Stop() { func (s *serviceWatcher) Stop() {

View File

@ -718,7 +718,7 @@ func (r *router) Process(a *Advert) error {
route := event.Route route := event.Route
action := event.Type action := event.Type
log.Debugf("Router %s applying %s from router %s for service %s %s", r.options.Id, action, route.Router, route.Service, route.Address) log.Debugf("Router %s applying %s from router %s for service %s %s", r.options.Id, action, route.Router, route.Service, route.Address)
if err := r.manageRoute(route, fmt.Sprintf("%s", action)); err != nil { if err := r.manageRoute(route, action.String()); err != nil {
return fmt.Errorf("failed applying action %s to routing table: %s", action, err) return fmt.Errorf("failed applying action %s to routing table: %s", action, err)
} }
} }

View File

@ -222,7 +222,9 @@ func (g *grpcServer) handler(srv interface{}, stream grpc.ServerStream) error {
// set the timeout if we have it // set the timeout if we have it
if len(to) > 0 { if len(to) > 0 {
if n, err := strconv.ParseUint(to, 10, 64); err == nil { if n, err := strconv.ParseUint(to, 10, 64); err == nil {
ctx, _ = context.WithTimeout(ctx, time.Duration(n)) var cancel context.CancelFunc
ctx, cancel = context.WithTimeout(ctx, time.Duration(n))
defer cancel()
} }
} }

View File

@ -277,7 +277,9 @@ func (s *rpcServer) ServeConn(sock transport.Socket) {
// set the timeout from the header if we have it // set the timeout from the header if we have it
if len(to) > 0 { if len(to) > 0 {
if n, err := strconv.ParseUint(to, 10, 64); err == nil { if n, err := strconv.ParseUint(to, 10, 64); err == nil {
ctx, _ = context.WithTimeout(ctx, time.Duration(n)) var cancel context.CancelFunc
ctx, cancel = context.WithTimeout(ctx, time.Duration(n))
defer cancel()
} }
} }

View File

@ -201,7 +201,7 @@ func Run() error {
} }
ch := make(chan os.Signal, 1) ch := make(chan os.Signal, 1)
signal.Notify(ch, syscall.SIGTERM, syscall.SIGINT, syscall.SIGKILL) signal.Notify(ch, syscall.SIGTERM, syscall.SIGINT)
log.Logf("Received signal %s", <-ch) log.Logf("Received signal %s", <-ch)
return Stop() return Stop()

View File

@ -39,9 +39,7 @@ func (e *etcdLeader) Elect(id string, opts ...leader.ElectOption) (leader.Electe
l := cc.NewElection(s, path) l := cc.NewElection(s, path)
ctx, _ := context.WithCancel(context.Background()) if err := l.Campaign(context.TODO(), id); err != nil {
if err := l.Campaign(ctx, id); err != nil {
return nil, err return nil, err
} }
@ -63,14 +61,8 @@ func (e *etcdLeader) Follow() chan string {
ech := l.Observe(context.Background()) ech := l.Observe(context.Background())
go func() { go func() {
for { for r := range ech {
select { ch <- string(r.Kvs[0].Value)
case r, ok := <-ech:
if !ok {
return
}
ch <- string(r.Kvs[0].Value)
}
} }
}() }()
@ -82,8 +74,7 @@ func (e *etcdLeader) String() string {
} }
func (e *etcdElected) Reelect() error { func (e *etcdElected) Reelect() error {
ctx, _ := context.WithCancel(context.Background()) return e.e.Campaign(context.TODO(), e.id)
return e.e.Campaign(ctx, e.id)
} }
func (e *etcdElected) Revoked() chan bool { func (e *etcdElected) Revoked() chan bool {

View File

@ -49,9 +49,7 @@ func (e *etcdLock) Acquire(id string, opts ...lock.AcquireOption) error {
m := cc.NewMutex(s, path) m := cc.NewMutex(s, path)
ctx, _ := context.WithCancel(context.Background()) if err := m.Lock(context.TODO()); err != nil {
if err := m.Lock(ctx); err != nil {
return err return err
} }

View File

@ -63,7 +63,9 @@ func (s Schedule) Run() <-chan time.Time {
} }
// start ticker // start ticker
for t := range time.Tick(s.Interval) { ticker := time.NewTicker(s.Interval)
defer ticker.Stop()
for t := range ticker.C {
ch <- t ch <- t
} }
}() }()

View File

@ -228,7 +228,7 @@ func (l *link) manage() {
} }
// check the type of message // check the type of message
switch { switch {
case bytes.Compare(p.message.Body, linkRequest) == 0: case bytes.Equal(p.message.Body, linkRequest):
log.Tracef("Link %s received link request %v", l.id, p.message.Body) log.Tracef("Link %s received link request %v", l.id, p.message.Body)
// send response // send response
if err := send(linkResponse); err != nil { if err := send(linkResponse); err != nil {
@ -236,7 +236,7 @@ func (l *link) manage() {
l.errCount++ l.errCount++
l.Unlock() l.Unlock()
} }
case bytes.Compare(p.message.Body, linkResponse) == 0: case bytes.Equal(p.message.Body, linkResponse):
// set round trip time // set round trip time
d := time.Since(now) d := time.Since(now)
log.Tracef("Link %s received link response in %v", p.message.Body, d) log.Tracef("Link %s received link response in %v", p.message.Body, d)

View File

@ -376,7 +376,7 @@ func (s *service) Run() error {
go s.run(ex) go s.run(ex)
ch := make(chan os.Signal, 1) ch := make(chan os.Signal, 1)
signal.Notify(ch, syscall.SIGTERM, syscall.SIGINT, syscall.SIGKILL) signal.Notify(ch, syscall.SIGTERM, syscall.SIGINT)
select { select {
// wait on kill signal // wait on kill signal