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 "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// 201// Experimental 202// 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// 223// Experimental 224// 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// 245// Experimental 246// 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// 285// Experimental 286// 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// 308// Experimental 309// 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// 331// Experimental 332// 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// 354// Experimental 355// 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// 372// Experimental 373// 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// 382// Experimental 383// 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// 419// Experimental 420// 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// 447// Experimental 448// 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// 458// Experimental 459// 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// 483// Experimental 484// 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// 500// Experimental 501// 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// 511// Experimental 512// 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: 551// * io.EOF, when no messages remain 552// * io.ErrUnexpectedEOF 553// * of type transport.ConnectionError 554// * an error from the status package 555// No other error values or types must be returned, which also means 556// that the underlying io.Reader must not return an incompatible 557// error. 558func (p *parser) recvMsg(maxReceiveMessageSize int) (pf payloadFormat, msg []byte, err error) { 559 if _, err := p.r.Read(p.header[:]); err != nil { 560 return 0, nil, err 561 } 562 563 pf = payloadFormat(p.header[0]) 564 length := binary.BigEndian.Uint32(p.header[1:]) 565 566 if length == 0 { 567 return pf, nil, nil 568 } 569 if int64(length) > int64(maxInt) { 570 return 0, nil, status.Errorf(codes.ResourceExhausted, "grpc: received message larger than max length allowed on current machine (%d vs. %d)", length, maxInt) 571 } 572 if int(length) > maxReceiveMessageSize { 573 return 0, nil, status.Errorf(codes.ResourceExhausted, "grpc: received message larger than max (%d vs. %d)", length, maxReceiveMessageSize) 574 } 575 // TODO(bradfitz,zhaoq): garbage. reuse buffer after proto decoding instead 576 // of making it for each message: 577 msg = make([]byte, int(length)) 578 if _, err := p.r.Read(msg); err != nil { 579 if err == io.EOF { 580 err = io.ErrUnexpectedEOF 581 } 582 return 0, nil, err 583 } 584 return pf, msg, nil 585} 586 587// encode serializes msg and returns a buffer containing the message, or an 588// error if it is too large to be transmitted by grpc. If msg is nil, it 589// generates an empty message. 590func encode(c baseCodec, msg interface{}) ([]byte, error) { 591 if msg == nil { // NOTE: typed nils will not be caught by this check 592 return nil, nil 593 } 594 b, err := c.Marshal(msg) 595 if err != nil { 596 return nil, status.Errorf(codes.Internal, "grpc: error while marshaling: %v", err.Error()) 597 } 598 if uint(len(b)) > math.MaxUint32 { 599 return nil, status.Errorf(codes.ResourceExhausted, "grpc: message too large (%d bytes)", len(b)) 600 } 601 return b, nil 602} 603 604// compress returns the input bytes compressed by compressor or cp. If both 605// compressors are nil, returns nil. 606// 607// TODO(dfawley): eliminate cp parameter by wrapping Compressor in an encoding.Compressor. 608func compress(in []byte, cp Compressor, compressor encoding.Compressor) ([]byte, error) { 609 if compressor == nil && cp == nil { 610 return nil, nil 611 } 612 wrapErr := func(err error) error { 613 return status.Errorf(codes.Internal, "grpc: error while compressing: %v", err.Error()) 614 } 615 cbuf := &bytes.Buffer{} 616 if compressor != nil { 617 z, err := compressor.Compress(cbuf) 618 if err != nil { 619 return nil, wrapErr(err) 620 } 621 if _, err := z.Write(in); err != nil { 622 return nil, wrapErr(err) 623 } 624 if err := z.Close(); err != nil { 625 return nil, wrapErr(err) 626 } 627 } else { 628 if err := cp.Do(cbuf, in); err != nil { 629 return nil, wrapErr(err) 630 } 631 } 632 return cbuf.Bytes(), nil 633} 634 635const ( 636 payloadLen = 1 637 sizeLen = 4 638 headerLen = payloadLen + sizeLen 639) 640 641// msgHeader returns a 5-byte header for the message being transmitted and the 642// payload, which is compData if non-nil or data otherwise. 643func msgHeader(data, compData []byte) (hdr []byte, payload []byte) { 644 hdr = make([]byte, headerLen) 645 if compData != nil { 646 hdr[0] = byte(compressionMade) 647 data = compData 648 } else { 649 hdr[0] = byte(compressionNone) 650 } 651 652 // Write length of payload into buf 653 binary.BigEndian.PutUint32(hdr[payloadLen:], uint32(len(data))) 654 return hdr, data 655} 656 657func outPayload(client bool, msg interface{}, data, payload []byte, t time.Time) *stats.OutPayload { 658 return &stats.OutPayload{ 659 Client: client, 660 Payload: msg, 661 Data: data, 662 Length: len(data), 663 WireLength: len(payload) + headerLen, 664 SentTime: t, 665 } 666} 667 668func checkRecvPayload(pf payloadFormat, recvCompress string, haveCompressor bool) *status.Status { 669 switch pf { 670 case compressionNone: 671 case compressionMade: 672 if recvCompress == "" || recvCompress == encoding.Identity { 673 return status.New(codes.Internal, "grpc: compressed flag set with identity or empty encoding") 674 } 675 if !haveCompressor { 676 return status.Newf(codes.Unimplemented, "grpc: Decompressor is not installed for grpc-encoding %q", recvCompress) 677 } 678 default: 679 return status.Newf(codes.Internal, "grpc: received unexpected payload format %d", pf) 680 } 681 return nil 682} 683 684type payloadInfo struct { 685 wireLength int // The compressed length got from wire. 686 uncompressedBytes []byte 687} 688 689func recvAndDecompress(p *parser, s *transport.Stream, dc Decompressor, maxReceiveMessageSize int, payInfo *payloadInfo, compressor encoding.Compressor) ([]byte, error) { 690 pf, d, err := p.recvMsg(maxReceiveMessageSize) 691 if err != nil { 692 return nil, err 693 } 694 if payInfo != nil { 695 payInfo.wireLength = len(d) 696 } 697 698 if st := checkRecvPayload(pf, s.RecvCompress(), compressor != nil || dc != nil); st != nil { 699 return nil, st.Err() 700 } 701 702 var size int 703 if pf == compressionMade { 704 // To match legacy behavior, if the decompressor is set by WithDecompressor or RPCDecompressor, 705 // use this decompressor as the default. 706 if dc != nil { 707 d, err = dc.Do(bytes.NewReader(d)) 708 size = len(d) 709 } else { 710 d, size, err = decompress(compressor, d, maxReceiveMessageSize) 711 } 712 if err != nil { 713 return nil, status.Errorf(codes.Internal, "grpc: failed to decompress the received message %v", err) 714 } 715 } else { 716 size = len(d) 717 } 718 if size > maxReceiveMessageSize { 719 // TODO: Revisit the error code. Currently keep it consistent with java 720 // implementation. 721 return nil, status.Errorf(codes.ResourceExhausted, "grpc: received message larger than max (%d vs. %d)", size, maxReceiveMessageSize) 722 } 723 return d, nil 724} 725 726// Using compressor, decompress d, returning data and size. 727// Optionally, if data will be over maxReceiveMessageSize, just return the size. 728func decompress(compressor encoding.Compressor, d []byte, maxReceiveMessageSize int) ([]byte, int, error) { 729 dcReader, err := compressor.Decompress(bytes.NewReader(d)) 730 if err != nil { 731 return nil, 0, err 732 } 733 if sizer, ok := compressor.(interface { 734 DecompressedSize(compressedBytes []byte) int 735 }); ok { 736 if size := sizer.DecompressedSize(d); size >= 0 { 737 if size > maxReceiveMessageSize { 738 return nil, size, nil 739 } 740 // size is used as an estimate to size the buffer, but we 741 // will read more data if available. 742 // +MinRead so ReadFrom will not reallocate if size is correct. 743 buf := bytes.NewBuffer(make([]byte, 0, size+bytes.MinRead)) 744 bytesRead, err := buf.ReadFrom(io.LimitReader(dcReader, int64(maxReceiveMessageSize)+1)) 745 return buf.Bytes(), int(bytesRead), err 746 } 747 } 748 // Read from LimitReader with limit max+1. So if the underlying 749 // reader is over limit, the result will be bigger than max. 750 d, err = ioutil.ReadAll(io.LimitReader(dcReader, int64(maxReceiveMessageSize)+1)) 751 return d, len(d), err 752} 753 754// For the two compressor parameters, both should not be set, but if they are, 755// dc takes precedence over compressor. 756// TODO(dfawley): wrap the old compressor/decompressor using the new API? 757func recv(p *parser, c baseCodec, s *transport.Stream, dc Decompressor, m interface{}, maxReceiveMessageSize int, payInfo *payloadInfo, compressor encoding.Compressor) error { 758 d, err := recvAndDecompress(p, s, dc, maxReceiveMessageSize, payInfo, compressor) 759 if err != nil { 760 return err 761 } 762 if err := c.Unmarshal(d, m); err != nil { 763 return status.Errorf(codes.Internal, "grpc: failed to unmarshal the received message %v", err) 764 } 765 if payInfo != nil { 766 payInfo.uncompressedBytes = d 767 } 768 return nil 769} 770 771// Information about RPC 772type rpcInfo struct { 773 failfast bool 774 preloaderInfo *compressorInfo 775} 776 777// Information about Preloader 778// Responsible for storing codec, and compressors 779// If stream (s) has context s.Context which stores rpcInfo that has non nil 780// pointers to codec, and compressors, then we can use preparedMsg for Async message prep 781// and reuse marshalled bytes 782type compressorInfo struct { 783 codec baseCodec 784 cp Compressor 785 comp encoding.Compressor 786} 787 788type rpcInfoContextKey struct{} 789 790func newContextWithRPCInfo(ctx context.Context, failfast bool, codec baseCodec, cp Compressor, comp encoding.Compressor) context.Context { 791 return context.WithValue(ctx, rpcInfoContextKey{}, &rpcInfo{ 792 failfast: failfast, 793 preloaderInfo: &compressorInfo{ 794 codec: codec, 795 cp: cp, 796 comp: comp, 797 }, 798 }) 799} 800 801func rpcInfoFromContext(ctx context.Context) (s *rpcInfo, ok bool) { 802 s, ok = ctx.Value(rpcInfoContextKey{}).(*rpcInfo) 803 return 804} 805 806// Code returns the error code for err if it was produced by the rpc system. 807// Otherwise, it returns codes.Unknown. 808// 809// Deprecated: use status.Code instead. 810func Code(err error) codes.Code { 811 return status.Code(err) 812} 813 814// ErrorDesc returns the error description of err if it was produced by the rpc system. 815// Otherwise, it returns err.Error() or empty string when err is nil. 816// 817// Deprecated: use status.Convert and Message method instead. 818func ErrorDesc(err error) string { 819 return status.Convert(err).Message() 820} 821 822// Errorf returns an error containing an error code and a description; 823// Errorf returns nil if c is OK. 824// 825// Deprecated: use status.Errorf instead. 826func Errorf(c codes.Code, format string, a ...interface{}) error { 827 return status.Errorf(c, format, a...) 828} 829 830// toRPCErr converts an error into an error from the status package. 831func toRPCErr(err error) error { 832 switch err { 833 case nil, io.EOF: 834 return err 835 case context.DeadlineExceeded: 836 return status.Error(codes.DeadlineExceeded, err.Error()) 837 case context.Canceled: 838 return status.Error(codes.Canceled, err.Error()) 839 case io.ErrUnexpectedEOF: 840 return status.Error(codes.Internal, err.Error()) 841 } 842 843 switch e := err.(type) { 844 case transport.ConnectionError: 845 return status.Error(codes.Unavailable, e.Desc) 846 case *transport.NewStreamError: 847 return toRPCErr(e.Err) 848 } 849 850 if _, ok := status.FromError(err); ok { 851 return err 852 } 853 854 return status.Error(codes.Unknown, err.Error()) 855} 856 857// setCallInfoCodec should only be called after CallOptions have been applied. 858func setCallInfoCodec(c *callInfo) error { 859 if c.codec != nil { 860 // codec was already set by a CallOption; use it, but set the content 861 // subtype if it is not set. 862 if c.contentSubtype == "" { 863 // c.codec is a baseCodec to hide the difference between grpc.Codec and 864 // encoding.Codec (Name vs. String method name). We only support 865 // setting content subtype from encoding.Codec to avoid a behavior 866 // change with the deprecated version. 867 if ec, ok := c.codec.(encoding.Codec); ok { 868 c.contentSubtype = strings.ToLower(ec.Name()) 869 } 870 } 871 return nil 872 } 873 874 if c.contentSubtype == "" { 875 // No codec specified in CallOptions; use proto by default. 876 c.codec = encoding.GetCodec(proto.Name) 877 return nil 878 } 879 880 // c.contentSubtype is already lowercased in CallContentSubtype 881 c.codec = encoding.GetCodec(c.contentSubtype) 882 if c.codec == nil { 883 return status.Errorf(codes.Internal, "no codec registered for content-subtype %s", c.contentSubtype) 884 } 885 return nil 886} 887 888// channelzData is used to store channelz related data for ClientConn, addrConn and Server. 889// These fields cannot be embedded in the original structs (e.g. ClientConn), since to do atomic 890// operation on int64 variable on 32-bit machine, user is responsible to enforce memory alignment. 891// Here, by grouping those int64 fields inside a struct, we are enforcing the alignment. 892type channelzData struct { 893 callsStarted int64 894 callsFailed int64 895 callsSucceeded int64 896 // lastCallStartedTime stores the timestamp that last call starts. It is of int64 type instead of 897 // time.Time since it's more costly to atomically update time.Time variable than int64 variable. 898 lastCallStartedTime int64 899} 900 901// The SupportPackageIsVersion variables are referenced from generated protocol 902// buffer files to ensure compatibility with the gRPC version used. The latest 903// support package version is 7. 904// 905// Older versions are kept for compatibility. 906// 907// These constants should not be referenced from any other code. 908const ( 909 SupportPackageIsVersion3 = true 910 SupportPackageIsVersion4 = true 911 SupportPackageIsVersion5 = true 912 SupportPackageIsVersion6 = true 913 SupportPackageIsVersion7 = true 914) 915 916const grpcUA = "grpc-go/" + Version 917