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
19package grpc
20
21import (
22 "compress/gzip"
23 "context"
24 "encoding/binary"
25 "fmt"
26 "io"
27 "math"
28 "strings"
29 "sync"
30 "time"
31
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/mem"
38 "google.golang.org/grpc/metadata"
39 "google.golang.org/grpc/peer"
40 "google.golang.org/grpc/stats"
41 "google.golang.org/grpc/status"
42)
43
44// Compressor defines the interface gRPC uses to compress a message.
45//
46// Deprecated: use package encoding.
47type Compressor interface {
48 // Do compresses p into w.
49 Do(w io.Writer, p []byte) error
50 // Type returns the compression algorithm the Compressor uses.
51 Type() string
52}
53
54type gzipCompressor struct {
55 pool sync.Pool
56}
57
58// NewGZIPCompressor creates a Compressor based on GZIP.
59//
60// Deprecated: use package encoding/gzip.
61func NewGZIPCompressor() Compressor {
62 c, _ := NewGZIPCompressorWithLevel(gzip.DefaultCompression)
63 return c
64}
65
66// NewGZIPCompressorWithLevel is like NewGZIPCompressor but specifies the gzip compression level instead
67// of assuming DefaultCompression.
68//
69// The error returned will be nil if the level is valid.
70//
71// Deprecated: use package encoding/gzip.
72func NewGZIPCompressorWithLevel(level int) (Compressor, error) {
73 if level < gzip.DefaultCompression || level > gzip.BestCompression {
74 return nil, fmt.Errorf("grpc: invalid compression level: %d", level)
75 }
76 return &gzipCompressor{
77 pool: sync.Pool{
78 New: func() any {
79 w, err := gzip.NewWriterLevel(io.Discard, level)
80 if err != nil {
81 panic(err)
82 }
83 return w
84 },
85 },
86 }, nil
87}
88
89func (c *gzipCompressor) Do(w io.Writer, p []byte) error {
90 z := c.pool.Get().(*gzip.Writer)
91 defer c.pool.Put(z)
92 z.Reset(w)
93 if _, err := z.Write(p); err != nil {
94 return err
95 }
96 return z.Close()
97}
98
99func (c *gzipCompressor) Type() string {
100 return "gzip"
101}
102
103// Decompressor defines the interface gRPC uses to decompress a message.
104//
105// Deprecated: use package encoding.
106type Decompressor interface {
107 // Do reads the data from r and uncompress them.
108 Do(r io.Reader) ([]byte, error)
109 // Type returns the compression algorithm the Decompressor uses.
110 Type() string
111}
112
113type gzipDecompressor struct {
114 pool sync.Pool
115}
116
117// NewGZIPDecompressor creates a Decompressor based on GZIP.
118//
119// Deprecated: use package encoding/gzip.
120func NewGZIPDecompressor() Decompressor {
121 return &gzipDecompressor{}
122}
123
124func (d *gzipDecompressor) Do(r io.Reader) ([]byte, error) {
125 var z *gzip.Reader
126 switch maybeZ := d.pool.Get().(type) {
127 case nil:
128 newZ, err := gzip.NewReader(r)
129 if err != nil {
130 return nil, err
131 }
132 z = newZ
133 case *gzip.Reader:
134 z = maybeZ
135 if err := z.Reset(r); err != nil {
136 d.pool.Put(z)
137 return nil, err
138 }
139 }
140
141 defer func() {
142 z.Close()
143 d.pool.Put(z)
144 }()
145 return io.ReadAll(z)
146}
147
148func (d *gzipDecompressor) Type() string {
149 return "gzip"
150}
151
152// callInfo contains all related configuration and information about an RPC.
153type callInfo struct {
154 compressorName string
155 failFast bool
156 maxReceiveMessageSize *int
157 maxSendMessageSize *int
158 creds credentials.PerRPCCredentials
159 contentSubtype string
160 codec baseCodec
161 maxRetryRPCBufferSize int
162 onFinish []func(err error)
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// StaticMethod returns a CallOption which specifies that a call is being made
193// to a method that is static, which means the method is known at compile time
194// and doesn't change at runtime. This can be used as a signal to stats plugins
195// that this method is safe to include as a key to a measurement.
196func StaticMethod() CallOption {
197 return StaticMethodCallOption{}
198}
199
200// StaticMethodCallOption is a CallOption that specifies that a call comes
201// from a static method.
202type StaticMethodCallOption struct {
203 EmptyCallOption
204}
205
206// Header returns a CallOptions that retrieves the header metadata
207// for a unary RPC.
208func Header(md *metadata.MD) CallOption {
209 return HeaderCallOption{HeaderAddr: md}
210}
211
212// HeaderCallOption is a CallOption for collecting response header metadata.
213// The metadata field will be populated *after* the RPC completes.
214//
215// # Experimental
216//
217// Notice: This type is EXPERIMENTAL and may be changed or removed in a
218// later release.
219type HeaderCallOption struct {
220 HeaderAddr *metadata.MD
221}
222
223func (o HeaderCallOption) before(*callInfo) error { return nil }
224func (o HeaderCallOption) after(_ *callInfo, attempt *csAttempt) {
225 *o.HeaderAddr, _ = attempt.transportStream.Header()
226}
227
228// Trailer returns a CallOptions that retrieves the trailer metadata
229// for a unary RPC.
230func Trailer(md *metadata.MD) CallOption {
231 return TrailerCallOption{TrailerAddr: md}
232}
233
234// TrailerCallOption is a CallOption for collecting response trailer metadata.
235// The metadata field will be populated *after* the RPC completes.
236//
237// # Experimental
238//
239// Notice: This type is EXPERIMENTAL and may be changed or removed in a
240// later release.
241type TrailerCallOption struct {
242 TrailerAddr *metadata.MD
243}
244
245func (o TrailerCallOption) before(*callInfo) error { return nil }
246func (o TrailerCallOption) after(_ *callInfo, attempt *csAttempt) {
247 *o.TrailerAddr = attempt.transportStream.Trailer()
248}
249
250// Peer returns a CallOption that retrieves peer information for a unary RPC.
251// The peer field will be populated *after* the RPC completes.
252func Peer(p *peer.Peer) CallOption {
253 return PeerCallOption{PeerAddr: p}
254}
255
256// PeerCallOption is a CallOption for collecting the identity of the remote
257// peer. The peer field will be populated *after* the RPC completes.
258//
259// # Experimental
260//
261// Notice: This type is EXPERIMENTAL and may be changed or removed in a
262// later release.
263type PeerCallOption struct {
264 PeerAddr *peer.Peer
265}
266
267func (o PeerCallOption) before(*callInfo) error { return nil }
268func (o PeerCallOption) after(_ *callInfo, attempt *csAttempt) {
269 if x, ok := peer.FromContext(attempt.transportStream.Context()); ok {
270 *o.PeerAddr = *x
271 }
272}
273
274// WaitForReady configures the RPC's behavior when the client is in
275// TRANSIENT_FAILURE, which occurs when all addresses fail to connect. If
276// waitForReady is false, the RPC will fail immediately. Otherwise, the client
277// will wait until a connection becomes available or the RPC's deadline is
278// reached.
279//
280// By default, RPCs do not "wait for ready".
281func WaitForReady(waitForReady bool) CallOption {
282 return FailFastCallOption{FailFast: !waitForReady}
283}
284
285// FailFast is the opposite of WaitForReady.
286//
287// Deprecated: use WaitForReady.
288func FailFast(failFast bool) CallOption {
289 return FailFastCallOption{FailFast: failFast}
290}
291
292// FailFastCallOption is a CallOption for indicating whether an RPC should fail
293// fast or not.
294//
295// # Experimental
296//
297// Notice: This type is EXPERIMENTAL and may be changed or removed in a
298// later release.
299type FailFastCallOption struct {
300 FailFast bool
301}
302
303func (o FailFastCallOption) before(c *callInfo) error {
304 c.failFast = o.FailFast
305 return nil
306}
307func (o FailFastCallOption) after(*callInfo, *csAttempt) {}
308
309// OnFinish returns a CallOption that configures a callback to be called when
310// the call completes. The error passed to the callback is the status of the
311// RPC, and may be nil. The onFinish callback provided will only be called once
312// by gRPC. This is mainly used to be used by streaming interceptors, to be
313// notified when the RPC completes along with information about the status of
314// the RPC.
315//
316// # Experimental
317//
318// Notice: This API is EXPERIMENTAL and may be changed or removed in a
319// later release.
320func OnFinish(onFinish func(err error)) CallOption {
321 return OnFinishCallOption{
322 OnFinish: onFinish,
323 }
324}
325
326// OnFinishCallOption is CallOption that indicates a callback to be called when
327// the call completes.
328//
329// # Experimental
330//
331// Notice: This type is EXPERIMENTAL and may be changed or removed in a
332// later release.
333type OnFinishCallOption struct {
334 OnFinish func(error)
335}
336
337func (o OnFinishCallOption) before(c *callInfo) error {
338 c.onFinish = append(c.onFinish, o.OnFinish)
339 return nil
340}
341
342func (o OnFinishCallOption) after(*callInfo, *csAttempt) {}
343
344// MaxCallRecvMsgSize returns a CallOption which sets the maximum message size
345// in bytes the client can receive. If this is not set, gRPC uses the default
346// 4MB.
347func MaxCallRecvMsgSize(bytes int) CallOption {
348 return MaxRecvMsgSizeCallOption{MaxRecvMsgSize: bytes}
349}
350
351// MaxRecvMsgSizeCallOption is a CallOption that indicates the maximum message
352// size in bytes the client can receive.
353//
354// # Experimental
355//
356// Notice: This type is EXPERIMENTAL and may be changed or removed in a
357// later release.
358type MaxRecvMsgSizeCallOption struct {
359 MaxRecvMsgSize int
360}
361
362func (o MaxRecvMsgSizeCallOption) before(c *callInfo) error {
363 c.maxReceiveMessageSize = &o.MaxRecvMsgSize
364 return nil
365}
366func (o MaxRecvMsgSizeCallOption) after(*callInfo, *csAttempt) {}
367
368// MaxCallSendMsgSize returns a CallOption which sets the maximum message size
369// in bytes the client can send. If this is not set, gRPC uses the default
370// `math.MaxInt32`.
371func MaxCallSendMsgSize(bytes int) CallOption {
372 return MaxSendMsgSizeCallOption{MaxSendMsgSize: bytes}
373}
374
375// MaxSendMsgSizeCallOption is a CallOption that indicates the maximum message
376// size in bytes the client can send.
377//
378// # Experimental
379//
380// Notice: This type is EXPERIMENTAL and may be changed or removed in a
381// later release.
382type MaxSendMsgSizeCallOption struct {
383 MaxSendMsgSize int
384}
385
386func (o MaxSendMsgSizeCallOption) before(c *callInfo) error {
387 c.maxSendMessageSize = &o.MaxSendMsgSize
388 return nil
389}
390func (o MaxSendMsgSizeCallOption) after(*callInfo, *csAttempt) {}
391
392// PerRPCCredentials returns a CallOption that sets credentials.PerRPCCredentials
393// for a call.
394func PerRPCCredentials(creds credentials.PerRPCCredentials) CallOption {
395 return PerRPCCredsCallOption{Creds: creds}
396}
397
398// PerRPCCredsCallOption is a CallOption that indicates the per-RPC
399// credentials to use for the call.
400//
401// # Experimental
402//
403// Notice: This type is EXPERIMENTAL and may be changed or removed in a
404// later release.
405type PerRPCCredsCallOption struct {
406 Creds credentials.PerRPCCredentials
407}
408
409func (o PerRPCCredsCallOption) before(c *callInfo) error {
410 c.creds = o.Creds
411 return nil
412}
413func (o PerRPCCredsCallOption) after(*callInfo, *csAttempt) {}
414
415// UseCompressor returns a CallOption which sets the compressor used when
416// sending the request. If WithCompressor is also set, UseCompressor has
417// higher priority.
418//
419// # Experimental
420//
421// Notice: This API is EXPERIMENTAL and may be changed or removed in a
422// later release.
423func UseCompressor(name string) CallOption {
424 return CompressorCallOption{CompressorType: name}
425}
426
427// CompressorCallOption is a CallOption that indicates the compressor to use.
428//
429// # Experimental
430//
431// Notice: This type is EXPERIMENTAL and may be changed or removed in a
432// later release.
433type CompressorCallOption struct {
434 CompressorType string
435}
436
437func (o CompressorCallOption) before(c *callInfo) error {
438 c.compressorName = o.CompressorType
439 return nil
440}
441func (o CompressorCallOption) after(*callInfo, *csAttempt) {}
442
443// CallContentSubtype returns a CallOption that will set the content-subtype
444// for a call. For example, if content-subtype is "json", the Content-Type over
445// the wire will be "application/grpc+json". The content-subtype is converted
446// to lowercase before being included in Content-Type. See Content-Type on
447// https://github.com/grpc/grpc/blob/master/doc/PROTOCOL-HTTP2.md#requests for
448// more details.
449//
450// If ForceCodec is not also used, the content-subtype will be used to look up
451// the Codec to use in the registry controlled by RegisterCodec. See the
452// documentation on RegisterCodec for details on registration. The lookup of
453// content-subtype is case-insensitive. If no such Codec is found, the call
454// will result in an error with code codes.Internal.
455//
456// If ForceCodec is also used, that Codec will be used for all request and
457// response messages, with the content-subtype set to the given contentSubtype
458// here for requests.
459func CallContentSubtype(contentSubtype string) CallOption {
460 return ContentSubtypeCallOption{ContentSubtype: strings.ToLower(contentSubtype)}
461}
462
463// ContentSubtypeCallOption is a CallOption that indicates the content-subtype
464// used for marshaling messages.
465//
466// # Experimental
467//
468// Notice: This type is EXPERIMENTAL and may be changed or removed in a
469// later release.
470type ContentSubtypeCallOption struct {
471 ContentSubtype string
472}
473
474func (o ContentSubtypeCallOption) before(c *callInfo) error {
475 c.contentSubtype = o.ContentSubtype
476 return nil
477}
478func (o ContentSubtypeCallOption) after(*callInfo, *csAttempt) {}
479
480// ForceCodec returns a CallOption that will set codec to be used for all
481// request and response messages for a call. The result of calling Name() will
482// be used as the content-subtype after converting to lowercase, unless
483// CallContentSubtype is also used.
484//
485// See Content-Type on
486// https://github.com/grpc/grpc/blob/master/doc/PROTOCOL-HTTP2.md#requests for
487// more details. Also see the documentation on RegisterCodec and
488// CallContentSubtype for more details on the interaction between Codec and
489// content-subtype.
490//
491// This function is provided for advanced users; prefer to use only
492// CallContentSubtype to select a registered codec instead.
493//
494// # Experimental
495//
496// Notice: This API is EXPERIMENTAL and may be changed or removed in a
497// later release.
498func ForceCodec(codec encoding.Codec) CallOption {
499 return ForceCodecCallOption{Codec: codec}
500}
501
502// ForceCodecCallOption is a CallOption that indicates the codec used for
503// marshaling messages.
504//
505// # Experimental
506//
507// Notice: This type is EXPERIMENTAL and may be changed or removed in a
508// later release.
509type ForceCodecCallOption struct {
510 Codec encoding.Codec
511}
512
513func (o ForceCodecCallOption) before(c *callInfo) error {
514 c.codec = newCodecV1Bridge(o.Codec)
515 return nil
516}
517func (o ForceCodecCallOption) after(*callInfo, *csAttempt) {}
518
519// ForceCodecV2 returns a CallOption that will set codec to be used for all
520// request and response messages for a call. The result of calling Name() will
521// be used as the content-subtype after converting to lowercase, unless
522// CallContentSubtype is also used.
523//
524// See Content-Type on
525// https://github.com/grpc/grpc/blob/master/doc/PROTOCOL-HTTP2.md#requests for
526// more details. Also see the documentation on RegisterCodec and
527// CallContentSubtype for more details on the interaction between Codec and
528// content-subtype.
529//
530// This function is provided for advanced users; prefer to use only
531// CallContentSubtype to select a registered codec instead.
532//
533// # Experimental
534//
535// Notice: This API is EXPERIMENTAL and may be changed or removed in a
536// later release.
537func ForceCodecV2(codec encoding.CodecV2) CallOption {
538 return ForceCodecV2CallOption{CodecV2: codec}
539}
540
541// ForceCodecV2CallOption is a CallOption that indicates the codec used for
542// marshaling messages.
543//
544// # Experimental
545//
546// Notice: This type is EXPERIMENTAL and may be changed or removed in a
547// later release.
548type ForceCodecV2CallOption struct {
549 CodecV2 encoding.CodecV2
550}
551
552func (o ForceCodecV2CallOption) before(c *callInfo) error {
553 c.codec = o.CodecV2
554 return nil
555}
556
557func (o ForceCodecV2CallOption) after(*callInfo, *csAttempt) {}
558
559// CallCustomCodec behaves like ForceCodec, but accepts a grpc.Codec instead of
560// an encoding.Codec.
561//
562// Deprecated: use ForceCodec instead.
563func CallCustomCodec(codec Codec) CallOption {
564 return CustomCodecCallOption{Codec: codec}
565}
566
567// CustomCodecCallOption is a CallOption that indicates the codec used for
568// marshaling messages.
569//
570// # Experimental
571//
572// Notice: This type is EXPERIMENTAL and may be changed or removed in a
573// later release.
574type CustomCodecCallOption struct {
575 Codec Codec
576}
577
578func (o CustomCodecCallOption) before(c *callInfo) error {
579 c.codec = newCodecV0Bridge(o.Codec)
580 return nil
581}
582func (o CustomCodecCallOption) after(*callInfo, *csAttempt) {}
583
584// MaxRetryRPCBufferSize returns a CallOption that limits the amount of memory
585// used for buffering this RPC's requests for retry purposes.
586//
587// # Experimental
588//
589// Notice: This API is EXPERIMENTAL and may be changed or removed in a
590// later release.
591func MaxRetryRPCBufferSize(bytes int) CallOption {
592 return MaxRetryRPCBufferSizeCallOption{bytes}
593}
594
595// MaxRetryRPCBufferSizeCallOption is a CallOption indicating the amount of
596// memory to be used for caching this RPC for retry purposes.
597//
598// # Experimental
599//
600// Notice: This type is EXPERIMENTAL and may be changed or removed in a
601// later release.
602type MaxRetryRPCBufferSizeCallOption struct {
603 MaxRetryRPCBufferSize int
604}
605
606func (o MaxRetryRPCBufferSizeCallOption) before(c *callInfo) error {
607 c.maxRetryRPCBufferSize = o.MaxRetryRPCBufferSize
608 return nil
609}
610func (o MaxRetryRPCBufferSizeCallOption) after(*callInfo, *csAttempt) {}
611
612// The format of the payload: compressed or not?
613type payloadFormat uint8
614
615const (
616 compressionNone payloadFormat = 0 // no compression
617 compressionMade payloadFormat = 1 // compressed
618)
619
620func (pf payloadFormat) isCompressed() bool {
621 return pf == compressionMade
622}
623
624type streamReader interface {
625 ReadMessageHeader(header []byte) error
626 Read(n int) (mem.BufferSlice, error)
627}
628
629// parser reads complete gRPC messages from the underlying reader.
630type parser struct {
631 // r is the underlying reader.
632 // See the comment on recvMsg for the permissible
633 // error types.
634 r streamReader
635
636 // The header of a gRPC message. Find more detail at
637 // https://github.com/grpc/grpc/blob/master/doc/PROTOCOL-HTTP2.md
638 header [5]byte
639
640 // bufferPool is the pool of shared receive buffers.
641 bufferPool mem.BufferPool
642}
643
644// recvMsg reads a complete gRPC message from the stream.
645//
646// It returns the message and its payload (compression/encoding)
647// format. The caller owns the returned msg memory.
648//
649// If there is an error, possible values are:
650// - io.EOF, when no messages remain
651// - io.ErrUnexpectedEOF
652// - of type transport.ConnectionError
653// - an error from the status package
654//
655// No other error values or types must be returned, which also means
656// that the underlying streamReader must not return an incompatible
657// error.
658func (p *parser) recvMsg(maxReceiveMessageSize int) (payloadFormat, mem.BufferSlice, error) {
659 err := p.r.ReadMessageHeader(p.header[:])
660 if err != nil {
661 return 0, nil, err
662 }
663
664 pf := payloadFormat(p.header[0])
665 length := binary.BigEndian.Uint32(p.header[1:])
666
667 if int64(length) > int64(maxInt) {
668 return 0, nil, status.Errorf(codes.ResourceExhausted, "grpc: received message larger than max length allowed on current machine (%d vs. %d)", length, maxInt)
669 }
670 if int(length) > maxReceiveMessageSize {
671 return 0, nil, status.Errorf(codes.ResourceExhausted, "grpc: received message larger than max (%d vs. %d)", length, maxReceiveMessageSize)
672 }
673
674 data, err := p.r.Read(int(length))
675 if err != nil {
676 if err == io.EOF {
677 err = io.ErrUnexpectedEOF
678 }
679 return 0, nil, err
680 }
681 return pf, data, nil
682}
683
684// encode serializes msg and returns a buffer containing the message, or an
685// error if it is too large to be transmitted by grpc. If msg is nil, it
686// generates an empty message.
687func encode(c baseCodec, msg any) (mem.BufferSlice, error) {
688 if msg == nil { // NOTE: typed nils will not be caught by this check
689 return nil, nil
690 }
691 b, err := c.Marshal(msg)
692 if err != nil {
693 return nil, status.Errorf(codes.Internal, "grpc: error while marshaling: %v", err.Error())
694 }
695 if bufSize := uint(b.Len()); bufSize > math.MaxUint32 {
696 b.Free()
697 return nil, status.Errorf(codes.ResourceExhausted, "grpc: message too large (%d bytes)", bufSize)
698 }
699 return b, nil
700}
701
702// compress returns the input bytes compressed by compressor or cp.
703// If both compressors are nil, or if the message has zero length, returns nil,
704// indicating no compression was done.
705//
706// TODO(dfawley): eliminate cp parameter by wrapping Compressor in an encoding.Compressor.
707func compress(in mem.BufferSlice, cp Compressor, compressor encoding.Compressor, pool mem.BufferPool) (mem.BufferSlice, payloadFormat, error) {
708 if (compressor == nil && cp == nil) || in.Len() == 0 {
709 return nil, compressionNone, nil
710 }
711 var out mem.BufferSlice
712 w := mem.NewWriter(&out, pool)
713 wrapErr := func(err error) error {
714 out.Free()
715 return status.Errorf(codes.Internal, "grpc: error while compressing: %v", err.Error())
716 }
717 if compressor != nil {
718 z, err := compressor.Compress(w)
719 if err != nil {
720 return nil, 0, wrapErr(err)
721 }
722 for _, b := range in {
723 if _, err := z.Write(b.ReadOnlyData()); err != nil {
724 return nil, 0, wrapErr(err)
725 }
726 }
727 if err := z.Close(); err != nil {
728 return nil, 0, wrapErr(err)
729 }
730 } else {
731 // This is obviously really inefficient since it fully materializes the data, but
732 // there is no way around this with the old Compressor API. At least it attempts
733 // to return the buffer to the provider, in the hopes it can be reused (maybe
734 // even by a subsequent call to this very function).
735 buf := in.MaterializeToBuffer(pool)
736 defer buf.Free()
737 if err := cp.Do(w, buf.ReadOnlyData()); err != nil {
738 return nil, 0, wrapErr(err)
739 }
740 }
741 return out, compressionMade, nil
742}
743
744const (
745 payloadLen = 1
746 sizeLen = 4
747 headerLen = payloadLen + sizeLen
748)
749
750// msgHeader returns a 5-byte header for the message being transmitted and the
751// payload, which is compData if non-nil or data otherwise.
752func msgHeader(data, compData mem.BufferSlice, pf payloadFormat) (hdr []byte, payload mem.BufferSlice) {
753 hdr = make([]byte, headerLen)
754 hdr[0] = byte(pf)
755
756 var length uint32
757 if pf.isCompressed() {
758 length = uint32(compData.Len())
759 payload = compData
760 } else {
761 length = uint32(data.Len())
762 payload = data
763 }
764
765 // Write length of payload into buf
766 binary.BigEndian.PutUint32(hdr[payloadLen:], length)
767 return hdr, payload
768}
769
770func outPayload(client bool, msg any, dataLength, payloadLength int, t time.Time) *stats.OutPayload {
771 return &stats.OutPayload{
772 Client: client,
773 Payload: msg,
774 Length: dataLength,
775 WireLength: payloadLength + headerLen,
776 CompressedLength: payloadLength,
777 SentTime: t,
778 }
779}
780
781func checkRecvPayload(pf payloadFormat, recvCompress string, haveCompressor bool, isServer bool) *status.Status {
782 switch pf {
783 case compressionNone:
784 case compressionMade:
785 if recvCompress == "" || recvCompress == encoding.Identity {
786 return status.New(codes.Internal, "grpc: compressed flag set with identity or empty encoding")
787 }
788 if !haveCompressor {
789 if isServer {
790 return status.Newf(codes.Unimplemented, "grpc: Decompressor is not installed for grpc-encoding %q", recvCompress)
791 }
792 return status.Newf(codes.Internal, "grpc: Decompressor is not installed for grpc-encoding %q", recvCompress)
793 }
794 default:
795 return status.Newf(codes.Internal, "grpc: received unexpected payload format %d", pf)
796 }
797 return nil
798}
799
800type payloadInfo struct {
801 compressedLength int // The compressed length got from wire.
802 uncompressedBytes mem.BufferSlice
803}
804
805func (p *payloadInfo) free() {
806 if p != nil && p.uncompressedBytes != nil {
807 p.uncompressedBytes.Free()
808 }
809}
810
811// recvAndDecompress reads a message from the stream, decompressing it if necessary.
812//
813// Cancelling the returned cancel function releases the buffer back to the pool. So the caller should cancel as soon as
814// the buffer is no longer needed.
815// TODO: Refactor this function to reduce the number of arguments.
816// See: https://google.github.io/styleguide/go/best-practices.html#function-argument-lists
817func recvAndDecompress(p *parser, s recvCompressor, dc Decompressor, maxReceiveMessageSize int, payInfo *payloadInfo, compressor encoding.Compressor, isServer bool,
818) (out mem.BufferSlice, err error) {
819 pf, compressed, err := p.recvMsg(maxReceiveMessageSize)
820 if err != nil {
821 return nil, err
822 }
823
824 compressedLength := compressed.Len()
825
826 if st := checkRecvPayload(pf, s.RecvCompress(), compressor != nil || dc != nil, isServer); st != nil {
827 compressed.Free()
828 return nil, st.Err()
829 }
830
831 if pf.isCompressed() {
832 defer compressed.Free()
833 // To match legacy behavior, if the decompressor is set by WithDecompressor or RPCDecompressor,
834 // use this decompressor as the default.
835 out, err = decompress(compressor, compressed, dc, maxReceiveMessageSize, p.bufferPool)
836 if err != nil {
837 return nil, err
838 }
839 } else {
840 out = compressed
841 }
842
843 if payInfo != nil {
844 payInfo.compressedLength = compressedLength
845 out.Ref()
846 payInfo.uncompressedBytes = out
847 }
848
849 return out, nil
850}
851
852// decompress processes the given data by decompressing it using either a custom decompressor or a standard compressor.
853// If a custom decompressor is provided, it takes precedence. The function validates that the decompressed data
854// does not exceed the specified maximum size and returns an error if this limit is exceeded.
855// On success, it returns the decompressed data. Otherwise, it returns an error if decompression fails or the data exceeds the size limit.
856func decompress(compressor encoding.Compressor, d mem.BufferSlice, dc Decompressor, maxReceiveMessageSize int, pool mem.BufferPool) (mem.BufferSlice, error) {
857 if dc != nil {
858 uncompressed, err := dc.Do(d.Reader())
859 if err != nil {
860 return nil, status.Errorf(codes.Internal, "grpc: failed to decompress the received message: %v", err)
861 }
862 if len(uncompressed) > maxReceiveMessageSize {
863 return nil, status.Errorf(codes.ResourceExhausted, "grpc: message after decompression larger than max (%d vs. %d)", len(uncompressed), maxReceiveMessageSize)
864 }
865 return mem.BufferSlice{mem.SliceBuffer(uncompressed)}, nil
866 }
867 if compressor != nil {
868 dcReader, err := compressor.Decompress(d.Reader())
869 if err != nil {
870 return nil, status.Errorf(codes.Internal, "grpc: failed to decompress the message: %v", err)
871 }
872
873 out, err := mem.ReadAll(io.LimitReader(dcReader, int64(maxReceiveMessageSize)), pool)
874 if err != nil {
875 out.Free()
876 return nil, status.Errorf(codes.Internal, "grpc: failed to read decompressed data: %v", err)
877 }
878
879 if out.Len() == maxReceiveMessageSize && !atEOF(dcReader) {
880 out.Free()
881 return nil, status.Errorf(codes.ResourceExhausted, "grpc: received message after decompression larger than max %d", maxReceiveMessageSize)
882 }
883 return out, nil
884 }
885 return nil, status.Errorf(codes.Internal, "grpc: no decompressor available for compressed payload")
886}
887
888// atEOF reads data from r and returns true if zero bytes could be read and r.Read returns EOF.
889func atEOF(dcReader io.Reader) bool {
890 n, err := dcReader.Read(make([]byte, 1))
891 return n == 0 && err == io.EOF
892}
893
894type recvCompressor interface {
895 RecvCompress() string
896}
897
898// For the two compressor parameters, both should not be set, but if they are,
899// dc takes precedence over compressor.
900// TODO(dfawley): wrap the old compressor/decompressor using the new API?
901func recv(p *parser, c baseCodec, s recvCompressor, dc Decompressor, m any, maxReceiveMessageSize int, payInfo *payloadInfo, compressor encoding.Compressor, isServer bool) error {
902 data, err := recvAndDecompress(p, s, dc, maxReceiveMessageSize, payInfo, compressor, isServer)
903 if err != nil {
904 return err
905 }
906
907 // If the codec wants its own reference to the data, it can get it. Otherwise, always
908 // free the buffers.
909 defer data.Free()
910
911 if err := c.Unmarshal(data, m); err != nil {
912 return status.Errorf(codes.Internal, "grpc: failed to unmarshal the received message: %v", err)
913 }
914
915 return nil
916}
917
918// Information about RPC
919type rpcInfo struct {
920 failfast bool
921 preloaderInfo *compressorInfo
922}
923
924// Information about Preloader
925// Responsible for storing codec, and compressors
926// If stream (s) has context s.Context which stores rpcInfo that has non nil
927// pointers to codec, and compressors, then we can use preparedMsg for Async message prep
928// and reuse marshalled bytes
929type compressorInfo struct {
930 codec baseCodec
931 cp Compressor
932 comp encoding.Compressor
933}
934
935type rpcInfoContextKey struct{}
936
937func newContextWithRPCInfo(ctx context.Context, failfast bool, codec baseCodec, cp Compressor, comp encoding.Compressor) context.Context {
938 return context.WithValue(ctx, rpcInfoContextKey{}, &rpcInfo{
939 failfast: failfast,
940 preloaderInfo: &compressorInfo{
941 codec: codec,
942 cp: cp,
943 comp: comp,
944 },
945 })
946}
947
948func rpcInfoFromContext(ctx context.Context) (s *rpcInfo, ok bool) {
949 s, ok = ctx.Value(rpcInfoContextKey{}).(*rpcInfo)
950 return
951}
952
953// Code returns the error code for err if it was produced by the rpc system.
954// Otherwise, it returns codes.Unknown.
955//
956// Deprecated: use status.Code instead.
957func Code(err error) codes.Code {
958 return status.Code(err)
959}
960
961// ErrorDesc returns the error description of err if it was produced by the rpc system.
962// Otherwise, it returns err.Error() or empty string when err is nil.
963//
964// Deprecated: use status.Convert and Message method instead.
965func ErrorDesc(err error) string {
966 return status.Convert(err).Message()
967}
968
969// Errorf returns an error containing an error code and a description;
970// Errorf returns nil if c is OK.
971//
972// Deprecated: use status.Errorf instead.
973func Errorf(c codes.Code, format string, a ...any) error {
974 return status.Errorf(c, format, a...)
975}
976
977var errContextCanceled = status.Error(codes.Canceled, context.Canceled.Error())
978var errContextDeadline = status.Error(codes.DeadlineExceeded, context.DeadlineExceeded.Error())
979
980// toRPCErr converts an error into an error from the status package.
981func toRPCErr(err error) error {
982 switch err {
983 case nil, io.EOF:
984 return err
985 case context.DeadlineExceeded:
986 return errContextDeadline
987 case context.Canceled:
988 return errContextCanceled
989 case io.ErrUnexpectedEOF:
990 return status.Error(codes.Internal, err.Error())
991 }
992
993 switch e := err.(type) {
994 case transport.ConnectionError:
995 return status.Error(codes.Unavailable, e.Desc)
996 case *transport.NewStreamError:
997 return toRPCErr(e.Err)
998 }
999
1000 if _, ok := status.FromError(err); ok {
1001 return err
1002 }
1003
1004 return status.Error(codes.Unknown, err.Error())
1005}
1006
1007// setCallInfoCodec should only be called after CallOptions have been applied.
1008func setCallInfoCodec(c *callInfo) error {
1009 if c.codec != nil {
1010 // codec was already set by a CallOption; use it, but set the content
1011 // subtype if it is not set.
1012 if c.contentSubtype == "" {
1013 // c.codec is a baseCodec to hide the difference between grpc.Codec and
1014 // encoding.Codec (Name vs. String method name). We only support
1015 // setting content subtype from encoding.Codec to avoid a behavior
1016 // change with the deprecated version.
1017 if ec, ok := c.codec.(encoding.CodecV2); ok {
1018 c.contentSubtype = strings.ToLower(ec.Name())
1019 }
1020 }
1021 return nil
1022 }
1023
1024 if c.contentSubtype == "" {
1025 // No codec specified in CallOptions; use proto by default.
1026 c.codec = getCodec(proto.Name)
1027 return nil
1028 }
1029
1030 // c.contentSubtype is already lowercased in CallContentSubtype
1031 c.codec = getCodec(c.contentSubtype)
1032 if c.codec == nil {
1033 return status.Errorf(codes.Internal, "no codec registered for content-subtype %s", c.contentSubtype)
1034 }
1035 return nil
1036}
1037
1038// The SupportPackageIsVersion variables are referenced from generated protocol
1039// buffer files to ensure compatibility with the gRPC version used. The latest
1040// support package version is 9.
1041//
1042// Older versions are kept for compatibility.
1043//
1044// These constants should not be referenced from any other code.
1045const (
1046 SupportPackageIsVersion3 = true
1047 SupportPackageIsVersion4 = true
1048 SupportPackageIsVersion5 = true
1049 SupportPackageIsVersion6 = true
1050 SupportPackageIsVersion7 = true
1051 SupportPackageIsVersion8 = true
1052 SupportPackageIsVersion9 = true
1053)
1054
1055const grpcUA = "grpc-go/" + Version