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