rpc_util.go 24 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815
  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. "bytes"
  21. "compress/gzip"
  22. "context"
  23. "encoding/binary"
  24. "fmt"
  25. "io"
  26. "io/ioutil"
  27. "math"
  28. "net/url"
  29. "strings"
  30. "sync"
  31. "time"
  32. "google.golang.org/grpc/codes"
  33. "google.golang.org/grpc/credentials"
  34. "google.golang.org/grpc/encoding"
  35. "google.golang.org/grpc/encoding/proto"
  36. "google.golang.org/grpc/internal/transport"
  37. "google.golang.org/grpc/metadata"
  38. "google.golang.org/grpc/peer"
  39. "google.golang.org/grpc/stats"
  40. "google.golang.org/grpc/status"
  41. )
  42. // Compressor defines the interface gRPC uses to compress a message.
  43. //
  44. // Deprecated: use package encoding.
  45. type Compressor interface {
  46. // Do compresses p into w.
  47. Do(w io.Writer, p []byte) error
  48. // Type returns the compression algorithm the Compressor uses.
  49. Type() string
  50. }
  51. type gzipCompressor struct {
  52. pool sync.Pool
  53. }
  54. // NewGZIPCompressor creates a Compressor based on GZIP.
  55. //
  56. // Deprecated: use package encoding/gzip.
  57. func NewGZIPCompressor() Compressor {
  58. c, _ := NewGZIPCompressorWithLevel(gzip.DefaultCompression)
  59. return c
  60. }
  61. // NewGZIPCompressorWithLevel is like NewGZIPCompressor but specifies the gzip compression level instead
  62. // of assuming DefaultCompression.
  63. //
  64. // The error returned will be nil if the level is valid.
  65. //
  66. // Deprecated: use package encoding/gzip.
  67. func NewGZIPCompressorWithLevel(level int) (Compressor, error) {
  68. if level < gzip.DefaultCompression || level > gzip.BestCompression {
  69. return nil, fmt.Errorf("grpc: invalid compression level: %d", level)
  70. }
  71. return &gzipCompressor{
  72. pool: sync.Pool{
  73. New: func() interface{} {
  74. w, err := gzip.NewWriterLevel(ioutil.Discard, level)
  75. if err != nil {
  76. panic(err)
  77. }
  78. return w
  79. },
  80. },
  81. }, nil
  82. }
  83. func (c *gzipCompressor) Do(w io.Writer, p []byte) error {
  84. z := c.pool.Get().(*gzip.Writer)
  85. defer c.pool.Put(z)
  86. z.Reset(w)
  87. if _, err := z.Write(p); err != nil {
  88. return err
  89. }
  90. return z.Close()
  91. }
  92. func (c *gzipCompressor) Type() string {
  93. return "gzip"
  94. }
  95. // Decompressor defines the interface gRPC uses to decompress a message.
  96. //
  97. // Deprecated: use package encoding.
  98. type Decompressor interface {
  99. // Do reads the data from r and uncompress them.
  100. Do(r io.Reader) ([]byte, error)
  101. // Type returns the compression algorithm the Decompressor uses.
  102. Type() string
  103. }
  104. type gzipDecompressor struct {
  105. pool sync.Pool
  106. }
  107. // NewGZIPDecompressor creates a Decompressor based on GZIP.
  108. //
  109. // Deprecated: use package encoding/gzip.
  110. func NewGZIPDecompressor() Decompressor {
  111. return &gzipDecompressor{}
  112. }
  113. func (d *gzipDecompressor) Do(r io.Reader) ([]byte, error) {
  114. var z *gzip.Reader
  115. switch maybeZ := d.pool.Get().(type) {
  116. case nil:
  117. newZ, err := gzip.NewReader(r)
  118. if err != nil {
  119. return nil, err
  120. }
  121. z = newZ
  122. case *gzip.Reader:
  123. z = maybeZ
  124. if err := z.Reset(r); err != nil {
  125. d.pool.Put(z)
  126. return nil, err
  127. }
  128. }
  129. defer func() {
  130. z.Close()
  131. d.pool.Put(z)
  132. }()
  133. return ioutil.ReadAll(z)
  134. }
  135. func (d *gzipDecompressor) Type() string {
  136. return "gzip"
  137. }
  138. // callInfo contains all related configuration and information about an RPC.
  139. type callInfo struct {
  140. compressorType string
  141. failFast bool
  142. stream ClientStream
  143. maxReceiveMessageSize *int
  144. maxSendMessageSize *int
  145. creds credentials.PerRPCCredentials
  146. contentSubtype string
  147. codec baseCodec
  148. maxRetryRPCBufferSize int
  149. }
  150. func defaultCallInfo() *callInfo {
  151. return &callInfo{
  152. failFast: true,
  153. maxRetryRPCBufferSize: 256 * 1024, // 256KB
  154. }
  155. }
  156. // CallOption configures a Call before it starts or extracts information from
  157. // a Call after it completes.
  158. type CallOption interface {
  159. // before is called before the call is sent to any server. If before
  160. // returns a non-nil error, the RPC fails with that error.
  161. before(*callInfo) error
  162. // after is called after the call has completed. after cannot return an
  163. // error, so any failures should be reported via output parameters.
  164. after(*callInfo)
  165. }
  166. // EmptyCallOption does not alter the Call configuration.
  167. // It can be embedded in another structure to carry satellite data for use
  168. // by interceptors.
  169. type EmptyCallOption struct{}
  170. func (EmptyCallOption) before(*callInfo) error { return nil }
  171. func (EmptyCallOption) after(*callInfo) {}
  172. // Header returns a CallOptions that retrieves the header metadata
  173. // for a unary RPC.
  174. func Header(md *metadata.MD) CallOption {
  175. return HeaderCallOption{HeaderAddr: md}
  176. }
  177. // HeaderCallOption is a CallOption for collecting response header metadata.
  178. // The metadata field will be populated *after* the RPC completes.
  179. // This is an EXPERIMENTAL API.
  180. type HeaderCallOption struct {
  181. HeaderAddr *metadata.MD
  182. }
  183. func (o HeaderCallOption) before(c *callInfo) error { return nil }
  184. func (o HeaderCallOption) after(c *callInfo) {
  185. if c.stream != nil {
  186. *o.HeaderAddr, _ = c.stream.Header()
  187. }
  188. }
  189. // Trailer returns a CallOptions that retrieves the trailer metadata
  190. // for a unary RPC.
  191. func Trailer(md *metadata.MD) CallOption {
  192. return TrailerCallOption{TrailerAddr: md}
  193. }
  194. // TrailerCallOption is a CallOption for collecting response trailer metadata.
  195. // The metadata field will be populated *after* the RPC completes.
  196. // This is an EXPERIMENTAL API.
  197. type TrailerCallOption struct {
  198. TrailerAddr *metadata.MD
  199. }
  200. func (o TrailerCallOption) before(c *callInfo) error { return nil }
  201. func (o TrailerCallOption) after(c *callInfo) {
  202. if c.stream != nil {
  203. *o.TrailerAddr = c.stream.Trailer()
  204. }
  205. }
  206. // Peer returns a CallOption that retrieves peer information for a unary RPC.
  207. // The peer field will be populated *after* the RPC completes.
  208. func Peer(p *peer.Peer) CallOption {
  209. return PeerCallOption{PeerAddr: p}
  210. }
  211. // PeerCallOption is a CallOption for collecting the identity of the remote
  212. // peer. The peer field will be populated *after* the RPC completes.
  213. // This is an EXPERIMENTAL API.
  214. type PeerCallOption struct {
  215. PeerAddr *peer.Peer
  216. }
  217. func (o PeerCallOption) before(c *callInfo) error { return nil }
  218. func (o PeerCallOption) after(c *callInfo) {
  219. if c.stream != nil {
  220. if x, ok := peer.FromContext(c.stream.Context()); ok {
  221. *o.PeerAddr = *x
  222. }
  223. }
  224. }
  225. // FailFast configures the action to take when an RPC is attempted on broken
  226. // connections or unreachable servers. If failFast is true, the RPC will fail
  227. // immediately. Otherwise, the RPC client will block the call until a
  228. // connection is available (or the call is canceled or times out) and will
  229. // retry the call if it fails due to a transient error. gRPC will not retry if
  230. // data was written to the wire unless the server indicates it did not process
  231. // the data. Please refer to
  232. // https://github.com/grpc/grpc/blob/master/doc/wait-for-ready.md.
  233. //
  234. // By default, RPCs are "Fail Fast".
  235. func FailFast(failFast bool) CallOption {
  236. return FailFastCallOption{FailFast: failFast}
  237. }
  238. // FailFastCallOption is a CallOption for indicating whether an RPC should fail
  239. // fast or not.
  240. // This is an EXPERIMENTAL API.
  241. type FailFastCallOption struct {
  242. FailFast bool
  243. }
  244. func (o FailFastCallOption) before(c *callInfo) error {
  245. c.failFast = o.FailFast
  246. return nil
  247. }
  248. func (o FailFastCallOption) after(c *callInfo) {}
  249. // MaxCallRecvMsgSize returns a CallOption which sets the maximum message size the client can receive.
  250. func MaxCallRecvMsgSize(s int) CallOption {
  251. return MaxRecvMsgSizeCallOption{MaxRecvMsgSize: s}
  252. }
  253. // MaxRecvMsgSizeCallOption is a CallOption that indicates the maximum message
  254. // size the client can receive.
  255. // This is an EXPERIMENTAL API.
  256. type MaxRecvMsgSizeCallOption struct {
  257. MaxRecvMsgSize int
  258. }
  259. func (o MaxRecvMsgSizeCallOption) before(c *callInfo) error {
  260. c.maxReceiveMessageSize = &o.MaxRecvMsgSize
  261. return nil
  262. }
  263. func (o MaxRecvMsgSizeCallOption) after(c *callInfo) {}
  264. // MaxCallSendMsgSize returns a CallOption which sets the maximum message size the client can send.
  265. func MaxCallSendMsgSize(s int) CallOption {
  266. return MaxSendMsgSizeCallOption{MaxSendMsgSize: s}
  267. }
  268. // MaxSendMsgSizeCallOption is a CallOption that indicates the maximum message
  269. // size the client can send.
  270. // This is an EXPERIMENTAL API.
  271. type MaxSendMsgSizeCallOption struct {
  272. MaxSendMsgSize int
  273. }
  274. func (o MaxSendMsgSizeCallOption) before(c *callInfo) error {
  275. c.maxSendMessageSize = &o.MaxSendMsgSize
  276. return nil
  277. }
  278. func (o MaxSendMsgSizeCallOption) after(c *callInfo) {}
  279. // PerRPCCredentials returns a CallOption that sets credentials.PerRPCCredentials
  280. // for a call.
  281. func PerRPCCredentials(creds credentials.PerRPCCredentials) CallOption {
  282. return PerRPCCredsCallOption{Creds: creds}
  283. }
  284. // PerRPCCredsCallOption is a CallOption that indicates the per-RPC
  285. // credentials to use for the call.
  286. // This is an EXPERIMENTAL API.
  287. type PerRPCCredsCallOption struct {
  288. Creds credentials.PerRPCCredentials
  289. }
  290. func (o PerRPCCredsCallOption) before(c *callInfo) error {
  291. c.creds = o.Creds
  292. return nil
  293. }
  294. func (o PerRPCCredsCallOption) after(c *callInfo) {}
  295. // UseCompressor returns a CallOption which sets the compressor used when
  296. // sending the request. If WithCompressor is also set, UseCompressor has
  297. // higher priority.
  298. //
  299. // This API is EXPERIMENTAL.
  300. func UseCompressor(name string) CallOption {
  301. return CompressorCallOption{CompressorType: name}
  302. }
  303. // CompressorCallOption is a CallOption that indicates the compressor to use.
  304. // This is an EXPERIMENTAL API.
  305. type CompressorCallOption struct {
  306. CompressorType string
  307. }
  308. func (o CompressorCallOption) before(c *callInfo) error {
  309. c.compressorType = o.CompressorType
  310. return nil
  311. }
  312. func (o CompressorCallOption) after(c *callInfo) {}
  313. // CallContentSubtype returns a CallOption that will set the content-subtype
  314. // for a call. For example, if content-subtype is "json", the Content-Type over
  315. // the wire will be "application/grpc+json". The content-subtype is converted
  316. // to lowercase before being included in Content-Type. See Content-Type on
  317. // https://github.com/grpc/grpc/blob/master/doc/PROTOCOL-HTTP2.md#requests for
  318. // more details.
  319. //
  320. // If CallCustomCodec is not also used, the content-subtype will be used to
  321. // look up the Codec to use in the registry controlled by RegisterCodec. See
  322. // the documentation on RegisterCodec for details on registration. The lookup
  323. // of content-subtype is case-insensitive. If no such Codec is found, the call
  324. // will result in an error with code codes.Internal.
  325. //
  326. // If CallCustomCodec is also used, that Codec will be used for all request and
  327. // response messages, with the content-subtype set to the given contentSubtype
  328. // here for requests.
  329. func CallContentSubtype(contentSubtype string) CallOption {
  330. return ContentSubtypeCallOption{ContentSubtype: strings.ToLower(contentSubtype)}
  331. }
  332. // ContentSubtypeCallOption is a CallOption that indicates the content-subtype
  333. // used for marshaling messages.
  334. // This is an EXPERIMENTAL API.
  335. type ContentSubtypeCallOption struct {
  336. ContentSubtype string
  337. }
  338. func (o ContentSubtypeCallOption) before(c *callInfo) error {
  339. c.contentSubtype = o.ContentSubtype
  340. return nil
  341. }
  342. func (o ContentSubtypeCallOption) after(c *callInfo) {}
  343. // CallCustomCodec returns a CallOption that will set the given Codec to be
  344. // used for all request and response messages for a call. The result of calling
  345. // String() will be used as the content-subtype in a case-insensitive manner.
  346. //
  347. // See Content-Type on
  348. // https://github.com/grpc/grpc/blob/master/doc/PROTOCOL-HTTP2.md#requests for
  349. // more details. Also see the documentation on RegisterCodec and
  350. // CallContentSubtype for more details on the interaction between Codec and
  351. // content-subtype.
  352. //
  353. // This function is provided for advanced users; prefer to use only
  354. // CallContentSubtype to select a registered codec instead.
  355. func CallCustomCodec(codec Codec) CallOption {
  356. return CustomCodecCallOption{Codec: codec}
  357. }
  358. // CustomCodecCallOption is a CallOption that indicates the codec used for
  359. // marshaling messages.
  360. // This is an EXPERIMENTAL API.
  361. type CustomCodecCallOption struct {
  362. Codec Codec
  363. }
  364. func (o CustomCodecCallOption) before(c *callInfo) error {
  365. c.codec = o.Codec
  366. return nil
  367. }
  368. func (o CustomCodecCallOption) after(c *callInfo) {}
  369. // MaxRetryRPCBufferSize returns a CallOption that limits the amount of memory
  370. // used for buffering this RPC's requests for retry purposes.
  371. //
  372. // This API is EXPERIMENTAL.
  373. func MaxRetryRPCBufferSize(bytes int) CallOption {
  374. return MaxRetryRPCBufferSizeCallOption{bytes}
  375. }
  376. // MaxRetryRPCBufferSizeCallOption is a CallOption indicating the amount of
  377. // memory to be used for caching this RPC for retry purposes.
  378. // This is an EXPERIMENTAL API.
  379. type MaxRetryRPCBufferSizeCallOption struct {
  380. MaxRetryRPCBufferSize int
  381. }
  382. func (o MaxRetryRPCBufferSizeCallOption) before(c *callInfo) error {
  383. c.maxRetryRPCBufferSize = o.MaxRetryRPCBufferSize
  384. return nil
  385. }
  386. func (o MaxRetryRPCBufferSizeCallOption) after(c *callInfo) {}
  387. // The format of the payload: compressed or not?
  388. type payloadFormat uint8
  389. const (
  390. compressionNone payloadFormat = 0 // no compression
  391. compressionMade payloadFormat = 1 // compressed
  392. )
  393. // parser reads complete gRPC messages from the underlying reader.
  394. type parser struct {
  395. // r is the underlying reader.
  396. // See the comment on recvMsg for the permissible
  397. // error types.
  398. r io.Reader
  399. // The header of a gRPC message. Find more detail at
  400. // https://github.com/grpc/grpc/blob/master/doc/PROTOCOL-HTTP2.md
  401. header [5]byte
  402. }
  403. // recvMsg reads a complete gRPC message from the stream.
  404. //
  405. // It returns the message and its payload (compression/encoding)
  406. // format. The caller owns the returned msg memory.
  407. //
  408. // If there is an error, possible values are:
  409. // * io.EOF, when no messages remain
  410. // * io.ErrUnexpectedEOF
  411. // * of type transport.ConnectionError
  412. // * an error from the status package
  413. // No other error values or types must be returned, which also means
  414. // that the underlying io.Reader must not return an incompatible
  415. // error.
  416. func (p *parser) recvMsg(maxReceiveMessageSize int) (pf payloadFormat, msg []byte, err error) {
  417. if _, err := p.r.Read(p.header[:]); err != nil {
  418. return 0, nil, err
  419. }
  420. pf = payloadFormat(p.header[0])
  421. length := binary.BigEndian.Uint32(p.header[1:])
  422. if length == 0 {
  423. return pf, nil, nil
  424. }
  425. if int64(length) > int64(maxInt) {
  426. return 0, nil, status.Errorf(codes.ResourceExhausted, "grpc: received message larger than max length allowed on current machine (%d vs. %d)", length, maxInt)
  427. }
  428. if int(length) > maxReceiveMessageSize {
  429. return 0, nil, status.Errorf(codes.ResourceExhausted, "grpc: received message larger than max (%d vs. %d)", length, maxReceiveMessageSize)
  430. }
  431. // TODO(bradfitz,zhaoq): garbage. reuse buffer after proto decoding instead
  432. // of making it for each message:
  433. msg = make([]byte, int(length))
  434. if _, err := p.r.Read(msg); err != nil {
  435. if err == io.EOF {
  436. err = io.ErrUnexpectedEOF
  437. }
  438. return 0, nil, err
  439. }
  440. return pf, msg, nil
  441. }
  442. // encode serializes msg and returns a buffer containing the message, or an
  443. // error if it is too large to be transmitted by grpc. If msg is nil, it
  444. // generates an empty message.
  445. func encode(c baseCodec, msg interface{}) ([]byte, error) {
  446. if msg == nil { // NOTE: typed nils will not be caught by this check
  447. return nil, nil
  448. }
  449. b, err := c.Marshal(msg)
  450. if err != nil {
  451. return nil, status.Errorf(codes.Internal, "grpc: error while marshaling: %v", err.Error())
  452. }
  453. if uint(len(b)) > math.MaxUint32 {
  454. return nil, status.Errorf(codes.ResourceExhausted, "grpc: message too large (%d bytes)", len(b))
  455. }
  456. return b, nil
  457. }
  458. // compress returns the input bytes compressed by compressor or cp. If both
  459. // compressors are nil, returns nil.
  460. //
  461. // TODO(dfawley): eliminate cp parameter by wrapping Compressor in an encoding.Compressor.
  462. func compress(in []byte, cp Compressor, compressor encoding.Compressor) ([]byte, error) {
  463. if compressor == nil && cp == nil {
  464. return nil, nil
  465. }
  466. wrapErr := func(err error) error {
  467. return status.Errorf(codes.Internal, "grpc: error while compressing: %v", err.Error())
  468. }
  469. cbuf := &bytes.Buffer{}
  470. if compressor != nil {
  471. z, err := compressor.Compress(cbuf)
  472. if err != nil {
  473. return nil, wrapErr(err)
  474. }
  475. if _, err := z.Write(in); err != nil {
  476. return nil, wrapErr(err)
  477. }
  478. if err := z.Close(); err != nil {
  479. return nil, wrapErr(err)
  480. }
  481. } else {
  482. if err := cp.Do(cbuf, in); err != nil {
  483. return nil, wrapErr(err)
  484. }
  485. }
  486. return cbuf.Bytes(), nil
  487. }
  488. const (
  489. payloadLen = 1
  490. sizeLen = 4
  491. headerLen = payloadLen + sizeLen
  492. )
  493. // msgHeader returns a 5-byte header for the message being transmitted and the
  494. // payload, which is compData if non-nil or data otherwise.
  495. func msgHeader(data, compData []byte) (hdr []byte, payload []byte) {
  496. hdr = make([]byte, headerLen)
  497. if compData != nil {
  498. hdr[0] = byte(compressionMade)
  499. data = compData
  500. } else {
  501. hdr[0] = byte(compressionNone)
  502. }
  503. // Write length of payload into buf
  504. binary.BigEndian.PutUint32(hdr[payloadLen:], uint32(len(data)))
  505. return hdr, data
  506. }
  507. func outPayload(client bool, msg interface{}, data, payload []byte, t time.Time) *stats.OutPayload {
  508. return &stats.OutPayload{
  509. Client: client,
  510. Payload: msg,
  511. Data: data,
  512. Length: len(data),
  513. WireLength: len(payload) + headerLen,
  514. SentTime: t,
  515. }
  516. }
  517. func checkRecvPayload(pf payloadFormat, recvCompress string, haveCompressor bool) *status.Status {
  518. switch pf {
  519. case compressionNone:
  520. case compressionMade:
  521. if recvCompress == "" || recvCompress == encoding.Identity {
  522. return status.New(codes.Internal, "grpc: compressed flag set with identity or empty encoding")
  523. }
  524. if !haveCompressor {
  525. return status.Newf(codes.Unimplemented, "grpc: Decompressor is not installed for grpc-encoding %q", recvCompress)
  526. }
  527. default:
  528. return status.Newf(codes.Internal, "grpc: received unexpected payload format %d", pf)
  529. }
  530. return nil
  531. }
  532. type payloadInfo struct {
  533. wireLength int // The compressed length got from wire.
  534. uncompressedBytes []byte
  535. }
  536. func recvAndDecompress(p *parser, s *transport.Stream, dc Decompressor, maxReceiveMessageSize int, payInfo *payloadInfo, compressor encoding.Compressor) ([]byte, error) {
  537. pf, d, err := p.recvMsg(maxReceiveMessageSize)
  538. if err != nil {
  539. return nil, err
  540. }
  541. if payInfo != nil {
  542. payInfo.wireLength = len(d)
  543. }
  544. if st := checkRecvPayload(pf, s.RecvCompress(), compressor != nil || dc != nil); st != nil {
  545. return nil, st.Err()
  546. }
  547. if pf == compressionMade {
  548. // To match legacy behavior, if the decompressor is set by WithDecompressor or RPCDecompressor,
  549. // use this decompressor as the default.
  550. if dc != nil {
  551. d, err = dc.Do(bytes.NewReader(d))
  552. if err != nil {
  553. return nil, status.Errorf(codes.Internal, "grpc: failed to decompress the received message %v", err)
  554. }
  555. } else {
  556. dcReader, err := compressor.Decompress(bytes.NewReader(d))
  557. if err != nil {
  558. return nil, status.Errorf(codes.Internal, "grpc: failed to decompress the received message %v", err)
  559. }
  560. d, err = ioutil.ReadAll(dcReader)
  561. if err != nil {
  562. return nil, status.Errorf(codes.Internal, "grpc: failed to decompress the received message %v", err)
  563. }
  564. }
  565. }
  566. if len(d) > maxReceiveMessageSize {
  567. // TODO: Revisit the error code. Currently keep it consistent with java
  568. // implementation.
  569. return nil, status.Errorf(codes.ResourceExhausted, "grpc: received message larger than max (%d vs. %d)", len(d), maxReceiveMessageSize)
  570. }
  571. return d, nil
  572. }
  573. // For the two compressor parameters, both should not be set, but if they are,
  574. // dc takes precedence over compressor.
  575. // TODO(dfawley): wrap the old compressor/decompressor using the new API?
  576. func recv(p *parser, c baseCodec, s *transport.Stream, dc Decompressor, m interface{}, maxReceiveMessageSize int, payInfo *payloadInfo, compressor encoding.Compressor) error {
  577. d, err := recvAndDecompress(p, s, dc, maxReceiveMessageSize, payInfo, compressor)
  578. if err != nil {
  579. return err
  580. }
  581. if err := c.Unmarshal(d, m); err != nil {
  582. return status.Errorf(codes.Internal, "grpc: failed to unmarshal the received message %v", err)
  583. }
  584. if payInfo != nil {
  585. payInfo.uncompressedBytes = d
  586. }
  587. return nil
  588. }
  589. type rpcInfo struct {
  590. failfast bool
  591. }
  592. type rpcInfoContextKey struct{}
  593. func newContextWithRPCInfo(ctx context.Context, failfast bool) context.Context {
  594. return context.WithValue(ctx, rpcInfoContextKey{}, &rpcInfo{failfast: failfast})
  595. }
  596. func rpcInfoFromContext(ctx context.Context) (s *rpcInfo, ok bool) {
  597. s, ok = ctx.Value(rpcInfoContextKey{}).(*rpcInfo)
  598. return
  599. }
  600. // Code returns the error code for err if it was produced by the rpc system.
  601. // Otherwise, it returns codes.Unknown.
  602. //
  603. // Deprecated: use status.FromError and Code method instead.
  604. func Code(err error) codes.Code {
  605. if s, ok := status.FromError(err); ok {
  606. return s.Code()
  607. }
  608. return codes.Unknown
  609. }
  610. // ErrorDesc returns the error description of err if it was produced by the rpc system.
  611. // Otherwise, it returns err.Error() or empty string when err is nil.
  612. //
  613. // Deprecated: use status.FromError and Message method instead.
  614. func ErrorDesc(err error) string {
  615. if s, ok := status.FromError(err); ok {
  616. return s.Message()
  617. }
  618. return err.Error()
  619. }
  620. // Errorf returns an error containing an error code and a description;
  621. // Errorf returns nil if c is OK.
  622. //
  623. // Deprecated: use status.Errorf instead.
  624. func Errorf(c codes.Code, format string, a ...interface{}) error {
  625. return status.Errorf(c, format, a...)
  626. }
  627. // toRPCErr converts an error into an error from the status package.
  628. func toRPCErr(err error) error {
  629. if err == nil || err == io.EOF {
  630. return err
  631. }
  632. if err == io.ErrUnexpectedEOF {
  633. return status.Error(codes.Internal, err.Error())
  634. }
  635. if _, ok := status.FromError(err); ok {
  636. return err
  637. }
  638. switch e := err.(type) {
  639. case transport.ConnectionError:
  640. return status.Error(codes.Unavailable, e.Desc)
  641. default:
  642. switch err {
  643. case context.DeadlineExceeded:
  644. return status.Error(codes.DeadlineExceeded, err.Error())
  645. case context.Canceled:
  646. return status.Error(codes.Canceled, err.Error())
  647. }
  648. }
  649. return status.Error(codes.Unknown, err.Error())
  650. }
  651. // setCallInfoCodec should only be called after CallOptions have been applied.
  652. func setCallInfoCodec(c *callInfo) error {
  653. if c.codec != nil {
  654. // codec was already set by a CallOption; use it.
  655. return nil
  656. }
  657. if c.contentSubtype == "" {
  658. // No codec specified in CallOptions; use proto by default.
  659. c.codec = encoding.GetCodec(proto.Name)
  660. return nil
  661. }
  662. // c.contentSubtype is already lowercased in CallContentSubtype
  663. c.codec = encoding.GetCodec(c.contentSubtype)
  664. if c.codec == nil {
  665. return status.Errorf(codes.Internal, "no codec registered for content-subtype %s", c.contentSubtype)
  666. }
  667. return nil
  668. }
  669. // parseDialTarget returns the network and address to pass to dialer
  670. func parseDialTarget(target string) (net string, addr string) {
  671. net = "tcp"
  672. m1 := strings.Index(target, ":")
  673. m2 := strings.Index(target, ":/")
  674. // handle unix:addr which will fail with url.Parse
  675. if m1 >= 0 && m2 < 0 {
  676. if n := target[0:m1]; n == "unix" {
  677. net = n
  678. addr = target[m1+1:]
  679. return net, addr
  680. }
  681. }
  682. if m2 >= 0 {
  683. t, err := url.Parse(target)
  684. if err != nil {
  685. return net, target
  686. }
  687. scheme := t.Scheme
  688. addr = t.Path
  689. if scheme == "unix" {
  690. net = scheme
  691. if addr == "" {
  692. addr = t.Host
  693. }
  694. return net, addr
  695. }
  696. }
  697. return net, target
  698. }
  699. // channelzData is used to store channelz related data for ClientConn, addrConn and Server.
  700. // These fields cannot be embedded in the original structs (e.g. ClientConn), since to do atomic
  701. // operation on int64 variable on 32-bit machine, user is responsible to enforce memory alignment.
  702. // Here, by grouping those int64 fields inside a struct, we are enforcing the alignment.
  703. type channelzData struct {
  704. callsStarted int64
  705. callsFailed int64
  706. callsSucceeded int64
  707. // lastCallStartedTime stores the timestamp that last call starts. It is of int64 type instead of
  708. // time.Time since it's more costly to atomically update time.Time variable than int64 variable.
  709. lastCallStartedTime int64
  710. }
  711. // The SupportPackageIsVersion variables are referenced from generated protocol
  712. // buffer files to ensure compatibility with the gRPC version used. The latest
  713. // support package version is 5.
  714. //
  715. // Older versions are kept for compatibility. They may be removed if
  716. // compatibility cannot be maintained.
  717. //
  718. // These constants should not be referenced from any other code.
  719. const (
  720. SupportPackageIsVersion3 = true
  721. SupportPackageIsVersion4 = true
  722. SupportPackageIsVersion5 = true
  723. )
  724. const grpcUA = "grpc-go/" + Version