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