resolver_conn_wrapper.go 4.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158
  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. package grpc
  19. import (
  20. "fmt"
  21. "strings"
  22. "google.golang.org/grpc/grpclog"
  23. "google.golang.org/grpc/resolver"
  24. )
  25. // ccResolverWrapper is a wrapper on top of cc for resolvers.
  26. // It implements resolver.ClientConnection interface.
  27. type ccResolverWrapper struct {
  28. cc *ClientConn
  29. resolver resolver.Resolver
  30. addrCh chan []resolver.Address
  31. scCh chan string
  32. done chan struct{}
  33. }
  34. // split2 returns the values from strings.SplitN(s, sep, 2).
  35. // If sep is not found, it returns ("", s, false) instead.
  36. func split2(s, sep string) (string, string, bool) {
  37. spl := strings.SplitN(s, sep, 2)
  38. if len(spl) < 2 {
  39. return "", "", false
  40. }
  41. return spl[0], spl[1], true
  42. }
  43. // parseTarget splits target into a struct containing scheme, authority and
  44. // endpoint.
  45. //
  46. // If target is not a valid scheme://authority/endpoint, it returns {Endpoint:
  47. // target}.
  48. func parseTarget(target string) (ret resolver.Target) {
  49. var ok bool
  50. ret.Scheme, ret.Endpoint, ok = split2(target, "://")
  51. if !ok {
  52. return resolver.Target{Endpoint: target}
  53. }
  54. ret.Authority, ret.Endpoint, ok = split2(ret.Endpoint, "/")
  55. if !ok {
  56. return resolver.Target{Endpoint: target}
  57. }
  58. return ret
  59. }
  60. // newCCResolverWrapper parses cc.target for scheme and gets the resolver
  61. // builder for this scheme and builds the resolver. The monitoring goroutine
  62. // for it is not started yet and can be created by calling start().
  63. //
  64. // If withResolverBuilder dial option is set, the specified resolver will be
  65. // used instead.
  66. func newCCResolverWrapper(cc *ClientConn) (*ccResolverWrapper, error) {
  67. rb := cc.dopts.resolverBuilder
  68. if rb == nil {
  69. return nil, fmt.Errorf("could not get resolver for scheme: %q", cc.parsedTarget.Scheme)
  70. }
  71. ccr := &ccResolverWrapper{
  72. cc: cc,
  73. addrCh: make(chan []resolver.Address, 1),
  74. scCh: make(chan string, 1),
  75. done: make(chan struct{}),
  76. }
  77. var err error
  78. ccr.resolver, err = rb.Build(cc.parsedTarget, ccr, resolver.BuildOption{DisableServiceConfig: cc.dopts.disableServiceConfig})
  79. if err != nil {
  80. return nil, err
  81. }
  82. return ccr, nil
  83. }
  84. func (ccr *ccResolverWrapper) start() {
  85. go ccr.watcher()
  86. }
  87. // watcher processes address updates and service config updates sequentially.
  88. // Otherwise, we need to resolve possible races between address and service
  89. // config (e.g. they specify different balancer types).
  90. func (ccr *ccResolverWrapper) watcher() {
  91. for {
  92. select {
  93. case <-ccr.done:
  94. return
  95. default:
  96. }
  97. select {
  98. case addrs := <-ccr.addrCh:
  99. select {
  100. case <-ccr.done:
  101. return
  102. default:
  103. }
  104. grpclog.Infof("ccResolverWrapper: sending new addresses to cc: %v", addrs)
  105. ccr.cc.handleResolvedAddrs(addrs, nil)
  106. case sc := <-ccr.scCh:
  107. select {
  108. case <-ccr.done:
  109. return
  110. default:
  111. }
  112. grpclog.Infof("ccResolverWrapper: got new service config: %v", sc)
  113. ccr.cc.handleServiceConfig(sc)
  114. case <-ccr.done:
  115. return
  116. }
  117. }
  118. }
  119. func (ccr *ccResolverWrapper) resolveNow(o resolver.ResolveNowOption) {
  120. ccr.resolver.ResolveNow(o)
  121. }
  122. func (ccr *ccResolverWrapper) close() {
  123. ccr.resolver.Close()
  124. close(ccr.done)
  125. }
  126. // NewAddress is called by the resolver implemenetion to send addresses to gRPC.
  127. func (ccr *ccResolverWrapper) NewAddress(addrs []resolver.Address) {
  128. select {
  129. case <-ccr.addrCh:
  130. default:
  131. }
  132. ccr.addrCh <- addrs
  133. }
  134. // NewServiceConfig is called by the resolver implemenetion to send service
  135. // configs to gRPC.
  136. func (ccr *ccResolverWrapper) NewServiceConfig(sc string) {
  137. select {
  138. case <-ccr.scCh:
  139. default:
  140. }
  141. ccr.scCh <- sc
  142. }