vendor/google.golang.org/grpc/resolver_conn_wrapper.go

   1 /*
   2  *
   3  * Copyright 2017 gRPC authors.
   4  *
   5  * Licensed under the Apache License, Version 2.0 (the "License");
   6  * you may not use this file except in compliance with the License.
   7  * You may obtain a copy of the License at
   8  *
   9  *     http://www.apache.org/licenses/LICENSE-2.0
  10  *
  11  * Unless required by applicable law or agreed to in writing, software
  12  * distributed under the License is distributed on an "AS IS" BASIS,
  13  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14  * See the License for the specific language governing permissions and
  15  * limitations under the License.
  16  *
  17  */
  18
  19 package grpc
  20
  21 import (
  22         "fmt"
  23         "strings"
  24
  25         "google.golang.org/grpc/grpclog"
  26         "google.golang.org/grpc/resolver"
  27 )
  28
  29 // ccResolverWrapper is a wrapper on top of cc for resolvers.
  30 // It implements resolver.ClientConnection interface.
  31 type ccResolverWrapper struct {
  32         cc       *ClientConn
  33         resolver resolver.Resolver
  34         addrCh   chan []resolver.Address
  35         scCh     chan string
  36         done     chan struct{}
  37 }
  38
  39 // split2 returns the values from strings.SplitN(s, sep, 2).
  40 // If sep is not found, it returns ("", s, false) instead.
  41 func split2(s, sep string) (string, string, bool) {
  42         spl := strings.SplitN(s, sep, 2)
  43         if len(spl) < 2 {
  44                 return "", "", false
  45         }
  46         return spl[0], spl[1], true
  47 }
  48
  49 // parseTarget splits target into a struct containing scheme, authority and
  50 // endpoint.
  51 func parseTarget(target string) (ret resolver.Target) {
  52         var ok bool
  53         ret.Scheme, ret.Endpoint, ok = split2(target, "://")
  54         if !ok {
  55                 return resolver.Target{Endpoint: target}
  56         }
  57         ret.Authority, ret.Endpoint, _ = split2(ret.Endpoint, "/")
  58         return ret
  59 }
  60
  61 // newCCResolverWrapper parses cc.target for scheme and gets the resolver
  62 // builder for this scheme. It then builds the resolver and starts the
  63 // monitoring goroutine for it.
  64 func newCCResolverWrapper(cc *ClientConn) (*ccResolverWrapper, error) {
  65         grpclog.Infof("dialing to target with scheme: %q", cc.parsedTarget.Scheme)
  66
  67         rb := resolver.Get(cc.parsedTarget.Scheme)
  68         if rb == nil {
  69                 return nil, fmt.Errorf("could not get resolver for scheme: %q", cc.parsedTarget.Scheme)
  70         }
  71
  72         ccr := &ccResolverWrapper{
  73                 cc:     cc,
  74                 addrCh: make(chan []resolver.Address, 1),
  75                 scCh:   make(chan string, 1),
  76                 done:   make(chan struct{}),
  77         }
  78
  79         var err error
  80         ccr.resolver, err = rb.Build(cc.parsedTarget, ccr, resolver.BuildOption{})
  81         if err != nil {
  82                 return nil, err
  83         }
  84         go ccr.watcher()
  85         return ccr, nil
  86 }
  87
  88 // watcher processes address updates and service config updates sequencially.
  89 // Otherwise, we need to resolve possible races between address and service
  90 // config (e.g. they specify different balancer types).
  91 func (ccr *ccResolverWrapper) watcher() {
  92         for {
  93                 select {
  94                 case <-ccr.done:
  95                         return
  96                 default:
  97                 }
  98
  99                 select {
 100                 case addrs := <-ccr.addrCh:
 101                         select {
 102                         case <-ccr.done:
 103                                 return
 104                         default:
 105                         }
 106                         grpclog.Infof("ccResolverWrapper: sending new addresses to cc: %v", addrs)
 107                         ccr.cc.handleResolvedAddrs(addrs, nil)
 108                 case sc := <-ccr.scCh:
 109                         select {
 110                         case <-ccr.done:
 111                                 return
 112                         default:
 113                         }
 114                         grpclog.Infof("ccResolverWrapper: got new service config: %v", sc)
 115                         ccr.cc.handleServiceConfig(sc)
 116                 case <-ccr.done:
 117                         return
 118                 }
 119         }
 120 }
 121
 122 func (ccr *ccResolverWrapper) close() {
 123         ccr.resolver.Close()
 124         close(ccr.done)
 125 }
 126
 127 // NewAddress is called by the resolver implemenetion to send addresses to gRPC.
 128 func (ccr *ccResolverWrapper) NewAddress(addrs []resolver.Address) {
 129         select {
 130         case <-ccr.addrCh:
 131         default:
 132         }
 133         ccr.addrCh <- addrs
 134 }
 135
 136 // NewServiceConfig is called by the resolver implemenetion to send service
 137 // configs to gPRC.
 138 func (ccr *ccResolverWrapper) NewServiceConfig(sc string) {
 139         select {
 140         case <-ccr.scCh:
 141         default:
 142         }
 143         ccr.scCh <- sc
 144 }