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, the RPC will fail 262// immediately. Otherwise, the RPC client will block the call until a 263// connection is available (or the call is canceled or times out) and will 264// retry the call if it fails due to a transient error. gRPC will not retry if 265// data was written to the wire unless the server indicates it did not process 266// the data. Please refer to 267// https://github.com/grpc/grpc/blob/master/doc/wait-for-ready.md. 268// 269// By default, RPCs don't "wait for ready". 270func WaitForReady(waitForReady bool) CallOption { 271 return FailFastCallOption{FailFast: !waitForReady} 272} 273 274// FailFast is the opposite of WaitForReady. 275// 276// Deprecated: use WaitForReady. 277func FailFast(failFast bool) CallOption { 278 return FailFastCallOption{FailFast: failFast} 279} 280 281// FailFastCallOption is a CallOption for indicating whether an RPC should fail 282// fast or not. 283// 284// Experimental 285// 286// Notice: This type is EXPERIMENTAL and may be changed or removed in a 287// later release. 288type FailFastCallOption struct { 289 FailFast bool 290} 291 292func (o FailFastCallOption) before(c *callInfo) error { 293 c.failFast = o.FailFast 294 return nil 295} 296func (o FailFastCallOption) after(c *callInfo, attempt *csAttempt) {} 297 298// MaxCallRecvMsgSize returns a CallOption which sets the maximum message size 299// in bytes the client can receive. 300func MaxCallRecvMsgSize(bytes int) CallOption { 301 return MaxRecvMsgSizeCallOption{MaxRecvMsgSize: bytes} 302} 303 304// MaxRecvMsgSizeCallOption is a CallOption that indicates the maximum message 305// size in bytes the client can receive. 306// 307// Experimental 308// 309// Notice: This type is EXPERIMENTAL and may be changed or removed in a 310// later release. 311type MaxRecvMsgSizeCallOption struct { 312 MaxRecvMsgSize int 313} 314 315func (o MaxRecvMsgSizeCallOption) before(c *callInfo) error { 316 c.maxReceiveMessageSize = &o.MaxRecvMsgSize 317 return nil 318} 319func (o MaxRecvMsgSizeCallOption) after(c *callInfo, attempt *csAttempt) {} 320 321// MaxCallSendMsgSize returns a CallOption which sets the maximum message size 322// in bytes the client can send. 323func MaxCallSendMsgSize(bytes int) CallOption { 324 return MaxSendMsgSizeCallOption{MaxSendMsgSize: bytes} 325} 326 327// MaxSendMsgSizeCallOption is a CallOption that indicates the maximum message 328// size in bytes the client can send. 329// 330// Experimental 331// 332// Notice: This type is EXPERIMENTAL and may be changed or removed in a 333// later release. 334type MaxSendMsgSizeCallOption struct { 335 MaxSendMsgSize int 336} 337 338func (o MaxSendMsgSizeCallOption) before(c *callInfo) error { 339 c.maxSendMessageSize = &o.MaxSendMsgSize 340 return nil 341} 342func (o MaxSendMsgSizeCallOption) after(c *callInfo, attempt *csAttempt) {} 343 344// PerRPCCredentials returns a CallOption that sets credentials.PerRPCCredentials 345// for a call. 346func PerRPCCredentials(creds credentials.PerRPCCredentials) CallOption { 347 return PerRPCCredsCallOption{Creds: creds} 348} 349 350// PerRPCCredsCallOption is a CallOption that indicates the per-RPC 351// credentials to use for the call. 352// 353// Experimental 354// 355// Notice: This type is EXPERIMENTAL and may be changed or removed in a 356// later release. 357type PerRPCCredsCallOption struct { 358 Creds credentials.PerRPCCredentials 359} 360 361func (o PerRPCCredsCallOption) before(c *callInfo) error { 362 c.creds = o.Creds 363 return nil 364} 365func (o PerRPCCredsCallOption) after(c *callInfo, attempt *csAttempt) {} 366 367// UseCompressor returns a CallOption which sets the compressor used when 368// sending the request. If WithCompressor is also set, UseCompressor has 369// higher priority. 370// 371// Experimental 372// 373// Notice: This API is EXPERIMENTAL and may be changed or removed in a 374// later release. 375func UseCompressor(name string) CallOption { 376 return CompressorCallOption{CompressorType: name} 377} 378 379// CompressorCallOption is a CallOption that indicates the compressor to use. 380// 381// Experimental 382// 383// Notice: This type is EXPERIMENTAL and may be changed or removed in a 384// later release. 385type CompressorCallOption struct { 386 CompressorType string 387} 388 389func (o CompressorCallOption) before(c *callInfo) error { 390 c.compressorType = o.CompressorType 391 return nil 392} 393func (o CompressorCallOption) after(c *callInfo, attempt *csAttempt) {} 394 395// CallContentSubtype returns a CallOption that will set the content-subtype 396// for a call. For example, if content-subtype is "json", the Content-Type over 397// the wire will be "application/grpc+json". The content-subtype is converted 398// to lowercase before being included in Content-Type. See Content-Type on 399// https://github.com/grpc/grpc/blob/master/doc/PROTOCOL-HTTP2.md#requests for 400// more details. 401// 402// If ForceCodec is not also used, the content-subtype will be used to look up 403// the Codec to use in the registry controlled by RegisterCodec. See the 404// documentation on RegisterCodec for details on registration. The lookup of 405// content-subtype is case-insensitive. If no such Codec is found, the call 406// will result in an error with code codes.Internal. 407// 408// If ForceCodec is also used, that Codec will be used for all request and 409// response messages, with the content-subtype set to the given contentSubtype 410// here for requests. 411func CallContentSubtype(contentSubtype string) CallOption { 412 return ContentSubtypeCallOption{ContentSubtype: strings.ToLower(contentSubtype)} 413} 414 415// ContentSubtypeCallOption is a CallOption that indicates the content-subtype 416// used for marshaling messages. 417// 418// Experimental 419// 420// Notice: This type is EXPERIMENTAL and may be changed or removed in a 421// later release. 422type ContentSubtypeCallOption struct { 423 ContentSubtype string 424} 425 426func (o ContentSubtypeCallOption) before(c *callInfo) error { 427 c.contentSubtype = o.ContentSubtype 428 return nil 429} 430func (o ContentSubtypeCallOption) after(c *callInfo, attempt *csAttempt) {} 431 432// ForceCodec returns a CallOption that will set the given Codec to be 433// used for all request and response messages for a call. The result of calling 434// String() will be used as the content-subtype in a case-insensitive manner. 435// 436// See Content-Type on 437// https://github.com/grpc/grpc/blob/master/doc/PROTOCOL-HTTP2.md#requests for 438// more details. Also see the documentation on RegisterCodec and 439// CallContentSubtype for more details on the interaction between Codec and 440// content-subtype. 441// 442// This function is provided for advanced users; prefer to use only 443// CallContentSubtype to select a registered codec instead. 444// 445// Experimental 446// 447// Notice: This API is EXPERIMENTAL and may be changed or removed in a 448// later release. 449func ForceCodec(codec encoding.Codec) CallOption { 450 return ForceCodecCallOption{Codec: codec} 451} 452 453// ForceCodecCallOption is a CallOption that indicates the codec used for 454// marshaling messages. 455// 456// Experimental 457// 458// Notice: This type is EXPERIMENTAL and may be changed or removed in a 459// later release. 460type ForceCodecCallOption struct { 461 Codec encoding.Codec 462} 463 464func (o ForceCodecCallOption) before(c *callInfo) error { 465 c.codec = o.Codec 466 return nil 467} 468func (o ForceCodecCallOption) after(c *callInfo, attempt *csAttempt) {} 469 470// CallCustomCodec behaves like ForceCodec, but accepts a grpc.Codec instead of 471// an encoding.Codec. 472// 473// Deprecated: use ForceCodec instead. 474func CallCustomCodec(codec Codec) CallOption { 475 return CustomCodecCallOption{Codec: codec} 476} 477 478// CustomCodecCallOption is a CallOption that indicates the codec used for 479// marshaling messages. 480// 481// Experimental 482// 483// Notice: This type is EXPERIMENTAL and may be changed or removed in a 484// later release. 485type CustomCodecCallOption struct { 486 Codec Codec 487} 488 489func (o CustomCodecCallOption) before(c *callInfo) error { 490 c.codec = o.Codec 491 return nil 492} 493func (o CustomCodecCallOption) after(c *callInfo, attempt *csAttempt) {} 494 495// MaxRetryRPCBufferSize returns a CallOption that limits the amount of memory 496// used for buffering this RPC's requests for retry purposes. 497// 498// Experimental 499// 500// Notice: This API is EXPERIMENTAL and may be changed or removed in a 501// later release. 502func MaxRetryRPCBufferSize(bytes int) CallOption { 503 return MaxRetryRPCBufferSizeCallOption{bytes} 504} 505 506// MaxRetryRPCBufferSizeCallOption is a CallOption indicating the amount of 507// memory to be used for caching this RPC for retry purposes. 508// 509// Experimental 510// 511// Notice: This type is EXPERIMENTAL and may be changed or removed in a 512// later release. 513type MaxRetryRPCBufferSizeCallOption struct { 514 MaxRetryRPCBufferSize int 515} 516 517func (o MaxRetryRPCBufferSizeCallOption) before(c *callInfo) error { 518 c.maxRetryRPCBufferSize = o.MaxRetryRPCBufferSize 519 return nil 520} 521func (o MaxRetryRPCBufferSizeCallOption) after(c *callInfo, attempt *csAttempt) {} 522 523// The format of the payload: compressed or not? 524type payloadFormat uint8 525 526const ( 527 compressionNone payloadFormat = 0 // no compression 528 compressionMade payloadFormat = 1 // compressed 529) 530 531// parser reads complete gRPC messages from the underlying reader. 532type parser struct { 533 // r is the underlying reader. 534 // See the comment on recvMsg for the permissible 535 // error types. 536 r io.Reader 537 538 // The header of a gRPC message. Find more detail at 539 // https://github.com/grpc/grpc/blob/master/doc/PROTOCOL-HTTP2.md 540 header [5]byte 541} 542 543// recvMsg reads a complete gRPC message from the stream. 544// 545// It returns the message and its payload (compression/encoding) 546// format. The caller owns the returned msg memory. 547// 548// If there is an error, possible values are: 549// * io.EOF, when no messages remain 550// * io.ErrUnexpectedEOF 551// * of type transport.ConnectionError 552// * an error from the status package 553// No other error values or types must be returned, which also means 554// that the underlying io.Reader must not return an incompatible 555// error. 556func (p *parser) recvMsg(maxReceiveMessageSize int) (pf payloadFormat, msg []byte, err error) { 557 if _, err := p.r.Read(p.header[:]); err != nil { 558 return 0, nil, err 559 } 560 561 pf = payloadFormat(p.header[0]) 562 length := binary.BigEndian.Uint32(p.header[1:]) 563 564 if length == 0 { 565 return pf, nil, nil 566 } 567 if int64(length) > int64(maxInt) { 568 return 0, nil, status.Errorf(codes.ResourceExhausted, "grpc: received message larger than max length allowed on current machine (%d vs. %d)", length, maxInt) 569 } 570 if int(length) > maxReceiveMessageSize { 571 return 0, nil, status.Errorf(codes.ResourceExhausted, "grpc: received message larger than max (%d vs. %d)", length, maxReceiveMessageSize) 572 } 573 // TODO(bradfitz,zhaoq): garbage. reuse buffer after proto decoding instead 574 // of making it for each message: 575 msg = make([]byte, int(length)) 576 if _, err := p.r.Read(msg); err != nil { 577 if err == io.EOF { 578 err = io.ErrUnexpectedEOF 579 } 580 return 0, nil, err 581 } 582 return pf, msg, nil 583} 584 585// encode serializes msg and returns a buffer containing the message, or an 586// error if it is too large to be transmitted by grpc. If msg is nil, it 587// generates an empty message. 588func encode(c baseCodec, msg interface{}) ([]byte, error) { 589 if msg == nil { // NOTE: typed nils will not be caught by this check 590 return nil, nil 591 } 592 b, err := c.Marshal(msg) 593 if err != nil { 594 return nil, status.Errorf(codes.Internal, "grpc: error while marshaling: %v", err.Error()) 595 } 596 if uint(len(b)) > math.MaxUint32 { 597 return nil, status.Errorf(codes.ResourceExhausted, "grpc: message too large (%d bytes)", len(b)) 598 } 599 return b, nil 600} 601 602// compress returns the input bytes compressed by compressor or cp. If both 603// compressors are nil, returns nil. 604// 605// TODO(dfawley): eliminate cp parameter by wrapping Compressor in an encoding.Compressor. 606func compress(in []byte, cp Compressor, compressor encoding.Compressor) ([]byte, error) { 607 if compressor == nil && cp == nil { 608 return nil, nil 609 } 610 wrapErr := func(err error) error { 611 return status.Errorf(codes.Internal, "grpc: error while compressing: %v", err.Error()) 612 } 613 cbuf := &bytes.Buffer{} 614 if compressor != nil { 615 z, err := compressor.Compress(cbuf) 616 if err != nil { 617 return nil, wrapErr(err) 618 } 619 if _, err := z.Write(in); err != nil { 620 return nil, wrapErr(err) 621 } 622 if err := z.Close(); err != nil { 623 return nil, wrapErr(err) 624 } 625 } else { 626 if err := cp.Do(cbuf, in); err != nil { 627 return nil, wrapErr(err) 628 } 629 } 630 return cbuf.Bytes(), nil 631} 632 633const ( 634 payloadLen = 1 635 sizeLen = 4 636 headerLen = payloadLen + sizeLen 637) 638 639// msgHeader returns a 5-byte header for the message being transmitted and the 640// payload, which is compData if non-nil or data otherwise. 641func msgHeader(data, compData []byte) (hdr []byte, payload []byte) { 642 hdr = make([]byte, headerLen) 643 if compData != nil { 644 hdr[0] = byte(compressionMade) 645 data = compData 646 } else { 647 hdr[0] = byte(compressionNone) 648 } 649 650 // Write length of payload into buf 651 binary.BigEndian.PutUint32(hdr[payloadLen:], uint32(len(data))) 652 return hdr, data 653} 654 655func outPayload(client bool, msg interface{}, data, payload []byte, t time.Time) *stats.OutPayload { 656 return &stats.OutPayload{ 657 Client: client, 658 Payload: msg, 659 Data: data, 660 Length: len(data), 661 WireLength: len(payload) + headerLen, 662 SentTime: t, 663 } 664} 665 666func checkRecvPayload(pf payloadFormat, recvCompress string, haveCompressor bool) *status.Status { 667 switch pf { 668 case compressionNone: 669 case compressionMade: 670 if recvCompress == "" || recvCompress == encoding.Identity { 671 return status.New(codes.Internal, "grpc: compressed flag set with identity or empty encoding") 672 } 673 if !haveCompressor { 674 return status.Newf(codes.Unimplemented, "grpc: Decompressor is not installed for grpc-encoding %q", recvCompress) 675 } 676 default: 677 return status.Newf(codes.Internal, "grpc: received unexpected payload format %d", pf) 678 } 679 return nil 680} 681 682type payloadInfo struct { 683 wireLength int // The compressed length got from wire. 684 uncompressedBytes []byte 685} 686 687func recvAndDecompress(p *parser, s *transport.Stream, dc Decompressor, maxReceiveMessageSize int, payInfo *payloadInfo, compressor encoding.Compressor) ([]byte, error) { 688 pf, d, err := p.recvMsg(maxReceiveMessageSize) 689 if err != nil { 690 return nil, err 691 } 692 if payInfo != nil { 693 payInfo.wireLength = len(d) 694 } 695 696 if st := checkRecvPayload(pf, s.RecvCompress(), compressor != nil || dc != nil); st != nil { 697 return nil, st.Err() 698 } 699 700 var size int 701 if pf == compressionMade { 702 // To match legacy behavior, if the decompressor is set by WithDecompressor or RPCDecompressor, 703 // use this decompressor as the default. 704 if dc != nil { 705 d, err = dc.Do(bytes.NewReader(d)) 706 size = len(d) 707 } else { 708 d, size, err = decompress(compressor, d, maxReceiveMessageSize) 709 } 710 if err != nil { 711 return nil, status.Errorf(codes.Internal, "grpc: failed to decompress the received message %v", err) 712 } 713 } else { 714 size = len(d) 715 } 716 if size > maxReceiveMessageSize { 717 // TODO: Revisit the error code. Currently keep it consistent with java 718 // implementation. 719 return nil, status.Errorf(codes.ResourceExhausted, "grpc: received message larger than max (%d vs. %d)", size, maxReceiveMessageSize) 720 } 721 return d, nil 722} 723 724// Using compressor, decompress d, returning data and size. 725// Optionally, if data will be over maxReceiveMessageSize, just return the size. 726func decompress(compressor encoding.Compressor, d []byte, maxReceiveMessageSize int) ([]byte, int, error) { 727 dcReader, err := compressor.Decompress(bytes.NewReader(d)) 728 if err != nil { 729 return nil, 0, err 730 } 731 if sizer, ok := compressor.(interface { 732 DecompressedSize(compressedBytes []byte) int 733 }); ok { 734 if size := sizer.DecompressedSize(d); size >= 0 { 735 if size > maxReceiveMessageSize { 736 return nil, size, nil 737 } 738 // size is used as an estimate to size the buffer, but we 739 // will read more data if available. 740 // +MinRead so ReadFrom will not reallocate if size is correct. 741 buf := bytes.NewBuffer(make([]byte, 0, size+bytes.MinRead)) 742 bytesRead, err := buf.ReadFrom(io.LimitReader(dcReader, int64(maxReceiveMessageSize)+1)) 743 return buf.Bytes(), int(bytesRead), err 744 } 745 } 746 // Read from LimitReader with limit max+1. So if the underlying 747 // reader is over limit, the result will be bigger than max. 748 d, err = ioutil.ReadAll(io.LimitReader(dcReader, int64(maxReceiveMessageSize)+1)) 749 return d, len(d), err 750} 751 752// For the two compressor parameters, both should not be set, but if they are, 753// dc takes precedence over compressor. 754// TODO(dfawley): wrap the old compressor/decompressor using the new API? 755func recv(p *parser, c baseCodec, s *transport.Stream, dc Decompressor, m interface{}, maxReceiveMessageSize int, payInfo *payloadInfo, compressor encoding.Compressor) error { 756 d, err := recvAndDecompress(p, s, dc, maxReceiveMessageSize, payInfo, compressor) 757 if err != nil { 758 return err 759 } 760 if err := c.Unmarshal(d, m); err != nil { 761 return status.Errorf(codes.Internal, "grpc: failed to unmarshal the received message %v", err) 762 } 763 if payInfo != nil { 764 payInfo.uncompressedBytes = d 765 } 766 return nil 767} 768 769// Information about RPC 770type rpcInfo struct { 771 failfast bool 772 preloaderInfo *compressorInfo 773} 774 775// Information about Preloader 776// Responsible for storing codec, and compressors 777// If stream (s) has context s.Context which stores rpcInfo that has non nil 778// pointers to codec, and compressors, then we can use preparedMsg for Async message prep 779// and reuse marshalled bytes 780type compressorInfo struct { 781 codec baseCodec 782 cp Compressor 783 comp encoding.Compressor 784} 785 786type rpcInfoContextKey struct{} 787 788func newContextWithRPCInfo(ctx context.Context, failfast bool, codec baseCodec, cp Compressor, comp encoding.Compressor) context.Context { 789 return context.WithValue(ctx, rpcInfoContextKey{}, &rpcInfo{ 790 failfast: failfast, 791 preloaderInfo: &compressorInfo{ 792 codec: codec, 793 cp: cp, 794 comp: comp, 795 }, 796 }) 797} 798 799func rpcInfoFromContext(ctx context.Context) (s *rpcInfo, ok bool) { 800 s, ok = ctx.Value(rpcInfoContextKey{}).(*rpcInfo) 801 return 802} 803 804// Code returns the error code for err if it was produced by the rpc system. 805// Otherwise, it returns codes.Unknown. 806// 807// Deprecated: use status.Code instead. 808func Code(err error) codes.Code { 809 return status.Code(err) 810} 811 812// ErrorDesc returns the error description of err if it was produced by the rpc system. 813// Otherwise, it returns err.Error() or empty string when err is nil. 814// 815// Deprecated: use status.Convert and Message method instead. 816func ErrorDesc(err error) string { 817 return status.Convert(err).Message() 818} 819 820// Errorf returns an error containing an error code and a description; 821// Errorf returns nil if c is OK. 822// 823// Deprecated: use status.Errorf instead. 824func Errorf(c codes.Code, format string, a ...interface{}) error { 825 return status.Errorf(c, format, a...) 826} 827 828// toRPCErr converts an error into an error from the status package. 829func toRPCErr(err error) error { 830 if err == nil || err == io.EOF { 831 return err 832 } 833 if err == io.ErrUnexpectedEOF { 834 return status.Error(codes.Internal, err.Error()) 835 } 836 if _, ok := status.FromError(err); ok { 837 return err 838 } 839 switch e := err.(type) { 840 case transport.ConnectionError: 841 return status.Error(codes.Unavailable, e.Desc) 842 default: 843 switch err { 844 case context.DeadlineExceeded: 845 return status.Error(codes.DeadlineExceeded, err.Error()) 846 case context.Canceled: 847 return status.Error(codes.Canceled, err.Error()) 848 } 849 } 850 return status.Error(codes.Unknown, err.Error()) 851} 852 853// setCallInfoCodec should only be called after CallOptions have been applied. 854func setCallInfoCodec(c *callInfo) error { 855 if c.codec != nil { 856 // codec was already set by a CallOption; use it. 857 return nil 858 } 859 860 if c.contentSubtype == "" { 861 // No codec specified in CallOptions; use proto by default. 862 c.codec = encoding.GetCodec(proto.Name) 863 return nil 864 } 865 866 // c.contentSubtype is already lowercased in CallContentSubtype 867 c.codec = encoding.GetCodec(c.contentSubtype) 868 if c.codec == nil { 869 return status.Errorf(codes.Internal, "no codec registered for content-subtype %s", c.contentSubtype) 870 } 871 return nil 872} 873 874// channelzData is used to store channelz related data for ClientConn, addrConn and Server. 875// These fields cannot be embedded in the original structs (e.g. ClientConn), since to do atomic 876// operation on int64 variable on 32-bit machine, user is responsible to enforce memory alignment. 877// Here, by grouping those int64 fields inside a struct, we are enforcing the alignment. 878type channelzData struct { 879 callsStarted int64 880 callsFailed int64 881 callsSucceeded int64 882 // lastCallStartedTime stores the timestamp that last call starts. It is of int64 type instead of 883 // time.Time since it's more costly to atomically update time.Time variable than int64 variable. 884 lastCallStartedTime int64 885} 886 887// The SupportPackageIsVersion variables are referenced from generated protocol 888// buffer files to ensure compatibility with the gRPC version used. The latest 889// support package version is 7. 890// 891// Older versions are kept for compatibility. 892// 893// These constants should not be referenced from any other code. 894const ( 895 SupportPackageIsVersion3 = true 896 SupportPackageIsVersion4 = true 897 SupportPackageIsVersion5 = true 898 SupportPackageIsVersion6 = true 899 SupportPackageIsVersion7 = true 900) 901 902const grpcUA = "grpc-go/" + Version 903