--- /dev/null
+/*
+ *
+ * Copyright 2014 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package grpc
+
+import (
+ "bytes"
+ "compress/gzip"
+ "encoding/binary"
+ "io"
+ "io/ioutil"
+ "math"
+ "sync"
+ "time"
+
+ "golang.org/x/net/context"
+ "google.golang.org/grpc/codes"
+ "google.golang.org/grpc/credentials"
+ "google.golang.org/grpc/metadata"
+ "google.golang.org/grpc/peer"
+ "google.golang.org/grpc/stats"
+ "google.golang.org/grpc/status"
+ "google.golang.org/grpc/transport"
+)
+
+// Compressor defines the interface gRPC uses to compress a message.
+type Compressor interface {
+ // Do compresses p into w.
+ Do(w io.Writer, p []byte) error
+ // Type returns the compression algorithm the Compressor uses.
+ Type() string
+}
+
+type gzipCompressor struct {
+ pool sync.Pool
+}
+
+// NewGZIPCompressor creates a Compressor based on GZIP.
+func NewGZIPCompressor() Compressor {
+ return &gzipCompressor{
+ pool: sync.Pool{
+ New: func() interface{} {
+ return gzip.NewWriter(ioutil.Discard)
+ },
+ },
+ }
+}
+
+func (c *gzipCompressor) Do(w io.Writer, p []byte) error {
+ z := c.pool.Get().(*gzip.Writer)
+ defer c.pool.Put(z)
+ z.Reset(w)
+ if _, err := z.Write(p); err != nil {
+ return err
+ }
+ return z.Close()
+}
+
+func (c *gzipCompressor) Type() string {
+ return "gzip"
+}
+
+// Decompressor defines the interface gRPC uses to decompress a message.
+type Decompressor interface {
+ // Do reads the data from r and uncompress them.
+ Do(r io.Reader) ([]byte, error)
+ // Type returns the compression algorithm the Decompressor uses.
+ Type() string
+}
+
+type gzipDecompressor struct {
+ pool sync.Pool
+}
+
+// NewGZIPDecompressor creates a Decompressor based on GZIP.
+func NewGZIPDecompressor() Decompressor {
+ return &gzipDecompressor{}
+}
+
+func (d *gzipDecompressor) Do(r io.Reader) ([]byte, error) {
+ var z *gzip.Reader
+ switch maybeZ := d.pool.Get().(type) {
+ case nil:
+ newZ, err := gzip.NewReader(r)
+ if err != nil {
+ return nil, err
+ }
+ z = newZ
+ case *gzip.Reader:
+ z = maybeZ
+ if err := z.Reset(r); err != nil {
+ d.pool.Put(z)
+ return nil, err
+ }
+ }
+
+ defer func() {
+ z.Close()
+ d.pool.Put(z)
+ }()
+ return ioutil.ReadAll(z)
+}
+
+func (d *gzipDecompressor) Type() string {
+ return "gzip"
+}
+
+// callInfo contains all related configuration and information about an RPC.
+type callInfo struct {
+ failFast bool
+ headerMD metadata.MD
+ trailerMD metadata.MD
+ peer *peer.Peer
+ traceInfo traceInfo // in trace.go
+ maxReceiveMessageSize *int
+ maxSendMessageSize *int
+ creds credentials.PerRPCCredentials
+}
+
+func defaultCallInfo() *callInfo {
+ return &callInfo{failFast: true}
+}
+
+// CallOption configures a Call before it starts or extracts information from
+// a Call after it completes.
+type CallOption interface {
+ // before is called before the call is sent to any server. If before
+ // returns a non-nil error, the RPC fails with that error.
+ before(*callInfo) error
+
+ // after is called after the call has completed. after cannot return an
+ // error, so any failures should be reported via output parameters.
+ after(*callInfo)
+}
+
+// EmptyCallOption does not alter the Call configuration.
+// It can be embedded in another structure to carry satellite data for use
+// by interceptors.
+type EmptyCallOption struct{}
+
+func (EmptyCallOption) before(*callInfo) error { return nil }
+func (EmptyCallOption) after(*callInfo) {}
+
+type beforeCall func(c *callInfo) error
+
+func (o beforeCall) before(c *callInfo) error { return o(c) }
+func (o beforeCall) after(c *callInfo) {}
+
+type afterCall func(c *callInfo)
+
+func (o afterCall) before(c *callInfo) error { return nil }
+func (o afterCall) after(c *callInfo) { o(c) }
+
+// Header returns a CallOptions that retrieves the header metadata
+// for a unary RPC.
+func Header(md *metadata.MD) CallOption {
+ return afterCall(func(c *callInfo) {
+ *md = c.headerMD
+ })
+}
+
+// Trailer returns a CallOptions that retrieves the trailer metadata
+// for a unary RPC.
+func Trailer(md *metadata.MD) CallOption {
+ return afterCall(func(c *callInfo) {
+ *md = c.trailerMD
+ })
+}
+
+// Peer returns a CallOption that retrieves peer information for a
+// unary RPC.
+func Peer(peer *peer.Peer) CallOption {
+ return afterCall(func(c *callInfo) {
+ if c.peer != nil {
+ *peer = *c.peer
+ }
+ })
+}
+
+// FailFast configures the action to take when an RPC is attempted on broken
+// connections or unreachable servers. If failFast is true, the RPC will fail
+// immediately. Otherwise, the RPC client will block the call until a
+// connection is available (or the call is canceled or times out). Please refer
+// to https://github.com/grpc/grpc/blob/master/doc/wait-for-ready.md.
+// The default behavior of RPCs is to fail fast.
+func FailFast(failFast bool) CallOption {
+ return beforeCall(func(c *callInfo) error {
+ c.failFast = failFast
+ return nil
+ })
+}
+
+// MaxCallRecvMsgSize returns a CallOption which sets the maximum message size the client can receive.
+func MaxCallRecvMsgSize(s int) CallOption {
+ return beforeCall(func(o *callInfo) error {
+ o.maxReceiveMessageSize = &s
+ return nil
+ })
+}
+
+// MaxCallSendMsgSize returns a CallOption which sets the maximum message size the client can send.
+func MaxCallSendMsgSize(s int) CallOption {
+ return beforeCall(func(o *callInfo) error {
+ o.maxSendMessageSize = &s
+ return nil
+ })
+}
+
+// PerRPCCredentials returns a CallOption that sets credentials.PerRPCCredentials
+// for a call.
+func PerRPCCredentials(creds credentials.PerRPCCredentials) CallOption {
+ return beforeCall(func(c *callInfo) error {
+ c.creds = creds
+ return nil
+ })
+}
+
+// The format of the payload: compressed or not?
+type payloadFormat uint8
+
+const (
+ compressionNone payloadFormat = iota // no compression
+ compressionMade
+)
+
+// parser reads complete gRPC messages from the underlying reader.
+type parser struct {
+ // r is the underlying reader.
+ // See the comment on recvMsg for the permissible
+ // error types.
+ r io.Reader
+
+ // The header of a gRPC message. Find more detail
+ // at https://grpc.io/docs/guides/wire.html.
+ header [5]byte
+}
+
+// recvMsg reads a complete gRPC message from the stream.
+//
+// It returns the message and its payload (compression/encoding)
+// format. The caller owns the returned msg memory.
+//
+// If there is an error, possible values are:
+// * io.EOF, when no messages remain
+// * io.ErrUnexpectedEOF
+// * of type transport.ConnectionError
+// * of type transport.StreamError
+// No other error values or types must be returned, which also means
+// that the underlying io.Reader must not return an incompatible
+// error.
+func (p *parser) recvMsg(maxReceiveMessageSize int) (pf payloadFormat, msg []byte, err error) {
+ if _, err := p.r.Read(p.header[:]); err != nil {
+ return 0, nil, err
+ }
+
+ pf = payloadFormat(p.header[0])
+ length := binary.BigEndian.Uint32(p.header[1:])
+
+ if length == 0 {
+ return pf, nil, nil
+ }
+ if int64(length) > int64(maxInt) {
+ return 0, nil, Errorf(codes.ResourceExhausted, "grpc: received message larger than max length allowed on current machine (%d vs. %d)", length, maxInt)
+ }
+ if int(length) > maxReceiveMessageSize {
+ return 0, nil, Errorf(codes.ResourceExhausted, "grpc: received message larger than max (%d vs. %d)", length, maxReceiveMessageSize)
+ }
+ // TODO(bradfitz,zhaoq): garbage. reuse buffer after proto decoding instead
+ // of making it for each message:
+ msg = make([]byte, int(length))
+ if _, err := p.r.Read(msg); err != nil {
+ if err == io.EOF {
+ err = io.ErrUnexpectedEOF
+ }
+ return 0, nil, err
+ }
+ return pf, msg, nil
+}
+
+// encode serializes msg and returns a buffer of message header and a buffer of msg.
+// If msg is nil, it generates the message header and an empty msg buffer.
+func encode(c Codec, msg interface{}, cp Compressor, cbuf *bytes.Buffer, outPayload *stats.OutPayload) ([]byte, []byte, error) {
+ var b []byte
+ const (
+ payloadLen = 1
+ sizeLen = 4
+ )
+
+ if msg != nil {
+ var err error
+ b, err = c.Marshal(msg)
+ if err != nil {
+ return nil, nil, Errorf(codes.Internal, "grpc: error while marshaling: %v", err.Error())
+ }
+ if outPayload != nil {
+ outPayload.Payload = msg
+ // TODO truncate large payload.
+ outPayload.Data = b
+ outPayload.Length = len(b)
+ }
+ if cp != nil {
+ if err := cp.Do(cbuf, b); err != nil {
+ return nil, nil, Errorf(codes.Internal, "grpc: error while compressing: %v", err.Error())
+ }
+ b = cbuf.Bytes()
+ }
+ }
+
+ if uint(len(b)) > math.MaxUint32 {
+ return nil, nil, Errorf(codes.ResourceExhausted, "grpc: message too large (%d bytes)", len(b))
+ }
+
+ bufHeader := make([]byte, payloadLen+sizeLen)
+ if cp == nil {
+ bufHeader[0] = byte(compressionNone)
+ } else {
+ bufHeader[0] = byte(compressionMade)
+ }
+ // Write length of b into buf
+ binary.BigEndian.PutUint32(bufHeader[payloadLen:], uint32(len(b)))
+ if outPayload != nil {
+ outPayload.WireLength = payloadLen + sizeLen + len(b)
+ }
+ return bufHeader, b, nil
+}
+
+func checkRecvPayload(pf payloadFormat, recvCompress string, dc Decompressor) error {
+ switch pf {
+ case compressionNone:
+ case compressionMade:
+ if dc == nil || recvCompress != dc.Type() {
+ return Errorf(codes.Unimplemented, "grpc: Decompressor is not installed for grpc-encoding %q", recvCompress)
+ }
+ default:
+ return Errorf(codes.Internal, "grpc: received unexpected payload format %d", pf)
+ }
+ return nil
+}
+
+func recv(p *parser, c Codec, s *transport.Stream, dc Decompressor, m interface{}, maxReceiveMessageSize int, inPayload *stats.InPayload) error {
+ pf, d, err := p.recvMsg(maxReceiveMessageSize)
+ if err != nil {
+ return err
+ }
+ if inPayload != nil {
+ inPayload.WireLength = len(d)
+ }
+ if err := checkRecvPayload(pf, s.RecvCompress(), dc); err != nil {
+ return err
+ }
+ if pf == compressionMade {
+ d, err = dc.Do(bytes.NewReader(d))
+ if err != nil {
+ return Errorf(codes.Internal, "grpc: failed to decompress the received message %v", err)
+ }
+ }
+ if len(d) > maxReceiveMessageSize {
+ // TODO: Revisit the error code. Currently keep it consistent with java
+ // implementation.
+ return Errorf(codes.ResourceExhausted, "grpc: received message larger than max (%d vs. %d)", len(d), maxReceiveMessageSize)
+ }
+ if err := c.Unmarshal(d, m); err != nil {
+ return Errorf(codes.Internal, "grpc: failed to unmarshal the received message %v", err)
+ }
+ if inPayload != nil {
+ inPayload.RecvTime = time.Now()
+ inPayload.Payload = m
+ // TODO truncate large payload.
+ inPayload.Data = d
+ inPayload.Length = len(d)
+ }
+ return nil
+}
+
+type rpcInfo struct {
+ failfast bool
+ bytesSent bool
+ bytesReceived bool
+}
+
+type rpcInfoContextKey struct{}
+
+func newContextWithRPCInfo(ctx context.Context, failfast bool) context.Context {
+ return context.WithValue(ctx, rpcInfoContextKey{}, &rpcInfo{failfast: failfast})
+}
+
+func rpcInfoFromContext(ctx context.Context) (s *rpcInfo, ok bool) {
+ s, ok = ctx.Value(rpcInfoContextKey{}).(*rpcInfo)
+ return
+}
+
+func updateRPCInfoInContext(ctx context.Context, s rpcInfo) {
+ if ss, ok := rpcInfoFromContext(ctx); ok {
+ ss.bytesReceived = s.bytesReceived
+ ss.bytesSent = s.bytesSent
+ }
+ return
+}
+
+// Code returns the error code for err if it was produced by the rpc system.
+// Otherwise, it returns codes.Unknown.
+//
+// Deprecated; use status.FromError and Code method instead.
+func Code(err error) codes.Code {
+ if s, ok := status.FromError(err); ok {
+ return s.Code()
+ }
+ return codes.Unknown
+}
+
+// ErrorDesc returns the error description of err if it was produced by the rpc system.
+// Otherwise, it returns err.Error() or empty string when err is nil.
+//
+// Deprecated; use status.FromError and Message method instead.
+func ErrorDesc(err error) string {
+ if s, ok := status.FromError(err); ok {
+ return s.Message()
+ }
+ return err.Error()
+}
+
+// Errorf returns an error containing an error code and a description;
+// Errorf returns nil if c is OK.
+//
+// Deprecated; use status.Errorf instead.
+func Errorf(c codes.Code, format string, a ...interface{}) error {
+ return status.Errorf(c, format, a...)
+}
+
+// The SupportPackageIsVersion variables are referenced from generated protocol
+// buffer files to ensure compatibility with the gRPC version used. The latest
+// support package version is 5.
+//
+// Older versions are kept for compatibility. They may be removed if
+// compatibility cannot be maintained.
+//
+// These constants should not be referenced from any other code.
+const (
+ SupportPackageIsVersion3 = true
+ SupportPackageIsVersion4 = true
+ SupportPackageIsVersion5 = true
+)
+
+// Version is the current grpc version.
+const Version = "1.8.0-dev"
+
+const grpcUA = "grpc-go/" + Version
OSDN Git Service
