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