blob: 934fc1aa015e371509c88149f94d03f7ab689d0b [file] [log] [blame]
khenaidoo5fc5cea2021-08-11 17:39:16 -04001/*
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
19package grpc
20
21import (
22 "bytes"
23 "compress/gzip"
24 "context"
25 "encoding/binary"
26 "fmt"
27 "io"
28 "io/ioutil"
29 "math"
30 "strings"
31 "sync"
32 "time"
33
34 "google.golang.org/grpc/codes"
35 "google.golang.org/grpc/credentials"
36 "google.golang.org/grpc/encoding"
37 "google.golang.org/grpc/encoding/proto"
38 "google.golang.org/grpc/internal/transport"
39 "google.golang.org/grpc/metadata"
40 "google.golang.org/grpc/peer"
41 "google.golang.org/grpc/stats"
42 "google.golang.org/grpc/status"
43)
44
45// Compressor defines the interface gRPC uses to compress a message.
46//
47// Deprecated: use package encoding.
48type Compressor interface {
49 // Do compresses p into w.
50 Do(w io.Writer, p []byte) error
51 // Type returns the compression algorithm the Compressor uses.
52 Type() string
53}
54
55type gzipCompressor struct {
56 pool sync.Pool
57}
58
59// NewGZIPCompressor creates a Compressor based on GZIP.
60//
61// Deprecated: use package encoding/gzip.
62func NewGZIPCompressor() Compressor {
63 c, _ := NewGZIPCompressorWithLevel(gzip.DefaultCompression)
64 return c
65}
66
67// NewGZIPCompressorWithLevel is like NewGZIPCompressor but specifies the gzip compression level instead
68// of assuming DefaultCompression.
69//
70// The error returned will be nil if the level is valid.
71//
72// Deprecated: use package encoding/gzip.
73func NewGZIPCompressorWithLevel(level int) (Compressor, error) {
74 if level < gzip.DefaultCompression || level > gzip.BestCompression {
75 return nil, fmt.Errorf("grpc: invalid compression level: %d", level)
76 }
77 return &gzipCompressor{
78 pool: sync.Pool{
79 New: func() interface{} {
80 w, err := gzip.NewWriterLevel(ioutil.Discard, level)
81 if err != nil {
82 panic(err)
83 }
84 return w
85 },
86 },
87 }, nil
88}
89
90func (c *gzipCompressor) Do(w io.Writer, p []byte) error {
91 z := c.pool.Get().(*gzip.Writer)
92 defer c.pool.Put(z)
93 z.Reset(w)
94 if _, err := z.Write(p); err != nil {
95 return err
96 }
97 return z.Close()
98}
99
100func (c *gzipCompressor) Type() string {
101 return "gzip"
102}
103
104// Decompressor defines the interface gRPC uses to decompress a message.
105//
106// Deprecated: use package encoding.
107type Decompressor interface {
108 // Do reads the data from r and uncompress them.
109 Do(r io.Reader) ([]byte, error)
110 // Type returns the compression algorithm the Decompressor uses.
111 Type() string
112}
113
114type gzipDecompressor struct {
115 pool sync.Pool
116}
117
118// NewGZIPDecompressor creates a Decompressor based on GZIP.
119//
120// Deprecated: use package encoding/gzip.
121func NewGZIPDecompressor() Decompressor {
122 return &gzipDecompressor{}
123}
124
125func (d *gzipDecompressor) Do(r io.Reader) ([]byte, error) {
126 var z *gzip.Reader
127 switch maybeZ := d.pool.Get().(type) {
128 case nil:
129 newZ, err := gzip.NewReader(r)
130 if err != nil {
131 return nil, err
132 }
133 z = newZ
134 case *gzip.Reader:
135 z = maybeZ
136 if err := z.Reset(r); err != nil {
137 d.pool.Put(z)
138 return nil, err
139 }
140 }
141
142 defer func() {
143 z.Close()
144 d.pool.Put(z)
145 }()
146 return ioutil.ReadAll(z)
147}
148
149func (d *gzipDecompressor) Type() string {
150 return "gzip"
151}
152
153// callInfo contains all related configuration and information about an RPC.
154type callInfo struct {
155 compressorType string
156 failFast bool
157 maxReceiveMessageSize *int
158 maxSendMessageSize *int
159 creds credentials.PerRPCCredentials
160 contentSubtype string
161 codec baseCodec
162 maxRetryRPCBufferSize int
163}
164
165func defaultCallInfo() *callInfo {
166 return &callInfo{
167 failFast: true,
168 maxRetryRPCBufferSize: 256 * 1024, // 256KB
169 }
170}
171
172// CallOption configures a Call before it starts or extracts information from
173// a Call after it completes.
174type CallOption interface {
175 // before is called before the call is sent to any server. If before
176 // returns a non-nil error, the RPC fails with that error.
177 before(*callInfo) error
178
179 // after is called after the call has completed. after cannot return an
180 // error, so any failures should be reported via output parameters.
181 after(*callInfo, *csAttempt)
182}
183
184// EmptyCallOption does not alter the Call configuration.
185// It can be embedded in another structure to carry satellite data for use
186// by interceptors.
187type EmptyCallOption struct{}
188
189func (EmptyCallOption) before(*callInfo) error { return nil }
190func (EmptyCallOption) after(*callInfo, *csAttempt) {}
191
192// Header returns a CallOptions that retrieves the header metadata
193// for a unary RPC.
194func Header(md *metadata.MD) CallOption {
195 return HeaderCallOption{HeaderAddr: md}
196}
197
198// HeaderCallOption is a CallOption for collecting response header metadata.
199// The metadata field will be populated *after* the RPC completes.
200//
Joey Armstrongba3d9d12024-01-15 14:22:11 -0500201// # Experimental
khenaidoo5fc5cea2021-08-11 17:39:16 -0400202//
203// Notice: This type is EXPERIMENTAL and may be changed or removed in a
204// later release.
205type HeaderCallOption struct {
206 HeaderAddr *metadata.MD
207}
208
209func (o HeaderCallOption) before(c *callInfo) error { return nil }
210func (o HeaderCallOption) after(c *callInfo, attempt *csAttempt) {
211 *o.HeaderAddr, _ = attempt.s.Header()
212}
213
214// Trailer returns a CallOptions that retrieves the trailer metadata
215// for a unary RPC.
216func Trailer(md *metadata.MD) CallOption {
217 return TrailerCallOption{TrailerAddr: md}
218}
219
220// TrailerCallOption is a CallOption for collecting response trailer metadata.
221// The metadata field will be populated *after* the RPC completes.
222//
Joey Armstrongba3d9d12024-01-15 14:22:11 -0500223// # Experimental
khenaidoo5fc5cea2021-08-11 17:39:16 -0400224//
225// Notice: This type is EXPERIMENTAL and may be changed or removed in a
226// later release.
227type TrailerCallOption struct {
228 TrailerAddr *metadata.MD
229}
230
231func (o TrailerCallOption) before(c *callInfo) error { return nil }
232func (o TrailerCallOption) after(c *callInfo, attempt *csAttempt) {
233 *o.TrailerAddr = attempt.s.Trailer()
234}
235
236// Peer returns a CallOption that retrieves peer information for a unary RPC.
237// The peer field will be populated *after* the RPC completes.
238func Peer(p *peer.Peer) CallOption {
239 return PeerCallOption{PeerAddr: p}
240}
241
242// PeerCallOption is a CallOption for collecting the identity of the remote
243// peer. The peer field will be populated *after* the RPC completes.
244//
Joey Armstrongba3d9d12024-01-15 14:22:11 -0500245// # Experimental
khenaidoo5fc5cea2021-08-11 17:39:16 -0400246//
247// Notice: This type is EXPERIMENTAL and may be changed or removed in a
248// later release.
249type PeerCallOption struct {
250 PeerAddr *peer.Peer
251}
252
253func (o PeerCallOption) before(c *callInfo) error { return nil }
254func (o PeerCallOption) after(c *callInfo, attempt *csAttempt) {
255 if x, ok := peer.FromContext(attempt.s.Context()); ok {
256 *o.PeerAddr = *x
257 }
258}
259
260// WaitForReady configures the action to take when an RPC is attempted on broken
261// connections or unreachable servers. If waitForReady is false and the
262// connection is in the TRANSIENT_FAILURE state, the RPC will fail
263// immediately. Otherwise, the RPC client will block the call until a
264// connection is available (or the call is canceled or times out) and will
265// retry the call if it fails due to a transient error. gRPC will not retry if
266// data was written to the wire unless the server indicates it did not process
267// the data. Please refer to
268// https://github.com/grpc/grpc/blob/master/doc/wait-for-ready.md.
269//
270// By default, RPCs don't "wait for ready".
271func WaitForReady(waitForReady bool) CallOption {
272 return FailFastCallOption{FailFast: !waitForReady}
273}
274
275// FailFast is the opposite of WaitForReady.
276//
277// Deprecated: use WaitForReady.
278func FailFast(failFast bool) CallOption {
279 return FailFastCallOption{FailFast: failFast}
280}
281
282// FailFastCallOption is a CallOption for indicating whether an RPC should fail
283// fast or not.
284//
Joey Armstrongba3d9d12024-01-15 14:22:11 -0500285// # Experimental
khenaidoo5fc5cea2021-08-11 17:39:16 -0400286//
287// Notice: This type is EXPERIMENTAL and may be changed or removed in a
288// later release.
289type FailFastCallOption struct {
290 FailFast bool
291}
292
293func (o FailFastCallOption) before(c *callInfo) error {
294 c.failFast = o.FailFast
295 return nil
296}
297func (o FailFastCallOption) after(c *callInfo, attempt *csAttempt) {}
298
299// MaxCallRecvMsgSize returns a CallOption which sets the maximum message size
300// in bytes the client can receive.
301func MaxCallRecvMsgSize(bytes int) CallOption {
302 return MaxRecvMsgSizeCallOption{MaxRecvMsgSize: bytes}
303}
304
305// MaxRecvMsgSizeCallOption is a CallOption that indicates the maximum message
306// size in bytes the client can receive.
307//
Joey Armstrongba3d9d12024-01-15 14:22:11 -0500308// # Experimental
khenaidoo5fc5cea2021-08-11 17:39:16 -0400309//
310// Notice: This type is EXPERIMENTAL and may be changed or removed in a
311// later release.
312type MaxRecvMsgSizeCallOption struct {
313 MaxRecvMsgSize int
314}
315
316func (o MaxRecvMsgSizeCallOption) before(c *callInfo) error {
317 c.maxReceiveMessageSize = &o.MaxRecvMsgSize
318 return nil
319}
320func (o MaxRecvMsgSizeCallOption) after(c *callInfo, attempt *csAttempt) {}
321
322// MaxCallSendMsgSize returns a CallOption which sets the maximum message size
323// in bytes the client can send.
324func MaxCallSendMsgSize(bytes int) CallOption {
325 return MaxSendMsgSizeCallOption{MaxSendMsgSize: bytes}
326}
327
328// MaxSendMsgSizeCallOption is a CallOption that indicates the maximum message
329// size in bytes the client can send.
330//
Joey Armstrongba3d9d12024-01-15 14:22:11 -0500331// # Experimental
khenaidoo5fc5cea2021-08-11 17:39:16 -0400332//
333// Notice: This type is EXPERIMENTAL and may be changed or removed in a
334// later release.
335type MaxSendMsgSizeCallOption struct {
336 MaxSendMsgSize int
337}
338
339func (o MaxSendMsgSizeCallOption) before(c *callInfo) error {
340 c.maxSendMessageSize = &o.MaxSendMsgSize
341 return nil
342}
343func (o MaxSendMsgSizeCallOption) after(c *callInfo, attempt *csAttempt) {}
344
345// PerRPCCredentials returns a CallOption that sets credentials.PerRPCCredentials
346// for a call.
347func PerRPCCredentials(creds credentials.PerRPCCredentials) CallOption {
348 return PerRPCCredsCallOption{Creds: creds}
349}
350
351// PerRPCCredsCallOption is a CallOption that indicates the per-RPC
352// credentials to use for the call.
353//
Joey Armstrongba3d9d12024-01-15 14:22:11 -0500354// # Experimental
khenaidoo5fc5cea2021-08-11 17:39:16 -0400355//
356// Notice: This type is EXPERIMENTAL and may be changed or removed in a
357// later release.
358type PerRPCCredsCallOption struct {
359 Creds credentials.PerRPCCredentials
360}
361
362func (o PerRPCCredsCallOption) before(c *callInfo) error {
363 c.creds = o.Creds
364 return nil
365}
366func (o PerRPCCredsCallOption) after(c *callInfo, attempt *csAttempt) {}
367
368// UseCompressor returns a CallOption which sets the compressor used when
369// sending the request. If WithCompressor is also set, UseCompressor has
370// higher priority.
371//
Joey Armstrongba3d9d12024-01-15 14:22:11 -0500372// # Experimental
khenaidoo5fc5cea2021-08-11 17:39:16 -0400373//
374// Notice: This API is EXPERIMENTAL and may be changed or removed in a
375// later release.
376func UseCompressor(name string) CallOption {
377 return CompressorCallOption{CompressorType: name}
378}
379
380// CompressorCallOption is a CallOption that indicates the compressor to use.
381//
Joey Armstrongba3d9d12024-01-15 14:22:11 -0500382// # Experimental
khenaidoo5fc5cea2021-08-11 17:39:16 -0400383//
384// Notice: This type is EXPERIMENTAL and may be changed or removed in a
385// later release.
386type CompressorCallOption struct {
387 CompressorType string
388}
389
390func (o CompressorCallOption) before(c *callInfo) error {
391 c.compressorType = o.CompressorType
392 return nil
393}
394func (o CompressorCallOption) after(c *callInfo, attempt *csAttempt) {}
395
396// CallContentSubtype returns a CallOption that will set the content-subtype
397// for a call. For example, if content-subtype is "json", the Content-Type over
398// the wire will be "application/grpc+json". The content-subtype is converted
399// to lowercase before being included in Content-Type. See Content-Type on
400// https://github.com/grpc/grpc/blob/master/doc/PROTOCOL-HTTP2.md#requests for
401// more details.
402//
403// If ForceCodec is not also used, the content-subtype will be used to look up
404// the Codec to use in the registry controlled by RegisterCodec. See the
405// documentation on RegisterCodec for details on registration. The lookup of
406// content-subtype is case-insensitive. If no such Codec is found, the call
407// will result in an error with code codes.Internal.
408//
409// If ForceCodec is also used, that Codec will be used for all request and
410// response messages, with the content-subtype set to the given contentSubtype
411// here for requests.
412func CallContentSubtype(contentSubtype string) CallOption {
413 return ContentSubtypeCallOption{ContentSubtype: strings.ToLower(contentSubtype)}
414}
415
416// ContentSubtypeCallOption is a CallOption that indicates the content-subtype
417// used for marshaling messages.
418//
Joey Armstrongba3d9d12024-01-15 14:22:11 -0500419// # Experimental
khenaidoo5fc5cea2021-08-11 17:39:16 -0400420//
421// Notice: This type is EXPERIMENTAL and may be changed or removed in a
422// later release.
423type ContentSubtypeCallOption struct {
424 ContentSubtype string
425}
426
427func (o ContentSubtypeCallOption) before(c *callInfo) error {
428 c.contentSubtype = o.ContentSubtype
429 return nil
430}
431func (o ContentSubtypeCallOption) after(c *callInfo, attempt *csAttempt) {}
432
433// ForceCodec returns a CallOption that will set codec to be used for all
434// request and response messages for a call. The result of calling Name() will
435// be used as the content-subtype after converting to lowercase, unless
436// CallContentSubtype is also used.
437//
438// See Content-Type on
439// https://github.com/grpc/grpc/blob/master/doc/PROTOCOL-HTTP2.md#requests for
440// more details. Also see the documentation on RegisterCodec and
441// CallContentSubtype for more details on the interaction between Codec and
442// content-subtype.
443//
444// This function is provided for advanced users; prefer to use only
445// CallContentSubtype to select a registered codec instead.
446//
Joey Armstrongba3d9d12024-01-15 14:22:11 -0500447// # Experimental
khenaidoo5fc5cea2021-08-11 17:39:16 -0400448//
449// Notice: This API is EXPERIMENTAL and may be changed or removed in a
450// later release.
451func ForceCodec(codec encoding.Codec) CallOption {
452 return ForceCodecCallOption{Codec: codec}
453}
454
455// ForceCodecCallOption is a CallOption that indicates the codec used for
456// marshaling messages.
457//
Joey Armstrongba3d9d12024-01-15 14:22:11 -0500458// # Experimental
khenaidoo5fc5cea2021-08-11 17:39:16 -0400459//
460// Notice: This type is EXPERIMENTAL and may be changed or removed in a
461// later release.
462type ForceCodecCallOption struct {
463 Codec encoding.Codec
464}
465
466func (o ForceCodecCallOption) before(c *callInfo) error {
467 c.codec = o.Codec
468 return nil
469}
470func (o ForceCodecCallOption) after(c *callInfo, attempt *csAttempt) {}
471
472// CallCustomCodec behaves like ForceCodec, but accepts a grpc.Codec instead of
473// an encoding.Codec.
474//
475// Deprecated: use ForceCodec instead.
476func CallCustomCodec(codec Codec) CallOption {
477 return CustomCodecCallOption{Codec: codec}
478}
479
480// CustomCodecCallOption is a CallOption that indicates the codec used for
481// marshaling messages.
482//
Joey Armstrongba3d9d12024-01-15 14:22:11 -0500483// # Experimental
khenaidoo5fc5cea2021-08-11 17:39:16 -0400484//
485// Notice: This type is EXPERIMENTAL and may be changed or removed in a
486// later release.
487type CustomCodecCallOption struct {
488 Codec Codec
489}
490
491func (o CustomCodecCallOption) before(c *callInfo) error {
492 c.codec = o.Codec
493 return nil
494}
495func (o CustomCodecCallOption) after(c *callInfo, attempt *csAttempt) {}
496
497// MaxRetryRPCBufferSize returns a CallOption that limits the amount of memory
498// used for buffering this RPC's requests for retry purposes.
499//
Joey Armstrongba3d9d12024-01-15 14:22:11 -0500500// # Experimental
khenaidoo5fc5cea2021-08-11 17:39:16 -0400501//
502// Notice: This API is EXPERIMENTAL and may be changed or removed in a
503// later release.
504func MaxRetryRPCBufferSize(bytes int) CallOption {
505 return MaxRetryRPCBufferSizeCallOption{bytes}
506}
507
508// MaxRetryRPCBufferSizeCallOption is a CallOption indicating the amount of
509// memory to be used for caching this RPC for retry purposes.
510//
Joey Armstrongba3d9d12024-01-15 14:22:11 -0500511// # Experimental
khenaidoo5fc5cea2021-08-11 17:39:16 -0400512//
513// Notice: This type is EXPERIMENTAL and may be changed or removed in a
514// later release.
515type MaxRetryRPCBufferSizeCallOption struct {
516 MaxRetryRPCBufferSize int
517}
518
519func (o MaxRetryRPCBufferSizeCallOption) before(c *callInfo) error {
520 c.maxRetryRPCBufferSize = o.MaxRetryRPCBufferSize
521 return nil
522}
523func (o MaxRetryRPCBufferSizeCallOption) after(c *callInfo, attempt *csAttempt) {}
524
525// The format of the payload: compressed or not?
526type payloadFormat uint8
527
528const (
529 compressionNone payloadFormat = 0 // no compression
530 compressionMade payloadFormat = 1 // compressed
531)
532
533// parser reads complete gRPC messages from the underlying reader.
534type parser struct {
535 // r is the underlying reader.
536 // See the comment on recvMsg for the permissible
537 // error types.
538 r io.Reader
539
540 // The header of a gRPC message. Find more detail at
541 // https://github.com/grpc/grpc/blob/master/doc/PROTOCOL-HTTP2.md
542 header [5]byte
543}
544
545// recvMsg reads a complete gRPC message from the stream.
546//
547// It returns the message and its payload (compression/encoding)
548// format. The caller owns the returned msg memory.
549//
550// If there is an error, possible values are:
Joey Armstrongba3d9d12024-01-15 14:22:11 -0500551// - io.EOF, when no messages remain
552// - io.ErrUnexpectedEOF
553// - of type transport.ConnectionError
554// - an error from the status package
555//
khenaidoo5fc5cea2021-08-11 17:39:16 -0400556// No other error values or types must be returned, which also means
557// that the underlying io.Reader must not return an incompatible
558// error.
559func (p *parser) recvMsg(maxReceiveMessageSize int) (pf payloadFormat, msg []byte, err error) {
560 if _, err := p.r.Read(p.header[:]); err != nil {
561 return 0, nil, err
562 }
563
564 pf = payloadFormat(p.header[0])
565 length := binary.BigEndian.Uint32(p.header[1:])
566
567 if length == 0 {
568 return pf, nil, nil
569 }
570 if int64(length) > int64(maxInt) {
571 return 0, nil, status.Errorf(codes.ResourceExhausted, "grpc: received message larger than max length allowed on current machine (%d vs. %d)", length, maxInt)
572 }
573 if int(length) > maxReceiveMessageSize {
574 return 0, nil, status.Errorf(codes.ResourceExhausted, "grpc: received message larger than max (%d vs. %d)", length, maxReceiveMessageSize)
575 }
576 // TODO(bradfitz,zhaoq): garbage. reuse buffer after proto decoding instead
577 // of making it for each message:
578 msg = make([]byte, int(length))
579 if _, err := p.r.Read(msg); err != nil {
580 if err == io.EOF {
581 err = io.ErrUnexpectedEOF
582 }
583 return 0, nil, err
584 }
585 return pf, msg, nil
586}
587
588// encode serializes msg and returns a buffer containing the message, or an
589// error if it is too large to be transmitted by grpc. If msg is nil, it
590// generates an empty message.
591func encode(c baseCodec, msg interface{}) ([]byte, error) {
592 if msg == nil { // NOTE: typed nils will not be caught by this check
593 return nil, nil
594 }
595 b, err := c.Marshal(msg)
596 if err != nil {
597 return nil, status.Errorf(codes.Internal, "grpc: error while marshaling: %v", err.Error())
598 }
599 if uint(len(b)) > math.MaxUint32 {
600 return nil, status.Errorf(codes.ResourceExhausted, "grpc: message too large (%d bytes)", len(b))
601 }
602 return b, nil
603}
604
605// compress returns the input bytes compressed by compressor or cp. If both
606// compressors are nil, returns nil.
607//
608// TODO(dfawley): eliminate cp parameter by wrapping Compressor in an encoding.Compressor.
609func compress(in []byte, cp Compressor, compressor encoding.Compressor) ([]byte, error) {
610 if compressor == nil && cp == nil {
611 return nil, nil
612 }
613 wrapErr := func(err error) error {
614 return status.Errorf(codes.Internal, "grpc: error while compressing: %v", err.Error())
615 }
616 cbuf := &bytes.Buffer{}
617 if compressor != nil {
618 z, err := compressor.Compress(cbuf)
619 if err != nil {
620 return nil, wrapErr(err)
621 }
622 if _, err := z.Write(in); err != nil {
623 return nil, wrapErr(err)
624 }
625 if err := z.Close(); err != nil {
626 return nil, wrapErr(err)
627 }
628 } else {
629 if err := cp.Do(cbuf, in); err != nil {
630 return nil, wrapErr(err)
631 }
632 }
633 return cbuf.Bytes(), nil
634}
635
636const (
637 payloadLen = 1
638 sizeLen = 4
639 headerLen = payloadLen + sizeLen
640)
641
642// msgHeader returns a 5-byte header for the message being transmitted and the
643// payload, which is compData if non-nil or data otherwise.
644func msgHeader(data, compData []byte) (hdr []byte, payload []byte) {
645 hdr = make([]byte, headerLen)
646 if compData != nil {
647 hdr[0] = byte(compressionMade)
648 data = compData
649 } else {
650 hdr[0] = byte(compressionNone)
651 }
652
653 // Write length of payload into buf
654 binary.BigEndian.PutUint32(hdr[payloadLen:], uint32(len(data)))
655 return hdr, data
656}
657
658func outPayload(client bool, msg interface{}, data, payload []byte, t time.Time) *stats.OutPayload {
659 return &stats.OutPayload{
660 Client: client,
661 Payload: msg,
662 Data: data,
663 Length: len(data),
664 WireLength: len(payload) + headerLen,
665 SentTime: t,
666 }
667}
668
669func checkRecvPayload(pf payloadFormat, recvCompress string, haveCompressor bool) *status.Status {
670 switch pf {
671 case compressionNone:
672 case compressionMade:
673 if recvCompress == "" || recvCompress == encoding.Identity {
674 return status.New(codes.Internal, "grpc: compressed flag set with identity or empty encoding")
675 }
676 if !haveCompressor {
677 return status.Newf(codes.Unimplemented, "grpc: Decompressor is not installed for grpc-encoding %q", recvCompress)
678 }
679 default:
680 return status.Newf(codes.Internal, "grpc: received unexpected payload format %d", pf)
681 }
682 return nil
683}
684
685type payloadInfo struct {
686 wireLength int // The compressed length got from wire.
687 uncompressedBytes []byte
688}
689
690func recvAndDecompress(p *parser, s *transport.Stream, dc Decompressor, maxReceiveMessageSize int, payInfo *payloadInfo, compressor encoding.Compressor) ([]byte, error) {
691 pf, d, err := p.recvMsg(maxReceiveMessageSize)
692 if err != nil {
693 return nil, err
694 }
695 if payInfo != nil {
696 payInfo.wireLength = len(d)
697 }
698
699 if st := checkRecvPayload(pf, s.RecvCompress(), compressor != nil || dc != nil); st != nil {
700 return nil, st.Err()
701 }
702
703 var size int
704 if pf == compressionMade {
705 // To match legacy behavior, if the decompressor is set by WithDecompressor or RPCDecompressor,
706 // use this decompressor as the default.
707 if dc != nil {
708 d, err = dc.Do(bytes.NewReader(d))
709 size = len(d)
710 } else {
711 d, size, err = decompress(compressor, d, maxReceiveMessageSize)
712 }
713 if err != nil {
714 return nil, status.Errorf(codes.Internal, "grpc: failed to decompress the received message %v", err)
715 }
khenaidoo257f3192021-12-15 16:46:37 -0500716 if size > maxReceiveMessageSize {
717 // TODO: Revisit the error code. Currently keep it consistent with java
718 // implementation.
719 return nil, status.Errorf(codes.ResourceExhausted, "grpc: received message after decompression larger than max (%d vs. %d)", size, maxReceiveMessageSize)
720 }
khenaidoo5fc5cea2021-08-11 17:39:16 -0400721 }
722 return d, nil
723}
724
725// Using compressor, decompress d, returning data and size.
726// Optionally, if data will be over maxReceiveMessageSize, just return the size.
727func decompress(compressor encoding.Compressor, d []byte, maxReceiveMessageSize int) ([]byte, int, error) {
728 dcReader, err := compressor.Decompress(bytes.NewReader(d))
729 if err != nil {
730 return nil, 0, err
731 }
732 if sizer, ok := compressor.(interface {
733 DecompressedSize(compressedBytes []byte) int
734 }); ok {
735 if size := sizer.DecompressedSize(d); size >= 0 {
736 if size > maxReceiveMessageSize {
737 return nil, size, nil
738 }
739 // size is used as an estimate to size the buffer, but we
740 // will read more data if available.
741 // +MinRead so ReadFrom will not reallocate if size is correct.
742 buf := bytes.NewBuffer(make([]byte, 0, size+bytes.MinRead))
743 bytesRead, err := buf.ReadFrom(io.LimitReader(dcReader, int64(maxReceiveMessageSize)+1))
744 return buf.Bytes(), int(bytesRead), err
745 }
746 }
747 // Read from LimitReader with limit max+1. So if the underlying
748 // reader is over limit, the result will be bigger than max.
749 d, err = ioutil.ReadAll(io.LimitReader(dcReader, int64(maxReceiveMessageSize)+1))
750 return d, len(d), err
751}
752
753// For the two compressor parameters, both should not be set, but if they are,
754// dc takes precedence over compressor.
755// TODO(dfawley): wrap the old compressor/decompressor using the new API?
756func recv(p *parser, c baseCodec, s *transport.Stream, dc Decompressor, m interface{}, maxReceiveMessageSize int, payInfo *payloadInfo, compressor encoding.Compressor) error {
757 d, err := recvAndDecompress(p, s, dc, maxReceiveMessageSize, payInfo, compressor)
758 if err != nil {
759 return err
760 }
761 if err := c.Unmarshal(d, m); err != nil {
762 return status.Errorf(codes.Internal, "grpc: failed to unmarshal the received message %v", err)
763 }
764 if payInfo != nil {
765 payInfo.uncompressedBytes = d
766 }
767 return nil
768}
769
770// Information about RPC
771type rpcInfo struct {
772 failfast bool
773 preloaderInfo *compressorInfo
774}
775
776// Information about Preloader
777// Responsible for storing codec, and compressors
778// If stream (s) has context s.Context which stores rpcInfo that has non nil
779// pointers to codec, and compressors, then we can use preparedMsg for Async message prep
780// and reuse marshalled bytes
781type compressorInfo struct {
782 codec baseCodec
783 cp Compressor
784 comp encoding.Compressor
785}
786
787type rpcInfoContextKey struct{}
788
789func newContextWithRPCInfo(ctx context.Context, failfast bool, codec baseCodec, cp Compressor, comp encoding.Compressor) context.Context {
790 return context.WithValue(ctx, rpcInfoContextKey{}, &rpcInfo{
791 failfast: failfast,
792 preloaderInfo: &compressorInfo{
793 codec: codec,
794 cp: cp,
795 comp: comp,
796 },
797 })
798}
799
800func rpcInfoFromContext(ctx context.Context) (s *rpcInfo, ok bool) {
801 s, ok = ctx.Value(rpcInfoContextKey{}).(*rpcInfo)
802 return
803}
804
805// Code returns the error code for err if it was produced by the rpc system.
806// Otherwise, it returns codes.Unknown.
807//
808// Deprecated: use status.Code instead.
809func Code(err error) codes.Code {
810 return status.Code(err)
811}
812
813// ErrorDesc returns the error description of err if it was produced by the rpc system.
814// Otherwise, it returns err.Error() or empty string when err is nil.
815//
816// Deprecated: use status.Convert and Message method instead.
817func ErrorDesc(err error) string {
818 return status.Convert(err).Message()
819}
820
821// Errorf returns an error containing an error code and a description;
822// Errorf returns nil if c is OK.
823//
824// Deprecated: use status.Errorf instead.
825func Errorf(c codes.Code, format string, a ...interface{}) error {
826 return status.Errorf(c, format, a...)
827}
828
829// toRPCErr converts an error into an error from the status package.
830func toRPCErr(err error) error {
831 switch err {
832 case nil, io.EOF:
833 return err
834 case context.DeadlineExceeded:
835 return status.Error(codes.DeadlineExceeded, err.Error())
836 case context.Canceled:
837 return status.Error(codes.Canceled, err.Error())
838 case io.ErrUnexpectedEOF:
839 return status.Error(codes.Internal, err.Error())
840 }
841
842 switch e := err.(type) {
843 case transport.ConnectionError:
844 return status.Error(codes.Unavailable, e.Desc)
845 case *transport.NewStreamError:
846 return toRPCErr(e.Err)
847 }
848
849 if _, ok := status.FromError(err); ok {
850 return err
851 }
852
853 return status.Error(codes.Unknown, err.Error())
854}
855
856// setCallInfoCodec should only be called after CallOptions have been applied.
857func setCallInfoCodec(c *callInfo) error {
858 if c.codec != nil {
859 // codec was already set by a CallOption; use it, but set the content
860 // subtype if it is not set.
861 if c.contentSubtype == "" {
862 // c.codec is a baseCodec to hide the difference between grpc.Codec and
863 // encoding.Codec (Name vs. String method name). We only support
864 // setting content subtype from encoding.Codec to avoid a behavior
865 // change with the deprecated version.
866 if ec, ok := c.codec.(encoding.Codec); ok {
867 c.contentSubtype = strings.ToLower(ec.Name())
868 }
869 }
870 return nil
871 }
872
873 if c.contentSubtype == "" {
874 // No codec specified in CallOptions; use proto by default.
875 c.codec = encoding.GetCodec(proto.Name)
876 return nil
877 }
878
879 // c.contentSubtype is already lowercased in CallContentSubtype
880 c.codec = encoding.GetCodec(c.contentSubtype)
881 if c.codec == nil {
882 return status.Errorf(codes.Internal, "no codec registered for content-subtype %s", c.contentSubtype)
883 }
884 return nil
885}
886
887// channelzData is used to store channelz related data for ClientConn, addrConn and Server.
888// These fields cannot be embedded in the original structs (e.g. ClientConn), since to do atomic
889// operation on int64 variable on 32-bit machine, user is responsible to enforce memory alignment.
890// Here, by grouping those int64 fields inside a struct, we are enforcing the alignment.
891type channelzData struct {
892 callsStarted int64
893 callsFailed int64
894 callsSucceeded int64
895 // lastCallStartedTime stores the timestamp that last call starts. It is of int64 type instead of
896 // time.Time since it's more costly to atomically update time.Time variable than int64 variable.
897 lastCallStartedTime int64
898}
899
900// The SupportPackageIsVersion variables are referenced from generated protocol
901// buffer files to ensure compatibility with the gRPC version used. The latest
902// support package version is 7.
903//
904// Older versions are kept for compatibility.
905//
906// These constants should not be referenced from any other code.
907const (
908 SupportPackageIsVersion3 = true
909 SupportPackageIsVersion4 = true
910 SupportPackageIsVersion5 = true
911 SupportPackageIsVersion6 = true
912 SupportPackageIsVersion7 = true
913)
914
915const grpcUA = "grpc-go/" + Version