call_test.go 7.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287
  1. /*
  2. *
  3. * Copyright 2014 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. "context"
  21. "fmt"
  22. "io"
  23. "math"
  24. "net"
  25. "strconv"
  26. "strings"
  27. "sync"
  28. "testing"
  29. "time"
  30. "google.golang.org/grpc/codes"
  31. "google.golang.org/grpc/internal/transport"
  32. "google.golang.org/grpc/status"
  33. )
  34. var (
  35. expectedRequest = "ping"
  36. expectedResponse = "pong"
  37. weirdError = "format verbs: %v%s"
  38. sizeLargeErr = 1024 * 1024
  39. canceled = 0
  40. )
  41. type testCodec struct {
  42. }
  43. func (testCodec) Marshal(v interface{}) ([]byte, error) {
  44. return []byte(*(v.(*string))), nil
  45. }
  46. func (testCodec) Unmarshal(data []byte, v interface{}) error {
  47. *(v.(*string)) = string(data)
  48. return nil
  49. }
  50. func (testCodec) String() string {
  51. return "test"
  52. }
  53. type testStreamHandler struct {
  54. port string
  55. t transport.ServerTransport
  56. }
  57. func (h *testStreamHandler) handleStream(t *testing.T, s *transport.Stream) {
  58. p := &parser{r: s}
  59. for {
  60. pf, req, err := p.recvMsg(math.MaxInt32)
  61. if err == io.EOF {
  62. break
  63. }
  64. if err != nil {
  65. return
  66. }
  67. if pf != compressionNone {
  68. t.Errorf("Received the mistaken message format %d, want %d", pf, compressionNone)
  69. return
  70. }
  71. var v string
  72. codec := testCodec{}
  73. if err := codec.Unmarshal(req, &v); err != nil {
  74. t.Errorf("Failed to unmarshal the received message: %v", err)
  75. return
  76. }
  77. if v == "weird error" {
  78. h.t.WriteStatus(s, status.New(codes.Internal, weirdError))
  79. return
  80. }
  81. if v == "canceled" {
  82. canceled++
  83. h.t.WriteStatus(s, status.New(codes.Internal, ""))
  84. return
  85. }
  86. if v == "port" {
  87. h.t.WriteStatus(s, status.New(codes.Internal, h.port))
  88. return
  89. }
  90. if v != expectedRequest {
  91. h.t.WriteStatus(s, status.New(codes.Internal, strings.Repeat("A", sizeLargeErr)))
  92. return
  93. }
  94. }
  95. // send a response back to end the stream.
  96. data, err := encode(testCodec{}, &expectedResponse)
  97. if err != nil {
  98. t.Errorf("Failed to encode the response: %v", err)
  99. return
  100. }
  101. hdr, payload := msgHeader(data, nil)
  102. h.t.Write(s, hdr, payload, &transport.Options{})
  103. h.t.WriteStatus(s, status.New(codes.OK, ""))
  104. }
  105. type server struct {
  106. lis net.Listener
  107. port string
  108. addr string
  109. startedErr chan error // sent nil or an error after server starts
  110. mu sync.Mutex
  111. conns map[transport.ServerTransport]bool
  112. }
  113. func newTestServer() *server {
  114. return &server{startedErr: make(chan error, 1)}
  115. }
  116. // start starts server. Other goroutines should block on s.startedErr for further operations.
  117. func (s *server) start(t *testing.T, port int, maxStreams uint32) {
  118. var err error
  119. if port == 0 {
  120. s.lis, err = net.Listen("tcp", "localhost:0")
  121. } else {
  122. s.lis, err = net.Listen("tcp", "localhost:"+strconv.Itoa(port))
  123. }
  124. if err != nil {
  125. s.startedErr <- fmt.Errorf("failed to listen: %v", err)
  126. return
  127. }
  128. s.addr = s.lis.Addr().String()
  129. _, p, err := net.SplitHostPort(s.addr)
  130. if err != nil {
  131. s.startedErr <- fmt.Errorf("failed to parse listener address: %v", err)
  132. return
  133. }
  134. s.port = p
  135. s.conns = make(map[transport.ServerTransport]bool)
  136. s.startedErr <- nil
  137. for {
  138. conn, err := s.lis.Accept()
  139. if err != nil {
  140. return
  141. }
  142. config := &transport.ServerConfig{
  143. MaxStreams: maxStreams,
  144. }
  145. st, err := transport.NewServerTransport("http2", conn, config)
  146. if err != nil {
  147. continue
  148. }
  149. s.mu.Lock()
  150. if s.conns == nil {
  151. s.mu.Unlock()
  152. st.Close()
  153. return
  154. }
  155. s.conns[st] = true
  156. s.mu.Unlock()
  157. h := &testStreamHandler{
  158. port: s.port,
  159. t: st,
  160. }
  161. go st.HandleStreams(func(s *transport.Stream) {
  162. go h.handleStream(t, s)
  163. }, func(ctx context.Context, method string) context.Context {
  164. return ctx
  165. })
  166. }
  167. }
  168. func (s *server) wait(t *testing.T, timeout time.Duration) {
  169. select {
  170. case err := <-s.startedErr:
  171. if err != nil {
  172. t.Fatal(err)
  173. }
  174. case <-time.After(timeout):
  175. t.Fatalf("Timed out after %v waiting for server to be ready", timeout)
  176. }
  177. }
  178. func (s *server) stop() {
  179. s.lis.Close()
  180. s.mu.Lock()
  181. for c := range s.conns {
  182. c.Close()
  183. }
  184. s.conns = nil
  185. s.mu.Unlock()
  186. }
  187. func setUp(t *testing.T, port int, maxStreams uint32) (*server, *ClientConn) {
  188. server := newTestServer()
  189. go server.start(t, port, maxStreams)
  190. server.wait(t, 2*time.Second)
  191. addr := "localhost:" + server.port
  192. cc, err := Dial(addr, WithBlock(), WithInsecure(), WithCodec(testCodec{}))
  193. if err != nil {
  194. t.Fatalf("Failed to create ClientConn: %v", err)
  195. }
  196. return server, cc
  197. }
  198. func (s) TestInvoke(t *testing.T) {
  199. server, cc := setUp(t, 0, math.MaxUint32)
  200. var reply string
  201. if err := cc.Invoke(context.Background(), "/foo/bar", &expectedRequest, &reply); err != nil || reply != expectedResponse {
  202. t.Fatalf("grpc.Invoke(_, _, _, _, _) = %v, want <nil>", err)
  203. }
  204. cc.Close()
  205. server.stop()
  206. }
  207. func (s) TestInvokeLargeErr(t *testing.T) {
  208. server, cc := setUp(t, 0, math.MaxUint32)
  209. var reply string
  210. req := "hello"
  211. err := cc.Invoke(context.Background(), "/foo/bar", &req, &reply)
  212. if _, ok := status.FromError(err); !ok {
  213. t.Fatalf("grpc.Invoke(_, _, _, _, _) receives non rpc error.")
  214. }
  215. if status.Code(err) != codes.Internal || len(errorDesc(err)) != sizeLargeErr {
  216. t.Fatalf("grpc.Invoke(_, _, _, _, _) = %v, want an error of code %d and desc size %d", err, codes.Internal, sizeLargeErr)
  217. }
  218. cc.Close()
  219. server.stop()
  220. }
  221. // TestInvokeErrorSpecialChars checks that error messages don't get mangled.
  222. func (s) TestInvokeErrorSpecialChars(t *testing.T) {
  223. server, cc := setUp(t, 0, math.MaxUint32)
  224. var reply string
  225. req := "weird error"
  226. err := cc.Invoke(context.Background(), "/foo/bar", &req, &reply)
  227. if _, ok := status.FromError(err); !ok {
  228. t.Fatalf("grpc.Invoke(_, _, _, _, _) receives non rpc error.")
  229. }
  230. if got, want := errorDesc(err), weirdError; got != want {
  231. t.Fatalf("grpc.Invoke(_, _, _, _, _) error = %q, want %q", got, want)
  232. }
  233. cc.Close()
  234. server.stop()
  235. }
  236. // TestInvokeCancel checks that an Invoke with a canceled context is not sent.
  237. func (s) TestInvokeCancel(t *testing.T) {
  238. server, cc := setUp(t, 0, math.MaxUint32)
  239. var reply string
  240. req := "canceled"
  241. for i := 0; i < 100; i++ {
  242. ctx, cancel := context.WithCancel(context.Background())
  243. cancel()
  244. cc.Invoke(ctx, "/foo/bar", &req, &reply)
  245. }
  246. if canceled != 0 {
  247. t.Fatalf("received %d of 100 canceled requests", canceled)
  248. }
  249. cc.Close()
  250. server.stop()
  251. }
  252. // TestInvokeCancelClosedNonFail checks that a canceled non-failfast RPC
  253. // on a closed client will terminate.
  254. func (s) TestInvokeCancelClosedNonFailFast(t *testing.T) {
  255. server, cc := setUp(t, 0, math.MaxUint32)
  256. var reply string
  257. cc.Close()
  258. req := "hello"
  259. ctx, cancel := context.WithCancel(context.Background())
  260. cancel()
  261. if err := cc.Invoke(ctx, "/foo/bar", &req, &reply, WaitForReady(true)); err == nil {
  262. t.Fatalf("canceled invoke on closed connection should fail")
  263. }
  264. server.stop()
  265. }