Merge branch 'go-1.5-compat' into 'master'

Downgrade grpc to get Go 1.5 compatibility

Closes #85

See merge request !133

.gitlab-ci.yml

@@ -44,12 +44,29 @@ rspec:ruby2.1:
   except:
     - tags
 
-go:
-  # Image taken from gitlab-ce@59f81b4ff8
-  image: "dev.gitlab.org:5005/gitlab/gitlab-build-images:ruby-2.3.3-golang-1.8-git-2.7-phantomjs-2.1-node-7.1"
+.go: &go_definition
+  before_script:
+  - apt-get update -qq && apt-get install -y ruby
+  - ruby -v
   script:
     - go version
     - which go
     - bin/compile
     - support/go-test
     - support/go-format check
+
+go:1.8:
+  <<: *go_definition
+  image: golang:1.8
+
+go:1.7:
+  <<: *go_definition
+  image: golang:1.7
+
+go:1.6:
+  <<: *go_definition
+  image: golang:1.6
+
+go:1.5:
+  <<: *go_definition
+  image: golang:1.5

go/vendor/google.golang.org/grpc/README.md

-# gRPC-Go
+#gRPC-Go
 
 [![Build Status](https://travis-ci.org/grpc/grpc-go.svg)](https://travis-ci.org/grpc/grpc-go) [![GoDoc](https://godoc.org/google.golang.org/grpc?status.svg)](https://godoc.org/google.golang.org/grpc)
 
@@ -16,7 +16,23 @@ $ go get google.golang.org/grpc
 Prerequisites
 -------------
 
-This requires Go 1.6 or later.
+This requires Go 1.5 or later.
+
+A note on the version used: significant performance improvements in benchmarks
+of grpc-go have been seen by upgrading the go version from 1.5 to the latest
+1.7.1.
+
+From https://golang.org/doc/install, one way to install the latest version of go is:
+```
+$ GO_VERSION=1.7.1
+$ OS=linux
+$ ARCH=amd64
+$ curl -O https://storage.googleapis.com/golang/go${GO_VERSION}.${OS}-${ARCH}.tar.gz
+$ sudo tar -C /usr/local -xzf go$GO_VERSION.$OS-$ARCH.tar.gz
+$ # Put go on the PATH, keep the usual installation dir
+$ sudo ln -s /usr/local/go/bin/go /usr/bin/go
+$ rm go$GO_VERSION.$OS-$ARCH.tar.gz
+```
 
 Constraints
 -----------

go/vendor/google.golang.org/grpc/call.go

@@ -36,14 +36,13 @@ package grpc
 import (
 	"bytes"
 	"io"
+	"math"
 	"time"
 
 	"golang.org/x/net/context"
 	"golang.org/x/net/trace"
 	"google.golang.org/grpc/codes"
-	"google.golang.org/grpc/peer"
 	"google.golang.org/grpc/stats"
-	"google.golang.org/grpc/status"
 	"google.golang.org/grpc/transport"
 )
 
@@ -73,22 +72,19 @@ func recvResponse(ctx context.Context, dopts dialOptions, t transport.ClientTran
 		}
 	}
 	for {
-		if err = recv(p, dopts.codec, stream, dopts.dc, reply, dopts.maxMsgSize, inPayload); err != nil {
+		if err = recv(p, dopts.codec, stream, dopts.dc, reply, math.MaxInt32, inPayload); err != nil {
 			if err == io.EOF {
 				break
 			}
 			return
 		}
 	}
-	if inPayload != nil && err == io.EOF && stream.Status().Code() == codes.OK {
+	if inPayload != nil && err == io.EOF && stream.StatusCode() == codes.OK {
 		// TODO in the current implementation, inTrailer may be handled before inPayload in some cases.
 		// Fix the order if necessary.
 		dopts.copts.StatsHandler.HandleRPC(ctx, inPayload)
 	}
 	c.trailerMD = stream.Trailer()
-	if peer, ok := peer.FromContext(stream.Context()); ok {
-		c.peer = peer
-	}
 	return nil
 }
 
@@ -231,7 +227,7 @@ func invoke(ctx context.Context, method string, args, reply interface{}, cc *Cli
 		t, put, err = cc.getTransport(ctx, gopts)
 		if err != nil {
 			// TODO(zhaoq): Probably revisit the error handling.
-			if _, ok := status.FromError(err); ok {
+			if _, ok := err.(*rpcError); ok {
 				return err
 			}
 			if err == errConnClosing || err == errConnUnavailable {
@@ -285,6 +281,6 @@ func invoke(ctx context.Context, method string, args, reply interface{}, cc *Cli
 			put()
 			put = nil
 		}
-		return stream.Status().Err()
+		return Errorf(stream.StatusCode(), "%s", stream.StatusDesc())
 	}
 }

go/vendor/google.golang.org/grpc/clientconn.go

@@ -36,8 +36,8 @@ package grpc
 import (
 	"errors"
 	"fmt"
-	"math"
 	"net"
+	"strings"
 	"sync"
 	"time"
 
@@ -45,7 +45,6 @@ import (
 	"golang.org/x/net/trace"
 	"google.golang.org/grpc/credentials"
 	"google.golang.org/grpc/grpclog"
-	"google.golang.org/grpc/keepalive"
 	"google.golang.org/grpc/stats"
 	"google.golang.org/grpc/transport"
 )
@@ -79,6 +78,7 @@ var (
 	errConnClosing = errors.New("grpc: the connection is closing")
 	// errConnUnavailable indicates that the connection is unavailable.
 	errConnUnavailable = errors.New("grpc: the connection is unavailable")
+	errNoAddr          = errors.New("grpc: there is no address available to dial")
 	// minimum time to give a connection to complete
 	minConnectTimeout = 20 * time.Second
 )
@@ -98,21 +98,11 @@ type dialOptions struct {
 	timeout   time.Duration
 	scChan    <-chan ServiceConfig
 	copts     transport.ConnectOptions
-	maxMsgSize int
 }
 
-const defaultClientMaxMsgSize = math.MaxInt32
-
 // DialOption configures how we set up the connection.
 type DialOption func(*dialOptions)
 
-// WithMaxMsgSize returns a DialOption which sets the maximum message size the client can receive.
-func WithMaxMsgSize(s int) DialOption {
-	return func(o *dialOptions) {
-		o.maxMsgSize = s
-	}
-}
-
 // WithCodec returns a DialOption which sets a codec for message marshaling and unmarshaling.
 func WithCodec(c Codec) DialOption {
 	return func(o *dialOptions) {
@@ -259,13 +249,6 @@ func WithUserAgent(s string) DialOption {
 	}
 }
 
-// WithKeepaliveParams returns a DialOption that specifies keepalive paramaters for the client transport.
-func WithKeepaliveParams(kp keepalive.ClientParameters) DialOption {
-	return func(o *dialOptions) {
-		o.copts.KeepaliveParams = kp
-	}
-}
-
 // WithUnaryInterceptor returns a DialOption that specifies the interceptor for unary RPCs.
 func WithUnaryInterceptor(f UnaryClientInterceptor) DialOption {
 	return func(o *dialOptions) {
@@ -280,15 +263,6 @@ func WithStreamInterceptor(f StreamClientInterceptor) DialOption {
 	}
 }
 
-// WithAuthority returns a DialOption that specifies the value to be used as
-// the :authority pseudo-header. This value only works with WithInsecure and
-// has no effect if TransportCredentials are present.
-func WithAuthority(a string) DialOption {
-	return func(o *dialOptions) {
-		o.copts.Authority = a
-	}
-}
-
 // Dial creates a client connection to the given target.
 func Dial(target string, opts ...DialOption) (*ClientConn, error) {
 	return DialContext(context.Background(), target, opts...)
@@ -305,19 +279,9 @@ func DialContext(ctx context.Context, target string, opts ...DialOption) (conn *
 		conns:  make(map[Address]*addrConn),
 	}
 	cc.ctx, cc.cancel = context.WithCancel(context.Background())
-	cc.dopts.maxMsgSize = defaultClientMaxMsgSize
 	for _, opt := range opts {
 		opt(&cc.dopts)
 	}
-	cc.mkp = cc.dopts.copts.KeepaliveParams
-
-	grpcUA := "grpc-go/" + Version
-	if cc.dopts.copts.UserAgent != "" {
-		cc.dopts.copts.UserAgent += " " + grpcUA
-	} else {
-		cc.dopts.copts.UserAgent = grpcUA
-	}
-
 	if cc.dopts.timeout > 0 {
 		var cancel context.CancelFunc
 		ctx, cancel = context.WithTimeout(ctx, cc.dopts.timeout)
@@ -357,18 +321,24 @@ func DialContext(ctx context.Context, target string, opts ...DialOption) (conn *
 	creds := cc.dopts.copts.TransportCredentials
 	if creds != nil && creds.Info().ServerName != "" {
 		cc.authority = creds.Info().ServerName
-	} else if cc.dopts.insecure && cc.dopts.copts.Authority != "" {
-		cc.authority = cc.dopts.copts.Authority
 	} else {
-		cc.authority = target
+		colonPos := strings.LastIndex(target, ":")
+		if colonPos == -1 {
+			colonPos = len(target)
 		}
+		cc.authority = target[:colonPos]
+	}
+	var ok bool
 	waitC := make(chan error, 1)
 	go func() {
-		defer close(waitC)
+		var addrs []Address
 		if cc.dopts.balancer == nil && cc.sc.LB != nil {
 			cc.dopts.balancer = cc.sc.LB
 		}
-		if cc.dopts.balancer != nil {
+		if cc.dopts.balancer == nil {
+			// Connect to target directly if balancer is nil.
+			addrs = append(addrs, Address{Addr: target})
+		} else {
 			var credsClone credentials.TransportCredentials
 			if creds != nil {
 				credsClone = creds.Clone()
@@ -381,22 +351,24 @@ func DialContext(ctx context.Context, target string, opts ...DialOption) (conn *
 				return
 			}
 			ch := cc.dopts.balancer.Notify()
-			if ch != nil {
-				if cc.dopts.block {
-					doneChan := make(chan struct{})
-					go cc.lbWatcher(doneChan)
-					<-doneChan
+			if ch == nil {
+				// There is no name resolver installed.
+				addrs = append(addrs, Address{Addr: target})
 			} else {
-					go cc.lbWatcher(nil)
-				}
+				addrs, ok = <-ch
+				if !ok || len(addrs) == 0 {
+					waitC <- errNoAddr
 					return
 				}
 			}
-		// No balancer, or no resolver within the balancer.  Connect directly.
-		if err := cc.resetAddrConn(Address{Addr: target}, cc.dopts.block, nil); err != nil {
+		}
+		for _, a := range addrs {
+			if err := cc.resetAddrConn(a, false, nil); err != nil {
 				waitC <- err
 				return
 			}
+		}
+		close(waitC)
 	}()
 	select {
 	case <-ctx.Done():
@@ -407,10 +379,15 @@ func DialContext(ctx context.Context, target string, opts ...DialOption) (conn *
 		}
 	}
 
+	// If balancer is nil or balancer.Notify() is nil, ok will be false here.
+	// The lbWatcher goroutine will not be created.
+	if ok {
+		go cc.lbWatcher()
+	}
+
 	if cc.dopts.scChan != nil {
 		go cc.scWatcher()
 	}
-
 	return cc, nil
 }
 
@@ -459,14 +436,9 @@ type ClientConn struct {
 	mu    sync.RWMutex
 	sc    ServiceConfig
 	conns map[Address]*addrConn
-	// Keepalive parameter can be udated if a GoAway is received.
-	mkp keepalive.ClientParameters
 }
 
-// lbWatcher watches the Notify channel of the balancer in cc and manages
-// connections accordingly.  If doneChan is not nil, it is closed after the
-// first successfull connection is made.
-func (cc *ClientConn) lbWatcher(doneChan chan struct{}) {
+func (cc *ClientConn) lbWatcher() {
 	for addrs := range cc.dopts.balancer.Notify() {
 		var (
 			add []Address   // Addresses need to setup connections.
@@ -493,15 +465,7 @@ func (cc *ClientConn) lbWatcher(doneChan chan struct{}) {
 		}
 		cc.mu.Unlock()
 		for _, a := range add {
-			if doneChan != nil {
-				err := cc.resetAddrConn(a, true, nil)
-				if err == nil {
-					close(doneChan)
-					doneChan = nil
-				}
-			} else {
-				cc.resetAddrConn(a, false, nil)
-			}
+			cc.resetAddrConn(a, true, nil)
 		}
 		for _, c := range del {
 			c.tearDown(errConnDrain)
@@ -530,15 +494,12 @@ func (cc *ClientConn) scWatcher() {
 // resetAddrConn creates an addrConn for addr and adds it to cc.conns.
 // If there is an old addrConn for addr, it will be torn down, using tearDownErr as the reason.
 // If tearDownErr is nil, errConnDrain will be used instead.
-func (cc *ClientConn) resetAddrConn(addr Address, block bool, tearDownErr error) error {
+func (cc *ClientConn) resetAddrConn(addr Address, skipWait bool, tearDownErr error) error {
 	ac := &addrConn{
 		cc:    cc,
 		addr:  addr,
 		dopts: cc.dopts,
 	}
-	cc.mu.RLock()
-	ac.dopts.copts.KeepaliveParams = cc.mkp
-	cc.mu.RUnlock()
 	ac.ctx, ac.cancel = context.WithCancel(cc.ctx)
 	ac.stateCV = sync.NewCond(&ac.mu)
 	if EnableTracing {
@@ -583,7 +544,8 @@ func (cc *ClientConn) resetAddrConn(addr Address, block bool, tearDownErr error)
 			stale.tearDown(tearDownErr)
 		}
 	}
-	if block {
+	// skipWait may overwrite the decision in ac.dopts.block.
+	if ac.dopts.block && !skipWait {
 		if err := ac.resetTransport(false); err != nil {
 			if err != errConnClosing {
 				// Tear down ac and delete it from cc.conns.
@@ -720,20 +682,6 @@ type addrConn struct {
 	tearDownErr error
 }
 
-// adjustParams updates parameters used to create transports upon
-// receiving a GoAway.
-func (ac *addrConn) adjustParams(r transport.GoAwayReason) {
-	switch r {
-	case transport.TooManyPings:
-		v := 2 * ac.dopts.copts.KeepaliveParams.Time
-		ac.cc.mu.Lock()
-		if v > ac.cc.mkp.Time {
-			ac.cc.mkp.Time = v
-		}
-		ac.cc.mu.Unlock()
-	}
-}
-
 // printf records an event in ac's event log, unless ac has been closed.
 // REQUIRES ac.mu is held.
 func (ac *addrConn) printf(format string, a ...interface{}) {
@@ -818,8 +766,6 @@ func (ac *addrConn) resetTransport(closeTransport bool) error {
 			Metadata: ac.addr.Metadata,
 		}
 		newTransport, err := transport.NewClientTransport(ctx, sinfo, ac.dopts.copts)
-		// Don't call cancel in success path due to a race in Go 1.6:
-		// https://github.com/golang/go/issues/15078.
 		if err != nil {
 			cancel()
 
@@ -890,7 +836,6 @@ func (ac *addrConn) transportMonitor() {
 			}
 			return
 		case <-t.GoAway():
-			ac.adjustParams(t.GetGoAwayReason())
 			// If GoAway happens without any network I/O error, ac is closed without shutting down the
 			// underlying transport (the transport will be closed when all the pending RPCs finished or
 			// failed.).
@@ -899,9 +844,9 @@ func (ac *addrConn) transportMonitor() {
 			// In both cases, a new ac is created.
 			select {
 			case <-t.Error():
-				ac.cc.resetAddrConn(ac.addr, false, errNetworkIO)
+				ac.cc.resetAddrConn(ac.addr, true, errNetworkIO)
 			default:
-				ac.cc.resetAddrConn(ac.addr, false, errConnDrain)
+				ac.cc.resetAddrConn(ac.addr, true, errConnDrain)
 			}
 			return
 		case <-t.Error():
@@ -910,8 +855,7 @@ func (ac *addrConn) transportMonitor() {
 				t.Close()
 				return
 			case <-t.GoAway():
-				ac.adjustParams(t.GetGoAwayReason())
-				ac.cc.resetAddrConn(ac.addr, false, errNetworkIO)
+				ac.cc.resetAddrConn(ac.addr, true, errNetworkIO)
 				return
 			default:
 			}

go/vendor/google.golang.org/grpc/interceptor.go

@@ -40,7 +40,7 @@ import (
 // UnaryInvoker is called by UnaryClientInterceptor to complete RPCs.
 type UnaryInvoker func(ctx context.Context, method string, req, reply interface{}, cc *ClientConn, opts ...CallOption) error
 
-// UnaryClientInterceptor intercepts the execution of a unary RPC on the client. invoker is the handler to complete the RPC
+// UnaryClientInterceptor intercepts the execution of a unary RPC on the client. inovker is the handler to complete the RPC
 // and it is the responsibility of the interceptor to call it.
 // This is the EXPERIMENTAL API.
 type UnaryClientInterceptor func(ctx context.Context, method string, req, reply interface{}, cc *ClientConn, invoker UnaryInvoker, opts ...CallOption) error

go/vendor/google.golang.org/grpc/rpc_util.go

@@ -37,6 +37,7 @@ import (
 	"bytes"
 	"compress/gzip"
 	"encoding/binary"
+	"fmt"
 	"io"
 	"io/ioutil"
 	"math"
@@ -47,9 +48,7 @@ import (
 	"golang.org/x/net/context"
 	"google.golang.org/grpc/codes"
 	"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"
 )
 
@@ -141,7 +140,6 @@ type callInfo struct {
 	failFast  bool
 	headerMD  metadata.MD
 	trailerMD metadata.MD
-	peer      *peer.Peer
 	traceInfo traceInfo // in trace.go
 }
 
@@ -185,22 +183,12 @@ func Trailer(md *metadata.MD) CallOption {
 	})
 }
 
-// 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) and will retry
 // the call if it fails due to a transient error. Please refer to
-// https://github.com/grpc/grpc/blob/master/doc/fail_fast.md. Note: failFast is default to true.
+// https://github.com/grpc/grpc/blob/master/doc/fail_fast.md
 func FailFast(failFast bool) CallOption {
 	return beforeCall(func(c *callInfo) error {
 		c.failFast = failFast
@@ -372,57 +360,88 @@ func recv(p *parser, c Codec, s *transport.Stream, dc Decompressor, m interface{
 	return nil
 }
 
+// rpcError defines the status from an RPC.
+type rpcError struct {
+	code codes.Code
+	desc string
+}
+
+func (e *rpcError) Error() string {
+	return fmt.Sprintf("rpc error: code = %d desc = %s", e.code, e.desc)
+}
+
 // 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()
+	if err == nil {
+		return codes.OK
+	}
+	if e, ok := err.(*rpcError); ok {
+		return e.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()
+	if err == nil {
+		return ""
+	}
+	if e, ok := err.(*rpcError); ok {
+		return e.desc
 	}
 	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...)
+	if c == codes.OK {
+		return nil
+	}
+	return &rpcError{
+		code: c,
+		desc: fmt.Sprintf(format, a...),
+	}
 }
 
-// toRPCErr converts an error into an error from the status package.
+// toRPCErr converts an error into a rpcError.
 func toRPCErr(err error) error {
-	if _, ok := status.FromError(err); ok {
-		return err
-	}
 	switch e := err.(type) {
+	case *rpcError:
+		return err
 	case transport.StreamError:
-		return status.Error(e.Code, e.Desc)
+		return &rpcError{
+			code: e.Code,
+			desc: e.Desc,
+		}
 	case transport.ConnectionError:
-		return status.Error(codes.Internal, e.Desc)
+		return &rpcError{
+			code: codes.Internal,
+			desc: e.Desc,
+		}
 	default:
 		switch err {
 		case context.DeadlineExceeded:
-			return status.Error(codes.DeadlineExceeded, err.Error())
+			return &rpcError{
+				code: codes.DeadlineExceeded,
+				desc: err.Error(),
+			}
 		case context.Canceled:
-			return status.Error(codes.Canceled, err.Error())
+			return &rpcError{
+				code: codes.Canceled,
+				desc: err.Error(),
+			}
 		case ErrClientConnClosing:
-			return status.Error(codes.FailedPrecondition, err.Error())
+			return &rpcError{
+				code: codes.FailedPrecondition,
+				desc: err.Error(),
 			}
 		}
-	return status.Error(codes.Unknown, err.Error())
+
+	}
+	return Errorf(codes.Unknown, "%v", err)
 }
 
 // convertCode converts a standard Go error into its canonical code. Note that
@@ -467,17 +486,17 @@ type MethodConfig struct {
 	// then the other will be used.  If neither is set, then the RPC has no deadline.
 	Timeout time.Duration
 	// MaxReqSize is the maximum allowed payload size for an individual request in a
-	// stream (client->server) in bytes. The size which is measured is the serialized
-	// payload after per-message compression (but before stream compression) in bytes.
-	// The actual value used is the minumum of the value specified here and the value set
-	// by the application via the gRPC client API. If either one is not set, then the other
-	// will be used.  If neither is set, then the built-in default is used.
+	// stream (client->server) in bytes. The size which is measured is the serialized,
+	// uncompressed payload in bytes. The actual value used is the minumum of the value
+	// specified here and the value set by the application via the gRPC client API. If
+	// either one is not set, then the other will be used.  If neither is set, then the
+	// built-in default is used.
 	// TODO: support this.
-	MaxReqSize uint32
+	MaxReqSize uint64
 	// MaxRespSize is the maximum allowed payload size for an individual response in a
 	// stream (server->client) in bytes.
 	// TODO: support this.
-	MaxRespSize uint32
+	MaxRespSize uint64
 }
 
 // ServiceConfig is provided by the service provider and contains parameters for how
@@ -498,6 +517,3 @@ type ServiceConfig struct {
 // requires a synchronised update of grpc-go and protoc-gen-go. This constant
 // should not be referenced from any other code.
 const SupportPackageIsVersion4 = true
-
-// Version is the current grpc version.
-const Version = "1.3.0-dev"

go/vendor/google.golang.org/grpc/server.go

@@ -53,10 +53,8 @@ import (
 	"google.golang.org/grpc/credentials"
 	"google.golang.org/grpc/grpclog"
 	"google.golang.org/grpc/internal"
-	"google.golang.org/grpc/keepalive"
 	"google.golang.org/grpc/metadata"
 	"google.golang.org/grpc/stats"
-	"google.golang.org/grpc/status"
 	"google.golang.org/grpc/tap"
 	"google.golang.org/grpc/transport"
 )
@@ -118,9 +116,6 @@ type options struct {
 	statsHandler         stats.Handler
 	maxConcurrentStreams uint32
 	useHandlerImpl       bool // use http.Handler-based server
-	unknownStreamDesc    *StreamDesc
-	keepaliveParams      keepalive.ServerParameters
-	keepalivePolicy      keepalive.EnforcementPolicy
 }
 
 var defaultMaxMsgSize = 1024 * 1024 * 4 // use 4MB as the default message size limit
@@ -128,20 +123,6 @@ var defaultMaxMsgSize = 1024 * 1024 * 4 // use 4MB as the default message size l
 // A ServerOption sets options.
 type ServerOption func(*options)
 
-// KeepaliveParams returns a ServerOption that sets keepalive and max-age parameters for the server.
-func KeepaliveParams(kp keepalive.ServerParameters) ServerOption {
-	return func(o *options) {
-		o.keepaliveParams = kp
-	}
-}
-
-// KeepaliveEnforcementPolicy returns a ServerOption that sets keepalive enforcement policy for the server.
-func KeepaliveEnforcementPolicy(kep keepalive.EnforcementPolicy) ServerOption {
-	return func(o *options) {
-		o.keepalivePolicy = kep
-	}
-}
-
 // CustomCodec returns a ServerOption that sets a codec for message marshaling and unmarshaling.
 func CustomCodec(codec Codec) ServerOption {
 	return func(o *options) {
@@ -227,24 +208,6 @@ func StatsHandler(h stats.Handler) ServerOption {
 	}
 }
 
-// UnknownServiceHandler returns a ServerOption that allows for adding a custom
-// unknown service handler. The provided method is a bidi-streaming RPC service
-// handler that will be invoked instead of returning the the "unimplemented" gRPC
-// error whenever a request is received for an unregistered service or method.
-// The handling function has full access to the Context of the request and the
-// stream, and the invocation passes through interceptors.
-func UnknownServiceHandler(streamHandler StreamHandler) ServerOption {
-	return func(o *options) {
-		o.unknownStreamDesc = &StreamDesc{
-			StreamName: "unknown_service_handler",
-			Handler:    streamHandler,
-			// We need to assume that the users of the streamHandler will want to use both.
-			ClientStreams: true,
-			ServerStreams: true,
-		}
-	}
-}
-
 // NewServer creates a gRPC server which has no service registered and has not
 // started to accept requests yet.
 func NewServer(opt ...ServerOption) *Server {
@@ -487,8 +450,6 @@ func (s *Server) serveHTTP2Transport(c net.Conn, authInfo credentials.AuthInfo)
 		AuthInfo:     authInfo,
 		InTapHandle:  s.opts.inTapHandle,
 		StatsHandler: s.opts.statsHandler,
-		KeepaliveParams: s.opts.keepaliveParams,
-		KeepalivePolicy: s.opts.keepalivePolicy,
 	}
 	st, err := transport.NewServerTransport("http2", c, config)
 	if err != nil {
@@ -672,7 +633,7 @@ func (s *Server) processUnaryRPC(t transport.ServerTransport, stream *transport.
 		stream.SetSendCompress(s.opts.cp.Type())
 	}
 	p := &parser{r: stream}
-	for { // TODO: delete
+	for {
 		pf, req, err := p.recvMsg(s.opts.maxMsgSize)
 		if err == io.EOF {
 			// The entire stream is done (for unary RPC only).
@@ -682,44 +643,45 @@ func (s *Server) processUnaryRPC(t transport.ServerTransport, stream *transport.
 			err = Errorf(codes.Internal, io.ErrUnexpectedEOF.Error())
 		}
 		if err != nil {
-			if st, ok := status.FromError(err); ok {
-				if e := t.WriteStatus(stream, st); e != nil {
+			switch err := err.(type) {
+			case *rpcError:
+				if e := t.WriteStatus(stream, err.code, err.desc); e != nil {
 					grpclog.Printf("grpc: Server.processUnaryRPC failed to write status %v", e)
 				}
-			} else {
-				switch st := err.(type) {
 			case transport.ConnectionError:
 				// Nothing to do here.
 			case transport.StreamError:
-					if e := t.WriteStatus(stream, status.New(st.Code, st.Desc)); e != nil {
+				if e := t.WriteStatus(stream, err.Code, err.Desc); e != nil {
 					grpclog.Printf("grpc: Server.processUnaryRPC failed to write status %v", e)
 				}
 			default:
-					panic(fmt.Sprintf("grpc: Unexpected error (%T) from recvMsg: %v", st, st))
-				}
+				panic(fmt.Sprintf("grpc: Unexpected error (%T) from recvMsg: %v", err, err))
 			}
 			return err
 		}
 
 		if err := checkRecvPayload(pf, stream.RecvCompress(), s.opts.dc); err != nil {
-			if st, ok := status.FromError(err); ok {
-				if e := t.WriteStatus(stream, st); e != nil {
+			switch err := err.(type) {
+			case *rpcError:
+				if e := t.WriteStatus(stream, err.code, err.desc); e != nil {
 					grpclog.Printf("grpc: Server.processUnaryRPC failed to write status %v", e)
 				}
 				return err
-			}
-			if e := t.WriteStatus(stream, status.New(codes.Internal, err.Error())); e != nil {
+			default:
+				if e := t.WriteStatus(stream, codes.Internal, err.Error()); e != nil {
 					grpclog.Printf("grpc: Server.processUnaryRPC failed to write status %v", e)
 				}
-
 				// TODO checkRecvPayload always return RPC error. Add a return here if necessary.
 			}
+		}
 		var inPayload *stats.InPayload
 		if sh != nil {
 			inPayload = &stats.InPayload{
 				RecvTime: time.Now(),
 			}
 		}
+		statusCode := codes.OK
+		statusDesc := ""
 		df := func(v interface{}) error {
 			if inPayload != nil {
 				inPayload.WireLength = len(req)
@@ -728,16 +690,20 @@ func (s *Server) processUnaryRPC(t transport.ServerTransport, stream *transport.
 				var err error
 				req, err = s.opts.dc.Do(bytes.NewReader(req))
 				if err != nil {
+					if err := t.WriteStatus(stream, codes.Internal, err.Error()); err != nil {
+						grpclog.Printf("grpc: Server.processUnaryRPC failed to write status %v", err)
+					}
 					return Errorf(codes.Internal, err.Error())
 				}
 			}
 			if len(req) > s.opts.maxMsgSize {
 				// TODO: Revisit the error code. Currently keep it consistent with
 				// java implementation.
-				return status.Errorf(codes.Internal, "grpc: server received a message of %d bytes exceeding %d limit", len(req), s.opts.maxMsgSize)
+				statusCode = codes.Internal
+				statusDesc = fmt.Sprintf("grpc: server received a message of %d bytes exceeding %d limit", len(req), s.opts.maxMsgSize)
 			}
 			if err := s.opts.codec.Unmarshal(req, v); err != nil {
-				return status.Errorf(codes.Internal, "grpc: error unmarshalling request: %v", err)
+				return err
 			}
 			if inPayload != nil {
 				inPayload.Payload = v
@@ -752,20 +718,21 @@ func (s *Server) processUnaryRPC(t transport.ServerTransport, stream *transport.
 		}
 		reply, appErr := md.Handler(srv.server, stream.Context(), df, s.opts.unaryInt)
 		if appErr != nil {
-			appStatus, ok := status.FromError(appErr)
-			if !ok {
-				// Convert appErr if it is not a grpc status error.
-				appErr = status.Error(convertCode(appErr), appErr.Error())
-				appStatus, _ = status.FromError(appErr)
+			if err, ok := appErr.(*rpcError); ok {
+				statusCode = err.code
+				statusDesc = err.desc
+			} else {
+				statusCode = convertCode(appErr)
+				statusDesc = appErr.Error()
 			}
-			if trInfo != nil {
-				trInfo.tr.LazyLog(stringer(appStatus.Message()), true)
+			if trInfo != nil && statusCode != codes.OK {
+				trInfo.tr.LazyLog(stringer(statusDesc), true)
 				trInfo.tr.SetError()
 			}
-			if e := t.WriteStatus(stream, appStatus); e != nil {
-				grpclog.Printf("grpc: Server.processUnaryRPC failed to write status: %v", e)
+			if err := t.WriteStatus(stream, statusCode, statusDesc); err != nil {
+				grpclog.Printf("grpc: Server.processUnaryRPC failed to write status: %v", err)
 			}
-			return appErr
+			return Errorf(statusCode, statusDesc)
 		}
 		if trInfo != nil {
 			trInfo.tr.LazyLog(stringer("OK"), false)
@@ -775,35 +742,26 @@ func (s *Server) processUnaryRPC(t transport.ServerTransport, stream *transport.
 			Delay: false,
 		}
 		if err := s.sendResponse(t, stream, reply, s.opts.cp, opts); err != nil {
-			if err == io.EOF {
-				// The entire stream is done (for unary RPC only).
-				return err
-			}
-			if s, ok := status.FromError(err); ok {
-				if e := t.WriteStatus(stream, s); e != nil {
-					grpclog.Printf("grpc: Server.processUnaryRPC failed to write status: %v", e)
-				}
-			} else {
-				switch st := err.(type) {
+			switch err := err.(type) {
 			case transport.ConnectionError:
 				// Nothing to do here.
 			case transport.StreamError:
-					if e := t.WriteStatus(stream, status.New(st.Code, st.Desc)); e != nil {
-						grpclog.Printf("grpc: Server.processUnaryRPC failed to write status %v", e)
-					}
+				statusCode = err.Code
+				statusDesc = err.Desc
 			default:
-					panic(fmt.Sprintf("grpc: Unexpected error (%T) from sendResponse: %v", st, st))
-				}
+				statusCode = codes.Unknown
+				statusDesc = err.Error()
 			}
 			return err
 		}
 		if trInfo != nil {
 			trInfo.tr.LazyLog(&payload{sent: true, msg: reply}, true)
 		}
-		// TODO: Should we be logging if writing status failed here, like above?
-		// Should the logging be in WriteStatus?  Should we ignore the WriteStatus
-		// error or allow the stats handler to see it?
-		return t.WriteStatus(stream, status.New(codes.OK, ""))
+		errWrite := t.WriteStatus(stream, statusCode, statusDesc)
+		if statusCode != codes.OK {
+			return Errorf(statusCode, statusDesc)
+		}
+		return errWrite
 	}
 }
 
@@ -857,47 +815,43 @@ func (s *Server) processStreamingRPC(t transport.ServerTransport, stream *transp
 		}()
 	}
 	var appErr error
-	var server interface{}
-	if srv != nil {
-		server = srv.server
-	}
 	if s.opts.streamInt == nil {
-		appErr = sd.Handler(server, ss)
+		appErr = sd.Handler(srv.server, ss)
 	} else {
 		info := &StreamServerInfo{
 			FullMethod:     stream.Method(),
 			IsClientStream: sd.ClientStreams,
 			IsServerStream: sd.ServerStreams,
 		}
-		appErr = s.opts.streamInt(server, ss, info, sd.Handler)
+		appErr = s.opts.streamInt(srv.server, ss, info, sd.Handler)
 	}
 	if appErr != nil {
-		appStatus, ok := status.FromError(appErr)
-		if !ok {
-			switch err := appErr.(type) {
-			case transport.StreamError:
-				appStatus = status.New(err.Code, err.Desc)
-			default:
-				appStatus = status.New(convertCode(appErr), appErr.Error())
+		if err, ok := appErr.(*rpcError); ok {
+			ss.statusCode = err.code
+			ss.statusDesc = err.desc
+		} else if err, ok := appErr.(transport.StreamError); ok {
+			ss.statusCode = err.Code
+			ss.statusDesc = err.Desc
+		} else {
+			ss.statusCode = convertCode(appErr)
+			ss.statusDesc = appErr.Error()
 		}
-			appErr = appStatus.Err()
 	}
 	if trInfo != nil {
 		ss.mu.Lock()
-			ss.trInfo.tr.LazyLog(stringer(appStatus.Message()), true)
+		if ss.statusCode != codes.OK {
+			ss.trInfo.tr.LazyLog(stringer(ss.statusDesc), true)
 			ss.trInfo.tr.SetError()
-			ss.mu.Unlock()
-		}
-		t.WriteStatus(ss.s, appStatus)
-		// TODO: Should we log an error from WriteStatus here and below?
-		return appErr
-	}
-	if trInfo != nil {
-		ss.mu.Lock()
+		} else {
 			ss.trInfo.tr.LazyLog(stringer("OK"), false)
+		}
 		ss.mu.Unlock()
 	}
-	return t.WriteStatus(ss.s, status.New(codes.OK, ""))
+	errWrite := t.WriteStatus(ss.s, ss.statusCode, ss.statusDesc)
+	if ss.statusCode != codes.OK {
+		return Errorf(ss.statusCode, ss.statusDesc)
+	}
+	return errWrite
 
 }
 
@@ -913,7 +867,7 @@ func (s *Server) handleStream(t transport.ServerTransport, stream *transport.Str
 			trInfo.tr.SetError()
 		}
 		errDesc := fmt.Sprintf("malformed method name: %q", stream.Method())
-		if err := t.WriteStatus(stream, status.New(codes.InvalidArgument, errDesc)); err != nil {
+		if err := t.WriteStatus(stream, codes.InvalidArgument, errDesc); err != nil {
 			if trInfo != nil {
 				trInfo.tr.LazyLog(&fmtStringer{"%v", []interface{}{err}}, true)
 				trInfo.tr.SetError()
@@ -929,16 +883,12 @@ func (s *Server) handleStream(t transport.ServerTransport, stream *transport.Str
 	method := sm[pos+1:]
 	srv, ok := s.m[service]
 	if !ok {
-		if unknownDesc := s.opts.unknownStreamDesc; unknownDesc != nil {
-			s.processStreamingRPC(t, stream, nil, unknownDesc, trInfo)
-			return
-		}
 		if trInfo != nil {
 			trInfo.tr.LazyLog(&fmtStringer{"Unknown service %v", []interface{}{service}}, true)
 			trInfo.tr.SetError()
 		}
 		errDesc := fmt.Sprintf("unknown service %v", service)
-		if err := t.WriteStatus(stream, status.New(codes.Unimplemented, errDesc)); err != nil {
+		if err := t.WriteStatus(stream, codes.Unimplemented, errDesc); err != nil {
 			if trInfo != nil {
 				trInfo.tr.LazyLog(&fmtStringer{"%v", []interface{}{err}}, true)
 				trInfo.tr.SetError()
@@ -963,12 +913,8 @@ func (s *Server) handleStream(t transport.ServerTransport, stream *transport.Str
 		trInfo.tr.LazyLog(&fmtStringer{"Unknown method %v", []interface{}{method}}, true)
 		trInfo.tr.SetError()
 	}
-	if unknownDesc := s.opts.unknownStreamDesc; unknownDesc != nil {
-		s.processStreamingRPC(t, stream, nil, unknownDesc, trInfo)
-		return
-	}
 	errDesc := fmt.Sprintf("unknown method %v", method)
-	if err := t.WriteStatus(stream, status.New(codes.Unimplemented, errDesc)); err != nil {
+	if err := t.WriteStatus(stream, codes.Unimplemented, errDesc); err != nil {
 		if trInfo != nil {
 			trInfo.tr.LazyLog(&fmtStringer{"%v", []interface{}{err}}, true)
 			trInfo.tr.SetError()

go/vendor/google.golang.org/grpc/stream.go

@@ -37,6 +37,7 @@ import (
 	"bytes"
 	"errors"
 	"io"
+	"math"
 	"sync"
 	"time"
 
@@ -45,7 +46,6 @@ import (
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/metadata"
 	"google.golang.org/grpc/stats"
-	"google.golang.org/grpc/status"
 	"google.golang.org/grpc/transport"
 )
 
@@ -178,7 +178,7 @@ func newClientStream(ctx context.Context, desc *StreamDesc, cc *ClientConn, meth
 		t, put, err = cc.getTransport(ctx, gopts)
 		if err != nil {
 			// TODO(zhaoq): Probably revisit the error handling.
-			if _, ok := status.FromError(err); ok {
+			if _, ok := err.(*rpcError); ok {
 				return nil, err
 			}
 			if err == errConnClosing || err == errConnUnavailable {
@@ -214,7 +214,6 @@ func newClientStream(ctx context.Context, desc *StreamDesc, cc *ClientConn, meth
 		codec:  cc.dopts.codec,
 		cp:     cc.dopts.cp,
 		dc:     cc.dopts.dc,
-		maxMsgSize: cc.dopts.maxMsgSize,
 		cancel: cancel,
 
 		put: put,
@@ -240,7 +239,11 @@ func newClientStream(ctx context.Context, desc *StreamDesc, cc *ClientConn, meth
 		case <-s.Done():
 			// TODO: The trace of the RPC is terminated here when there is no pending
 			// I/O, which is probably not the optimal solution.
-			cs.finish(s.Status().Err())
+			if s.StatusCode() == codes.OK {
+				cs.finish(nil)
+			} else {
+				cs.finish(Errorf(s.StatusCode(), "%s", s.StatusDesc()))
+			}
 			cs.closeTransportStream(nil)
 		case <-s.GoAway():
 			cs.finish(errConnDrain)
@@ -266,7 +269,6 @@ type clientStream struct {
 	cp     Compressor
 	cbuf   *bytes.Buffer
 	dc     Decompressor
-	maxMsgSize int
 	cancel context.CancelFunc
 
 	tracing bool // set to EnableTracing when the clientStream is created.
@@ -380,7 +382,7 @@ func (cs *clientStream) RecvMsg(m interface{}) (err error) {
 			Client: true,
 		}
 	}
-	err = recv(cs.p, cs.codec, cs.s, cs.dc, m, cs.maxMsgSize, inPayload)
+	err = recv(cs.p, cs.codec, cs.s, cs.dc, m, math.MaxInt32, inPayload)
 	defer func() {
 		// err != nil indicates the termination of the stream.
 		if err != nil {
@@ -403,30 +405,30 @@ func (cs *clientStream) RecvMsg(m interface{}) (err error) {
 		}
 		// Special handling for client streaming rpc.
 		// This recv expects EOF or errors, so we don't collect inPayload.
-		err = recv(cs.p, cs.codec, cs.s, cs.dc, m, cs.maxMsgSize, nil)
+		err = recv(cs.p, cs.codec, cs.s, cs.dc, m, math.MaxInt32, nil)
 		cs.closeTransportStream(err)
 		if err == nil {
 			return toRPCErr(errors.New("grpc: client streaming protocol violation: get <nil>, want <EOF>"))
 		}
 		if err == io.EOF {
-			if se := cs.s.Status().Err(); se != nil {
-				return se
-			}
+			if cs.s.StatusCode() == codes.OK {
 				cs.finish(err)
 				return nil
 			}
+			return Errorf(cs.s.StatusCode(), "%s", cs.s.StatusDesc())
+		}
 		return toRPCErr(err)
 	}
 	if _, ok := err.(transport.ConnectionError); !ok {
 		cs.closeTransportStream(err)
 	}
 	if err == io.EOF {
-		if statusErr := cs.s.Status().Err(); statusErr != nil {
-			return statusErr
-		}
+		if cs.s.StatusCode() == codes.OK {
 			// Returns io.EOF to indicate the end of the stream.
 			return
 		}
+		return Errorf(cs.s.StatusCode(), "%s", cs.s.StatusDesc())
+	}
 	return toRPCErr(err)
 }
 
@@ -517,6 +519,8 @@ type serverStream struct {
 	dc         Decompressor
 	cbuf       *bytes.Buffer
 	maxMsgSize int
+	statusCode codes.Code
+	statusDesc string
 	trInfo     *traceInfo
 
 	statsHandler stats.Handler

go/vendor/google.golang.org/grpc/transport/control.go

@@ -35,9 +35,7 @@ package transport
 
 import (
 	"fmt"
-	"math"
 	"sync"
-	"time"
 
 	"golang.org/x/net/http2"
 )
@@ -48,16 +46,6 @@ const (
 	// The initial window size for flow control.
 	initialWindowSize     = defaultWindowSize      // for an RPC
 	initialConnWindowSize = defaultWindowSize * 16 // for a connection
-	infinity                      = time.Duration(math.MaxInt64)
-	defaultClientKeepaliveTime    = infinity
-	defaultClientKeepaliveTimeout = time.Duration(20 * time.Second)
-	defaultMaxStreamsClient       = 100
-	defaultMaxConnectionIdle      = infinity
-	defaultMaxConnectionAge       = infinity
-	defaultMaxConnectionAgeGrace  = infinity
-	defaultServerKeepaliveTime    = time.Duration(2 * time.Hour)
-	defaultServerKeepaliveTimeout = time.Duration(20 * time.Second)
-	defaultKeepalivePolicyMinTime = time.Duration(5 * time.Minute)
 )
 
 // The following defines various control items which could flow through
@@ -85,8 +73,6 @@ type resetStream struct {
 func (*resetStream) item() {}
 
 type goAway struct {
-	code      http2.ErrCode
-	debugData []byte
 }
 
 func (*goAway) item() {}

go/vendor/google.golang.org/grpc/transport/handler_server.go

@@ -53,7 +53,6 @@ import (
 	"google.golang.org/grpc/credentials"
 	"google.golang.org/grpc/metadata"
 	"google.golang.org/grpc/peer"
-	"google.golang.org/grpc/status"
 )
 
 // NewServerHandlerTransport returns a ServerTransport handling gRPC
@@ -183,7 +182,7 @@ func (ht *serverHandlerTransport) do(fn func()) error {
 	}
 }
 
-func (ht *serverHandlerTransport) WriteStatus(s *Stream, st *status.Status) error {
+func (ht *serverHandlerTransport) WriteStatus(s *Stream, statusCode codes.Code, statusDesc string) error {
 	err := ht.do(func() {
 		ht.writeCommonHeaders(s)
 
@@ -193,13 +192,10 @@ func (ht *serverHandlerTransport) WriteStatus(s *Stream, st *status.Status) erro
 		ht.rw.(http.Flusher).Flush()
 
 		h := ht.rw.Header()
-		h.Set("Grpc-Status", fmt.Sprintf("%d", st.Code()))
-		if m := st.Message(); m != "" {
-			h.Set("Grpc-Message", encodeGrpcMessage(m))
+		h.Set("Grpc-Status", fmt.Sprintf("%d", statusCode))
+		if statusDesc != "" {
+			h.Set("Grpc-Message", encodeGrpcMessage(statusDesc))
 		}
-
-		// TODO: Support Grpc-Status-Details-Bin
-
 		if md := s.Trailer(); len(md) > 0 {
 			for k, vv := range md {
 				// Clients don't tolerate reading restricted headers after some non restricted ones were sent.
@@ -238,7 +234,6 @@ func (ht *serverHandlerTransport) writeCommonHeaders(s *Stream) {
 	// and https://golang.org/pkg/net/http/#example_ResponseWriter_trailers
 	h.Add("Trailer", "Grpc-Status")
 	h.Add("Trailer", "Grpc-Message")
-	// TODO: Support Grpc-Status-Details-Bin
 
 	if s.sendCompress != "" {
 		h.Set("Grpc-Encoding", s.sendCompress)
@@ -319,7 +314,7 @@ func (ht *serverHandlerTransport) HandleStreams(startStream func(*Stream), trace
 	if req.TLS != nil {
 		pr.AuthInfo = credentials.TLSInfo{State: *req.TLS}
 	}
-	ctx = metadata.NewIncomingContext(ctx, ht.headerMD)
+	ctx = metadata.NewContext(ctx, ht.headerMD)
 	ctx = peer.NewContext(ctx, pr)
 	s.ctx = newContextWithStream(ctx, s)
 	s.dec = &recvBufferReader{ctx: s.ctx, recv: s.buf}

go/vendor/google.golang.org/grpc/transport/http2_client.go

@@ -35,12 +35,12 @@ package transport
 
 import (
 	"bytes"
+	"fmt"
 	"io"
 	"math"
 	"net"
 	"strings"
 	"sync"
-	"sync/atomic"
 	"time"
 
 	"golang.org/x/net/context"
@@ -49,11 +49,9 @@ import (
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/credentials"
 	"google.golang.org/grpc/grpclog"
-	"google.golang.org/grpc/keepalive"
 	"google.golang.org/grpc/metadata"
 	"google.golang.org/grpc/peer"
 	"google.golang.org/grpc/stats"
-	"google.golang.org/grpc/status"
 )
 
 // http2Client implements the ClientTransport interface with HTTP2.
@@ -82,8 +80,6 @@ type http2Client struct {
 	// goAway is closed to notify the upper layer (i.e., addrConn.transportMonitor)
 	// that the server sent GoAway on this transport.
 	goAway chan struct{}
-	// awakenKeepalive is used to wake up keepalive when after it has gone dormant.
-	awakenKeepalive chan struct{}
 
 	framer *framer
 	hBuf   *bytes.Buffer  // the buffer for HPACK encoding
@@ -103,11 +99,6 @@ type http2Client struct {
 
 	creds []credentials.PerRPCCredentials
 
-	// Boolean to keep track of reading activity on transport.
-	// 1 is true and 0 is false.
-	activity uint32 // Accessed atomically.
-	kp       keepalive.ClientParameters
-
 	statsHandler stats.Handler
 
 	mu            sync.Mutex     // guard the following variables
@@ -121,9 +112,6 @@ type http2Client struct {
 	goAwayID uint32
 	// prevGoAway ID records the Last-Stream-ID in the previous GOAway frame.
 	prevGoAwayID uint32
-	// goAwayReason records the http2.ErrCode and debug data received with the
-	// GoAway frame.
-	goAwayReason GoAwayReason
 }
 
 func dial(ctx context.Context, fn func(context.Context, string) (net.Conn, error), addr string) (net.Conn, error) {
@@ -190,19 +178,15 @@ func newHTTP2Client(ctx context.Context, addr TargetInfo, opts ConnectOptions) (
 			return nil, connectionErrorf(temp, err, "transport: %v", err)
 		}
 	}
-	kp := opts.KeepaliveParams
-	// Validate keepalive parameters.
-	if kp.Time == 0 {
-		kp.Time = defaultClientKeepaliveTime
-	}
-	if kp.Timeout == 0 {
-		kp.Timeout = defaultClientKeepaliveTimeout
+	ua := primaryUA
+	if opts.UserAgent != "" {
+		ua = opts.UserAgent + " " + ua
 	}
 	var buf bytes.Buffer
 	t := &http2Client{
 		ctx:        ctx,
 		target:     addr.Addr,
-		userAgent:  opts.UserAgent,
+		userAgent:  ua,
 		md:         addr.Metadata,
 		conn:       conn,
 		remoteAddr: conn.RemoteAddr(),
@@ -214,7 +198,6 @@ func newHTTP2Client(ctx context.Context, addr TargetInfo, opts ConnectOptions) (
 		shutdownChan:    make(chan struct{}),
 		errorChan:       make(chan struct{}),
 		goAway:          make(chan struct{}),
-		awakenKeepalive: make(chan struct{}, 1),
 		framer:          newFramer(conn),
 		hBuf:            &buf,
 		hEnc:            hpack.NewEncoder(&buf),
@@ -225,15 +208,10 @@ func newHTTP2Client(ctx context.Context, addr TargetInfo, opts ConnectOptions) (
 		state:           reachable,
 		activeStreams:   make(map[uint32]*Stream),
 		creds:           opts.PerRPCCredentials,
-		maxStreams:      defaultMaxStreamsClient,
-		streamsQuota:    newQuotaPool(defaultMaxStreamsClient),
+		maxStreams:      math.MaxInt32,
 		streamSendQuota: defaultWindowSize,
-		kp:              kp,
 		statsHandler:    opts.StatsHandler,
 	}
-	// Make sure awakenKeepalive can't be written upon.
-	// keepalive routine will make it writable, if need be.
-	t.awakenKeepalive <- struct{}{}
 	if t.statsHandler != nil {
 		t.ctx = t.statsHandler.TagConn(t.ctx, &stats.ConnTagInfo{
 			RemoteAddr: t.remoteAddr,
@@ -278,9 +256,6 @@ func newHTTP2Client(ctx context.Context, addr TargetInfo, opts ConnectOptions) (
 		}
 	}
 	go t.controller()
-	if t.kp.Time != infinity {
-		go t.keepalive()
-	}
 	t.writableChan <- 0
 	return t, nil
 }
@@ -314,7 +289,7 @@ func (t *http2Client) newStream(ctx context.Context, callHdr *CallHdr) *Stream {
 	return s
 }
 
-// NewStream creates a stream and registers it into the transport as "active"
+// NewStream creates a stream and register it into the transport as "active"
 // streams.
 func (t *http2Client) NewStream(ctx context.Context, callHdr *CallHdr) (_ *Stream, err error) {
 	pr := &peer.Peer{
@@ -362,7 +337,9 @@ func (t *http2Client) NewStream(ctx context.Context, callHdr *CallHdr) (_ *Strea
 		t.mu.Unlock()
 		return nil, ErrConnClosing
 	}
+	checkStreamsQuota := t.streamsQuota != nil
 	t.mu.Unlock()
+	if checkStreamsQuota {
 		sq, err := wait(ctx, nil, nil, t.shutdownChan, t.streamsQuota.acquire())
 		if err != nil {
 			return nil, err
@@ -371,9 +348,10 @@ func (t *http2Client) NewStream(ctx context.Context, callHdr *CallHdr) (_ *Strea
 		if sq > 1 {
 			t.streamsQuota.add(sq - 1)
 		}
+	}
 	if _, err := wait(ctx, nil, nil, t.shutdownChan, t.writableChan); err != nil {
 		// Return the quota back now because there is no stream returned to the caller.
-		if _, ok := err.(StreamError); ok {
+		if _, ok := err.(StreamError); ok && checkStreamsQuota {
 			t.streamsQuota.add(1)
 		}
 		return nil, err
@@ -381,7 +359,9 @@ func (t *http2Client) NewStream(ctx context.Context, callHdr *CallHdr) (_ *Strea
 	t.mu.Lock()
 	if t.state == draining {
 		t.mu.Unlock()
+		if checkStreamsQuota {
 			t.streamsQuota.add(1)
+		}
 		// Need to make t writable again so that the rpc in flight can still proceed.
 		t.writableChan <- 0
 		return nil, ErrStreamDrain
@@ -393,17 +373,17 @@ func (t *http2Client) NewStream(ctx context.Context, callHdr *CallHdr) (_ *Strea
 	s := t.newStream(ctx, callHdr)
 	s.clientStatsCtx = userCtx
 	t.activeStreams[s.id] = s
-	// If the number of active streams change from 0 to 1, then check if keepalive
-	// has gone dormant. If so, wake it up.
-	if len(t.activeStreams) == 1 {
-		select {
-		case t.awakenKeepalive <- struct{}{}:
-			t.framer.writePing(false, false, [8]byte{})
-		default:
-		}
-	}
 
+	// This stream is not counted when applySetings(...) initialize t.streamsQuota.
+	// Reset t.streamsQuota to the right value.
+	var reset bool
+	if !checkStreamsQuota && t.streamsQuota != nil {
+		reset = true
+	}
 	t.mu.Unlock()
+	if reset {
+		t.streamsQuota.add(-1)
+	}
 
 	// HPACK encodes various headers. Note that once WriteField(...) is
 	// called, the corresponding headers/continuation frame has to be sent
@@ -435,7 +415,7 @@ func (t *http2Client) NewStream(ctx context.Context, callHdr *CallHdr) (_ *Strea
 		hasMD      bool
 		endHeaders bool
 	)
-	if md, ok := metadata.FromOutgoingContext(ctx); ok {
+	if md, ok := metadata.FromContext(ctx); ok {
 		hasMD = true
 		for k, v := range md {
 			// HTTP doesn't allow you to set pseudoheaders after non pseudoheaders were set.
@@ -511,11 +491,15 @@ func (t *http2Client) NewStream(ctx context.Context, callHdr *CallHdr) (_ *Strea
 // CloseStream clears the footprint of a stream when the stream is not needed any more.
 // This must not be executed in reader's goroutine.
 func (t *http2Client) CloseStream(s *Stream, err error) {
+	var updateStreams bool
 	t.mu.Lock()
 	if t.activeStreams == nil {
 		t.mu.Unlock()
 		return
 	}
+	if t.streamsQuota != nil {
+		updateStreams = true
+	}
 	delete(t.activeStreams, s.id)
 	if t.state == draining && len(t.activeStreams) == 0 {
 		// The transport is draining and s is the last live stream on t.
@@ -524,27 +508,10 @@ func (t *http2Client) CloseStream(s *Stream, err error) {
 		return
 	}
 	t.mu.Unlock()
-	// rstStream is true in case the stream is being closed at the client-side
-	// and the server needs to be intimated about it by sending a RST_STREAM
-	// frame.
-	// To make sure this frame is written to the wire before the headers of the
-	// next stream waiting for streamsQuota, we add to streamsQuota pool only
-	// after having acquired the writableChan to send RST_STREAM out (look at
-	// the controller() routine).
-	var rstStream bool
-	var rstError http2.ErrCode
-	defer func() {
-		// In case, the client doesn't have to send RST_STREAM to server
-		// we can safely add back to streamsQuota pool now.
-		if !rstStream {
+	if updateStreams {
 		t.streamsQuota.add(1)
-			return
 	}
-		t.controlBuf.put(&resetStream{s.id, rstError})
-	}()
 	s.mu.Lock()
-	rstStream = s.rstStream
-	rstError = s.rstError
 	if q := s.fc.resetPendingData(); q > 0 {
 		if n := t.fc.onRead(q); n > 0 {
 			t.controlBuf.put(&windowUpdate{0, n})
@@ -560,9 +527,8 @@ func (t *http2Client) CloseStream(s *Stream, err error) {
 	}
 	s.state = streamDone
 	s.mu.Unlock()
-	if _, ok := err.(StreamError); ok {
-		rstStream = true
-		rstError = http2.ErrCodeCancel
+	if se, ok := err.(StreamError); ok && se.Code != codes.DeadlineExceeded {
+		t.controlBuf.put(&resetStream{s.id, http2.ErrCodeCancel})
 	}
 }
 
@@ -776,7 +742,7 @@ func (t *http2Client) updateWindow(s *Stream, n uint32) {
 }
 
 func (t *http2Client) handleData(f *http2.DataFrame) {
-	size := f.Header().Length
+	size := len(f.Data())
 	if err := t.fc.onData(uint32(size)); err != nil {
 		t.notifyError(connectionErrorf(true, err, "%v", err))
 		return
@@ -790,11 +756,6 @@ func (t *http2Client) handleData(f *http2.DataFrame) {
 		return
 	}
 	if size > 0 {
-		if f.Header().Flags.Has(http2.FlagDataPadded) {
-			if w := t.fc.onRead(uint32(size) - uint32(len(f.Data()))); w > 0 {
-				t.controlBuf.put(&windowUpdate{0, w})
-			}
-		}
 		s.mu.Lock()
 		if s.state == streamDone {
 			s.mu.Unlock()
@@ -805,28 +766,23 @@ func (t *http2Client) handleData(f *http2.DataFrame) {
 			return
 		}
 		if err := s.fc.onData(uint32(size)); err != nil {
-			s.rstStream = true
-			s.rstError = http2.ErrCodeFlowControl
-			s.finish(status.New(codes.Internal, err.Error()))
+			s.state = streamDone
+			s.statusCode = codes.Internal
+			s.statusDesc = err.Error()
+			close(s.done)
 			s.mu.Unlock()
 			s.write(recvMsg{err: io.EOF})
+			t.controlBuf.put(&resetStream{s.id, http2.ErrCodeFlowControl})
 			return
 		}
-		if f.Header().Flags.Has(http2.FlagDataPadded) {
-			if w := s.fc.onRead(uint32(size) - uint32(len(f.Data()))); w > 0 {
-				t.controlBuf.put(&windowUpdate{s.id, w})
-			}
-		}
 		s.mu.Unlock()
 		// TODO(bradfitz, zhaoq): A copy is required here because there is no
 		// guarantee f.Data() is consumed before the arrival of next frame.
 		// Can this copy be eliminated?
-		if len(f.Data()) > 0 {
-			data := make([]byte, len(f.Data()))
+		data := make([]byte, size)
 		copy(data, f.Data())
 		s.write(recvMsg{data: data})
 	}
-	}
 	// The server has closed the stream without sending trailers.  Record that
 	// the read direction is closed, and set the status appropriately.
 	if f.FrameHeader.Flags.Has(http2.FlagDataEndStream) {
@@ -835,7 +791,10 @@ func (t *http2Client) handleData(f *http2.DataFrame) {
 			s.mu.Unlock()
 			return
 		}
-		s.finish(status.New(codes.Internal, "server closed the stream without sending trailers"))
+		s.state = streamDone
+		s.statusCode = codes.Internal
+		s.statusDesc = "server closed the stream without sending trailers"
+		close(s.done)
 		s.mu.Unlock()
 		s.write(recvMsg{err: io.EOF})
 	}
@@ -851,16 +810,18 @@ func (t *http2Client) handleRSTStream(f *http2.RSTStreamFrame) {
 		s.mu.Unlock()
 		return
 	}
+	s.state = streamDone
 	if !s.headerDone {
 		close(s.headerChan)
 		s.headerDone = true
 	}
-	statusCode, ok := http2ErrConvTab[http2.ErrCode(f.ErrCode)]
+	s.statusCode, ok = http2ErrConvTab[http2.ErrCode(f.ErrCode)]
 	if !ok {
 		grpclog.Println("transport: http2Client.handleRSTStream found no mapped gRPC status for the received http2 error ", f.ErrCode)
-		statusCode = codes.Unknown
+		s.statusCode = codes.Unknown
 	}
-	s.finish(status.Newf(statusCode, "stream terminated by RST_STREAM with error code: %d", f.ErrCode))
+	s.statusDesc = fmt.Sprintf("stream terminated by RST_STREAM with error code: %d", f.ErrCode)
+	close(s.done)
 	s.mu.Unlock()
 	s.write(recvMsg{err: io.EOF})
 }
@@ -888,9 +849,6 @@ func (t *http2Client) handlePing(f *http2.PingFrame) {
 }
 
 func (t *http2Client) handleGoAway(f *http2.GoAwayFrame) {
-	if f.ErrCode == http2.ErrCodeEnhanceYourCalm {
-		grpclog.Printf("Client received GoAway with http2.ErrCodeEnhanceYourCalm.")
-	}
 	t.mu.Lock()
 	if t.state == reachable || t.state == draining {
 		if f.LastStreamID > 0 && f.LastStreamID%2 != 1 {
@@ -912,7 +870,6 @@ func (t *http2Client) handleGoAway(f *http2.GoAwayFrame) {
 			t.mu.Unlock()
 			return
 		default:
-			t.setGoAwayReason(f)
 		}
 		t.goAwayID = f.LastStreamID
 		close(t.goAway)
@@ -920,26 +877,6 @@ func (t *http2Client) handleGoAway(f *http2.GoAwayFrame) {
 	t.mu.Unlock()
 }
 
-// setGoAwayReason sets the value of t.goAwayReason based
-// on the GoAway frame received.
-// It expects a lock on transport's mutext to be held by
-// the caller.
-func (t *http2Client) setGoAwayReason(f *http2.GoAwayFrame) {
-	t.goAwayReason = NoReason
-	switch f.ErrCode {
-	case http2.ErrCodeEnhanceYourCalm:
-		if string(f.DebugData()) == "too_many_pings" {
-			t.goAwayReason = TooManyPings
-		}
-	}
-}
-
-func (t *http2Client) GetGoAwayReason() GoAwayReason {
-	t.mu.Lock()
-	defer t.mu.Unlock()
-	return t.goAwayReason
-}
-
 func (t *http2Client) handleWindowUpdate(f *http2.WindowUpdateFrame) {
 	id := f.Header().StreamID
 	incr := f.Increment
@@ -960,18 +897,19 @@ func (t *http2Client) operateHeaders(frame *http2.MetaHeadersFrame) {
 	}
 	var state decodeState
 	for _, hf := range frame.Fields {
-		if err := state.processHeaderField(hf); err != nil {
+		state.processHeaderField(hf)
+	}
+	if state.err != nil {
 		s.mu.Lock()
 		if !s.headerDone {
 			close(s.headerChan)
 			s.headerDone = true
 		}
 		s.mu.Unlock()
-			s.write(recvMsg{err: err})
+		s.write(recvMsg{err: state.err})
 		// Something wrong. Stops reading even when there is remaining.
 		return
 	}
-	}
 
 	endStream := frame.StreamEnded()
 	var isHeader bool
@@ -1013,7 +951,10 @@ func (t *http2Client) operateHeaders(frame *http2.MetaHeadersFrame) {
 	if len(state.mdata) > 0 {
 		s.trailer = state.mdata
 	}
-	s.finish(state.status())
+	s.statusCode = state.statusCode
+	s.statusDesc = state.statusDesc
+	close(s.done)
+	s.state = streamDone
 	s.mu.Unlock()
 	s.write(recvMsg{err: io.EOF})
 }
@@ -1041,7 +982,6 @@ func (t *http2Client) reader() {
 		t.notifyError(err)
 		return
 	}
-	atomic.CompareAndSwapUint32(&t.activity, 0, 1)
 	sf, ok := frame.(*http2.SettingsFrame)
 	if !ok {
 		t.notifyError(err)
@@ -1052,7 +992,6 @@ func (t *http2Client) reader() {
 	// loop to keep reading incoming messages on this transport.
 	for {
 		frame, err := t.framer.readFrame()
-		atomic.CompareAndSwapUint32(&t.activity, 0, 1)
 		if err != nil {
 			// Abort an active stream if the http2.Framer returns a
 			// http2.StreamError. This can happen only if the server's response
@@ -1104,10 +1043,16 @@ func (t *http2Client) applySettings(ss []http2.Setting) {
 				s.Val = math.MaxInt32
 			}
 			t.mu.Lock()
+			reset := t.streamsQuota != nil
+			if !reset {
+				t.streamsQuota = newQuotaPool(int(s.Val) - len(t.activeStreams))
+			}
 			ms := t.maxStreams
 			t.maxStreams = int(s.Val)
 			t.mu.Unlock()
+			if reset {
 				t.streamsQuota.add(int(s.Val) - ms)
+			}
 		case http2.SettingInitialWindowSize:
 			t.mu.Lock()
 			for _, stream := range t.activeStreams {
@@ -1140,12 +1085,6 @@ func (t *http2Client) controller() {
 						t.framer.writeSettings(true, i.ss...)
 					}
 				case *resetStream:
-					// If the server needs to be to intimated about stream closing,
-					// then we need to make sure the RST_STREAM frame is written to
-					// the wire before the headers of the next stream waiting on
-					// streamQuota. We ensure this by adding to the streamsQuota pool
-					// only after having acquired the writableChan to send RST_STREAM.
-					t.streamsQuota.add(1)
 					t.framer.writeRSTStream(true, i.streamID, i.code)
 				case *flushIO:
 					t.framer.flushWrite()
@@ -1165,61 +1104,6 @@ func (t *http2Client) controller() {
 	}
 }
 
-// keepalive running in a separate goroutune makes sure the connection is alive by sending pings.
-func (t *http2Client) keepalive() {
-	p := &ping{data: [8]byte{}}
-	timer := time.NewTimer(t.kp.Time)
-	for {
-		select {
-		case <-timer.C:
-			if atomic.CompareAndSwapUint32(&t.activity, 1, 0) {
-				timer.Reset(t.kp.Time)
-				continue
-			}
-			// Check if keepalive should go dormant.
-			t.mu.Lock()
-			if len(t.activeStreams) < 1 && !t.kp.PermitWithoutStream {
-				// Make awakenKeepalive writable.
-				<-t.awakenKeepalive
-				t.mu.Unlock()
-				select {
-				case <-t.awakenKeepalive:
-					// If the control gets here a ping has been sent
-					// need to reset the timer with keepalive.Timeout.
-				case <-t.shutdownChan:
-					return
-				}
-			} else {
-				t.mu.Unlock()
-				// Send ping.
-				t.controlBuf.put(p)
-			}
-
-			// By the time control gets here a ping has been sent one way or the other.
-			timer.Reset(t.kp.Timeout)
-			select {
-			case <-timer.C:
-				if atomic.CompareAndSwapUint32(&t.activity, 1, 0) {
-					timer.Reset(t.kp.Time)
-					continue
-				}
-				t.Close()
-				return
-			case <-t.shutdownChan:
-				if !timer.Stop() {
-					<-timer.C
-				}
-				return
-			}
-		case <-t.shutdownChan:
-			if !timer.Stop() {
-				<-timer.C
-			}
-			return
-		}
-	}
-}
-
 func (t *http2Client) Error() <-chan struct{} {
 	return t.errorChan
 }

go/vendor/google.golang.org/grpc/transport/http2_server.go

@@ -38,25 +38,19 @@ import (
 	"errors"
 	"io"
 	"math"
-	"math/rand"
 	"net"
 	"strconv"
 	"sync"
-	"sync/atomic"
-	"time"
 
-	"github.com/golang/protobuf/proto"
 	"golang.org/x/net/context"
 	"golang.org/x/net/http2"
 	"golang.org/x/net/http2/hpack"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/credentials"
 	"google.golang.org/grpc/grpclog"
-	"google.golang.org/grpc/keepalive"
 	"google.golang.org/grpc/metadata"
 	"google.golang.org/grpc/peer"
 	"google.golang.org/grpc/stats"
-	"google.golang.org/grpc/status"
 	"google.golang.org/grpc/tap"
 )
 
@@ -96,33 +90,11 @@ type http2Server struct {
 
 	stats stats.Handler
 
-	// Flag to keep track of reading activity on transport.
-	// 1 is true and 0 is false.
-	activity uint32 // Accessed atomically.
-	// Keepalive and max-age parameters for the server.
-	kp keepalive.ServerParameters
-
-	// Keepalive enforcement policy.
-	kep keepalive.EnforcementPolicy
-	// The time instance last ping was received.
-	lastPingAt time.Time
-	// Number of times the client has violated keepalive ping policy so far.
-	pingStrikes uint8
-	// Flag to signify that number of ping strikes should be reset to 0.
-	// This is set whenever data or header frames are sent.
-	// 1 means yes.
-	resetPingStrikes uint32 // Accessed atomically.
-
 	mu            sync.Mutex // guard the following
 	state         transportState
 	activeStreams map[uint32]*Stream
 	// the per-stream outbound flow control window size set by the peer.
 	streamSendQuota uint32
-	// idle is the time instant when the connection went idle.
-	// This is either the begining of the connection or when the number of
-	// RPCs go down to 0.
-	// When the connection is busy, this value is set to 0.
-	idle time.Time
 }
 
 // newHTTP2Server constructs a ServerTransport based on HTTP2. ConnectionError is
@@ -156,28 +128,6 @@ func newHTTP2Server(conn net.Conn, config *ServerConfig) (_ ServerTransport, err
 			return nil, connectionErrorf(true, err, "transport: %v", err)
 		}
 	}
-	kp := config.KeepaliveParams
-	if kp.MaxConnectionIdle == 0 {
-		kp.MaxConnectionIdle = defaultMaxConnectionIdle
-	}
-	if kp.MaxConnectionAge == 0 {
-		kp.MaxConnectionAge = defaultMaxConnectionAge
-	}
-	// Add a jitter to MaxConnectionAge.
-	kp.MaxConnectionAge += getJitter(kp.MaxConnectionAge)
-	if kp.MaxConnectionAgeGrace == 0 {
-		kp.MaxConnectionAgeGrace = defaultMaxConnectionAgeGrace
-	}
-	if kp.Time == 0 {
-		kp.Time = defaultServerKeepaliveTime
-	}
-	if kp.Timeout == 0 {
-		kp.Timeout = defaultServerKeepaliveTimeout
-	}
-	kep := config.KeepalivePolicy
-	if kep.MinTime == 0 {
-		kep.MinTime = defaultKeepalivePolicyMinTime
-	}
 	var buf bytes.Buffer
 	t := &http2Server{
 		ctx:             context.Background(),
@@ -199,9 +149,6 @@ func newHTTP2Server(conn net.Conn, config *ServerConfig) (_ ServerTransport, err
 		activeStreams:   make(map[uint32]*Stream),
 		streamSendQuota: defaultWindowSize,
 		stats:           config.StatsHandler,
-		kp:              kp,
-		idle:            time.Now(),
-		kep:             kep,
 	}
 	if t.stats != nil {
 		t.ctx = t.stats.TagConn(t.ctx, &stats.ConnTagInfo{
@@ -212,7 +159,6 @@ func newHTTP2Server(conn net.Conn, config *ServerConfig) (_ ServerTransport, err
 		t.stats.HandleConn(t.ctx, connBegin)
 	}
 	go t.controller()
-	go t.keepalive()
 	t.writableChan <- 0
 	return t, nil
 }
@@ -229,13 +175,14 @@ func (t *http2Server) operateHeaders(frame *http2.MetaHeadersFrame, handle func(
 
 	var state decodeState
 	for _, hf := range frame.Fields {
-		if err := state.processHeaderField(hf); err != nil {
+		state.processHeaderField(hf)
+	}
+	if err := state.err; err != nil {
 		if se, ok := err.(StreamError); ok {
 			t.controlBuf.put(&resetStream{s.id, statusCodeConvTab[se.Code]})
 		}
 		return
 	}
-	}
 
 	if frame.StreamEnded() {
 		// s is just created by the caller. No lock needed.
@@ -261,7 +208,7 @@ func (t *http2Server) operateHeaders(frame *http2.MetaHeadersFrame, handle func(
 	s.ctx = newContextWithStream(s.ctx, s)
 	// Attach the received metadata to the context.
 	if len(state.mdata) > 0 {
-		s.ctx = metadata.NewIncomingContext(s.ctx, state.mdata)
+		s.ctx = metadata.NewContext(s.ctx, state.mdata)
 	}
 
 	s.dec = &recvBufferReader{
@@ -301,9 +248,6 @@ func (t *http2Server) operateHeaders(frame *http2.MetaHeadersFrame, handle func(
 	t.maxStreamID = s.id
 	s.sendQuotaPool = newQuotaPool(int(t.streamSendQuota))
 	t.activeStreams[s.id] = s
-	if len(t.activeStreams) == 1 {
-		t.idle = time.Time{}
-	}
 	t.mu.Unlock()
 	s.windowHandler = func(n int) {
 		t.updateWindow(s, uint32(n))
@@ -351,7 +295,6 @@ func (t *http2Server) HandleStreams(handle func(*Stream), traceCtx func(context.
 		t.Close()
 		return
 	}
-	atomic.StoreUint32(&t.activity, 1)
 	sf, ok := frame.(*http2.SettingsFrame)
 	if !ok {
 		grpclog.Printf("transport: http2Server.HandleStreams saw invalid preface type %T from client", frame)
@@ -362,7 +305,6 @@ func (t *http2Server) HandleStreams(handle func(*Stream), traceCtx func(context.
 
 	for {
 		frame, err := t.framer.readFrame()
-		atomic.StoreUint32(&t.activity, 1)
 		if err != nil {
 			if se, ok := err.(http2.StreamError); ok {
 				t.mu.Lock()
@@ -439,7 +381,7 @@ func (t *http2Server) updateWindow(s *Stream, n uint32) {
 }
 
 func (t *http2Server) handleData(f *http2.DataFrame) {
-	size := f.Header().Length
+	size := len(f.Data())
 	if err := t.fc.onData(uint32(size)); err != nil {
 		grpclog.Printf("transport: http2Server %v", err)
 		t.Close()
@@ -454,11 +396,6 @@ func (t *http2Server) handleData(f *http2.DataFrame) {
 		return
 	}
 	if size > 0 {
-		if f.Header().Flags.Has(http2.FlagDataPadded) {
-			if w := t.fc.onRead(uint32(size) - uint32(len(f.Data()))); w > 0 {
-				t.controlBuf.put(&windowUpdate{0, w})
-			}
-		}
 		s.mu.Lock()
 		if s.state == streamDone {
 			s.mu.Unlock()
@@ -474,21 +411,14 @@ func (t *http2Server) handleData(f *http2.DataFrame) {
 			t.controlBuf.put(&resetStream{s.id, http2.ErrCodeFlowControl})
 			return
 		}
-		if f.Header().Flags.Has(http2.FlagDataPadded) {
-			if w := s.fc.onRead(uint32(size) - uint32(len(f.Data()))); w > 0 {
-				t.controlBuf.put(&windowUpdate{s.id, w})
-			}
-		}
 		s.mu.Unlock()
 		// TODO(bradfitz, zhaoq): A copy is required here because there is no
 		// guarantee f.Data() is consumed before the arrival of next frame.
 		// Can this copy be eliminated?
-		if len(f.Data()) > 0 {
-			data := make([]byte, len(f.Data()))
+		data := make([]byte, size)
 		copy(data, f.Data())
 		s.write(recvMsg{data: data})
 	}
-	}
 	if f.Header().Flags.Has(http2.FlagDataEndStream) {
 		// Received the end of stream from the client.
 		s.mu.Lock()
@@ -521,11 +451,6 @@ func (t *http2Server) handleSettings(f *http2.SettingsFrame) {
 	t.controlBuf.put(&settings{ack: true, ss: ss})
 }
 
-const (
-	maxPingStrikes     = 2
-	defaultPingTimeout = 2 * time.Hour
-)
-
 func (t *http2Server) handlePing(f *http2.PingFrame) {
 	if f.IsAck() { // Do nothing.
 		return
@@ -533,38 +458,6 @@ func (t *http2Server) handlePing(f *http2.PingFrame) {
 	pingAck := &ping{ack: true}
 	copy(pingAck.data[:], f.Data[:])
 	t.controlBuf.put(pingAck)
-
-	now := time.Now()
-	defer func() {
-		t.lastPingAt = now
-	}()
-	// A reset ping strikes means that we don't need to check for policy
-	// violation for this ping and the pingStrikes counter should be set
-	// to 0.
-	if atomic.CompareAndSwapUint32(&t.resetPingStrikes, 1, 0) {
-		t.pingStrikes = 0
-		return
-	}
-	t.mu.Lock()
-	ns := len(t.activeStreams)
-	t.mu.Unlock()
-	if ns < 1 && !t.kep.PermitWithoutStream {
-		// Keepalive shouldn't be active thus, this new ping should
-		// have come after atleast defaultPingTimeout.
-		if t.lastPingAt.Add(defaultPingTimeout).After(now) {
-			t.pingStrikes++
-		}
-	} else {
-		// Check if keepalive policy is respected.
-		if t.lastPingAt.Add(t.kep.MinTime).After(now) {
-			t.pingStrikes++
-		}
-	}
-
-	if t.pingStrikes > maxPingStrikes {
-		// Send goaway and close the connection.
-		t.controlBuf.put(&goAway{code: http2.ErrCodeEnhanceYourCalm, debugData: []byte("too_many_pings")})
-	}
 }
 
 func (t *http2Server) handleWindowUpdate(f *http2.WindowUpdateFrame) {
@@ -583,13 +476,6 @@ func (t *http2Server) writeHeaders(s *Stream, b *bytes.Buffer, endStream bool) e
 	first := true
 	endHeaders := false
 	var err error
-	defer func() {
-		if err == nil {
-			// Reset ping strikes when seding headers since that might cause the
-			// peer to send ping.
-			atomic.StoreUint32(&t.resetPingStrikes, 1)
-		}
-	}()
 	// Sends the headers in a single batch.
 	for !endHeaders {
 		size := t.hBuf.Len()
@@ -671,7 +557,7 @@ func (t *http2Server) WriteHeader(s *Stream, md metadata.MD) error {
 // There is no further I/O operations being able to perform on this stream.
 // TODO(zhaoq): Now it indicates the end of entire stream. Revisit if early
 // OK is adopted.
-func (t *http2Server) WriteStatus(s *Stream, st *status.Status) error {
+func (t *http2Server) WriteStatus(s *Stream, statusCode codes.Code, statusDesc string) error {
 	var headersSent, hasHeader bool
 	s.mu.Lock()
 	if s.state == streamDone {
@@ -702,24 +588,9 @@ func (t *http2Server) WriteStatus(s *Stream, st *status.Status) error {
 	t.hEnc.WriteField(
 		hpack.HeaderField{
 			Name:  "grpc-status",
-			Value: strconv.Itoa(int(st.Code())),
+			Value: strconv.Itoa(int(statusCode)),
 		})
-	t.hEnc.WriteField(hpack.HeaderField{Name: "grpc-message", Value: encodeGrpcMessage(st.Message())})
-
-	if p := st.Proto(); p != nil && len(p.Details) > 0 {
-		stBytes, err := proto.Marshal(p)
-		if err != nil {
-			// TODO: return error instead, when callers are able to handle it.
-			panic(err)
-		}
-
-		for k, v := range metadata.New(map[string]string{"grpc-status-details-bin": (string)(stBytes)}) {
-			for _, entry := range v {
-				t.hEnc.WriteField(hpack.HeaderField{Name: k, Value: entry})
-			}
-		}
-	}
-
+	t.hEnc.WriteField(hpack.HeaderField{Name: "grpc-message", Value: encodeGrpcMessage(statusDesc)})
 	// Attach the trailer metadata.
 	for k, v := range s.trailer {
 		// Clients don't tolerate reading restricted headers after some non restricted ones were sent.
@@ -748,7 +619,7 @@ func (t *http2Server) WriteStatus(s *Stream, st *status.Status) error {
 
 // Write converts the data into HTTP2 data frame and sends it out. Non-nil error
 // is returns if it fails (e.g., framing error, transport error).
-func (t *http2Server) Write(s *Stream, data []byte, opts *Options) (err error) {
+func (t *http2Server) Write(s *Stream, data []byte, opts *Options) error {
 	// TODO(zhaoq): Support multi-writers for a single stream.
 	var writeHeaderFrame bool
 	s.mu.Lock()
@@ -763,13 +634,6 @@ func (t *http2Server) Write(s *Stream, data []byte, opts *Options) (err error) {
 	if writeHeaderFrame {
 		t.WriteHeader(s, nil)
 	}
-	defer func() {
-		if err == nil {
-			// Reset ping strikes when sending data since this might cause
-			// the peer to send ping.
-			atomic.StoreUint32(&t.resetPingStrikes, 1)
-		}
-	}()
 	r := bytes.NewBuffer(data)
 	for {
 		if r.Len() == 0 {
@@ -859,91 +723,6 @@ func (t *http2Server) applySettings(ss []http2.Setting) {
 	}
 }
 
-// keepalive running in a separate goroutine does the following:
-// 1. Gracefully closes an idle connection after a duration of keepalive.MaxConnectionIdle.
-// 2. Gracefully closes any connection after a duration of keepalive.MaxConnectionAge.
-// 3. Forcibly closes a connection after an additive period of keepalive.MaxConnectionAgeGrace over keepalive.MaxConnectionAge.
-// 4. Makes sure a connection is alive by sending pings with a frequency of keepalive.Time and closes a non-resposive connection
-// after an additional duration of keepalive.Timeout.
-func (t *http2Server) keepalive() {
-	p := &ping{}
-	var pingSent bool
-	maxIdle := time.NewTimer(t.kp.MaxConnectionIdle)
-	maxAge := time.NewTimer(t.kp.MaxConnectionAge)
-	keepalive := time.NewTimer(t.kp.Time)
-	// NOTE: All exit paths of this function should reset their
-	// respecitve timers. A failure to do so will cause the
-	// following clean-up to deadlock and eventually leak.
-	defer func() {
-		if !maxIdle.Stop() {
-			<-maxIdle.C
-		}
-		if !maxAge.Stop() {
-			<-maxAge.C
-		}
-		if !keepalive.Stop() {
-			<-keepalive.C
-		}
-	}()
-	for {
-		select {
-		case <-maxIdle.C:
-			t.mu.Lock()
-			idle := t.idle
-			if idle.IsZero() { // The connection is non-idle.
-				t.mu.Unlock()
-				maxIdle.Reset(t.kp.MaxConnectionIdle)
-				continue
-			}
-			val := t.kp.MaxConnectionIdle - time.Since(idle)
-			if val <= 0 {
-				// The connection has been idle for a duration of keepalive.MaxConnectionIdle or more.
-				// Gracefully close the connection.
-				t.state = draining
-				t.mu.Unlock()
-				t.Drain()
-				// Reseting the timer so that the clean-up doesn't deadlock.
-				maxIdle.Reset(infinity)
-				return
-			}
-			t.mu.Unlock()
-			maxIdle.Reset(val)
-		case <-maxAge.C:
-			t.mu.Lock()
-			t.state = draining
-			t.mu.Unlock()
-			t.Drain()
-			maxAge.Reset(t.kp.MaxConnectionAgeGrace)
-			select {
-			case <-maxAge.C:
-				// Close the connection after grace period.
-				t.Close()
-				// Reseting the timer so that the clean-up doesn't deadlock.
-				maxAge.Reset(infinity)
-			case <-t.shutdownChan:
-			}
-			return
-		case <-keepalive.C:
-			if atomic.CompareAndSwapUint32(&t.activity, 1, 0) {
-				pingSent = false
-				keepalive.Reset(t.kp.Time)
-				continue
-			}
-			if pingSent {
-				t.Close()
-				// Reseting the timer so that the clean-up doesn't deadlock.
-				keepalive.Reset(infinity)
-				return
-			}
-			pingSent = true
-			t.controlBuf.put(p)
-			keepalive.Reset(t.kp.Timeout)
-		case <-t.shutdownChan:
-			return
-		}
-	}
-}
-
 // controller running in a separate goroutine takes charge of sending control
 // frames (e.g., window update, reset stream, setting, etc.) to the server.
 func (t *http2Server) controller() {
@@ -975,10 +754,7 @@ func (t *http2Server) controller() {
 					sid := t.maxStreamID
 					t.state = draining
 					t.mu.Unlock()
-					t.framer.writeGoAway(true, sid, i.code, i.debugData)
-					if i.code == http2.ErrCodeEnhanceYourCalm {
-						t.Close()
-					}
+					t.framer.writeGoAway(true, sid, http2.ErrCodeNo, nil)
 				case *flushIO:
 					t.framer.flushWrite()
 				case *ping:
@@ -1028,9 +804,6 @@ func (t *http2Server) Close() (err error) {
 func (t *http2Server) closeStream(s *Stream) {
 	t.mu.Lock()
 	delete(t.activeStreams, s.id)
-	if len(t.activeStreams) == 0 {
-		t.idle = time.Now()
-	}
 	if t.state == draining && len(t.activeStreams) == 0 {
 		defer t.Close()
 	}
@@ -1058,17 +831,5 @@ func (t *http2Server) RemoteAddr() net.Addr {
 }
 
 func (t *http2Server) Drain() {
-	t.controlBuf.put(&goAway{code: http2.ErrCodeNo})
-}
-
-var rgen = rand.New(rand.NewSource(time.Now().UnixNano()))
-
-func getJitter(v time.Duration) time.Duration {
-	if v == infinity {
-		return 0
-	}
-	// Generate a jitter between +/- 10% of the value.
-	r := int64(v / 10)
-	j := rgen.Int63n(2*r) - r
-	return time.Duration(j)
+	t.controlBuf.put(&goAway{})
 }

go/vendor/google.golang.org/grpc/transport/http_util.go

@@ -44,17 +44,16 @@ import (
 	"sync/atomic"
 	"time"
 
-	"github.com/golang/protobuf/proto"
 	"golang.org/x/net/http2"
 	"golang.org/x/net/http2/hpack"
-	spb "google.golang.org/genproto/googleapis/rpc/status"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/grpclog"
 	"google.golang.org/grpc/metadata"
-	"google.golang.org/grpc/status"
 )
 
 const (
+	// The primary user agent
+	primaryUA = "grpc-go/1.0"
 	// http2MaxFrameLen specifies the max length of a HTTP2 frame.
 	http2MaxFrameLen = 16384 // 16KB frame
 	// http://http2.github.io/http2-spec/#SettingValues
@@ -93,15 +92,13 @@ var (
 // Records the states during HPACK decoding. Must be reset once the
 // decoding of the entire headers are finished.
 type decodeState struct {
+	err error // first error encountered decoding
+
 	encoding string
-	// statusGen caches the stream status received from the trailer the server
-	// sent.  Client side only.  Do not access directly.  After all trailers are
-	// parsed, use the status method to retrieve the status.
-	statusGen *status.Status
-	// rawStatusCode and rawStatusMsg are set from the raw trailer fields and are not
-	// intended for direct access outside of parsing.
-	rawStatusCode int32
-	rawStatusMsg  string
+	// statusCode caches the stream status received from the trailer
+	// the server sent. Client side only.
+	statusCode codes.Code
+	statusDesc string
 	// Server side only fields.
 	timeoutSet bool
 	timeout    time.Duration
@@ -124,7 +121,6 @@ func isReservedHeader(hdr string) bool {
 		"grpc-message",
 		"grpc-status",
 		"grpc-timeout",
-		"grpc-status-details-bin",
 		"te":
 		return true
 	default:
@@ -143,6 +139,12 @@ func isWhitelistedPseudoHeader(hdr string) bool {
 	}
 }
 
+func (d *decodeState) setErr(err error) {
+	if d.err == nil {
+		d.err = err
+	}
+}
+
 func validContentType(t string) bool {
 	e := "application/grpc"
 	if !strings.HasPrefix(t, e) {
@@ -156,62 +158,56 @@ func validContentType(t string) bool {
 	return true
 }
 
-func (d *decodeState) status() *status.Status {
-	if d.statusGen == nil {
-		// No status-details were provided; generate status using code/msg.
-		d.statusGen = status.New(codes.Code(d.rawStatusCode), d.rawStatusMsg)
-	}
-	return d.statusGen
-}
-
-func (d *decodeState) processHeaderField(f hpack.HeaderField) error {
+func (d *decodeState) processHeaderField(f hpack.HeaderField) {
 	switch f.Name {
 	case "content-type":
 		if !validContentType(f.Value) {
-			return streamErrorf(codes.FailedPrecondition, "transport: received the unexpected content-type %q", f.Value)
+			d.setErr(streamErrorf(codes.FailedPrecondition, "transport: received the unexpected content-type %q", f.Value))
+			return
 		}
 	case "grpc-encoding":
 		d.encoding = f.Value
 	case "grpc-status":
 		code, err := strconv.Atoi(f.Value)
 		if err != nil {
-			return streamErrorf(codes.Internal, "transport: malformed grpc-status: %v", err)
+			d.setErr(streamErrorf(codes.Internal, "transport: malformed grpc-status: %v", err))
+			return
 		}
-		d.rawStatusCode = int32(code)
+		d.statusCode = codes.Code(code)
 	case "grpc-message":
-		d.rawStatusMsg = decodeGrpcMessage(f.Value)
-	case "grpc-status-details-bin":
-		_, v, err := metadata.DecodeKeyValue("grpc-status-details-bin", f.Value)
-		if err != nil {
-			return streamErrorf(codes.Internal, "transport: malformed grpc-status-details-bin: %v", err)
-		}
-		s := &spb.Status{}
-		if err := proto.Unmarshal([]byte(v), s); err != nil {
-			return streamErrorf(codes.Internal, "transport: malformed grpc-status-details-bin: %v", err)
-		}
-		d.statusGen = status.FromProto(s)
+		d.statusDesc = decodeGrpcMessage(f.Value)
 	case "grpc-timeout":
 		d.timeoutSet = true
 		var err error
-		if d.timeout, err = decodeTimeout(f.Value); err != nil {
-			return streamErrorf(codes.Internal, "transport: malformed time-out: %v", err)
+		d.timeout, err = decodeTimeout(f.Value)
+		if err != nil {
+			d.setErr(streamErrorf(codes.Internal, "transport: malformed time-out: %v", err))
+			return
 		}
 	case ":path":
 		d.method = f.Value
 	default:
 		if !isReservedHeader(f.Name) || isWhitelistedPseudoHeader(f.Name) {
+			if f.Name == "user-agent" {
+				i := strings.LastIndex(f.Value, " ")
+				if i == -1 {
+					// There is no application user agent string being set.
+					return
+				}
+				// Extract the application user agent string.
+				f.Value = f.Value[:i]
+			}
 			if d.mdata == nil {
 				d.mdata = make(map[string][]string)
 			}
 			k, v, err := metadata.DecodeKeyValue(f.Name, f.Value)
 			if err != nil {
 				grpclog.Printf("Failed to decode (%q, %q): %v", f.Name, f.Value, err)
-				return nil
+				return
 			}
 			d.mdata[k] = append(d.mdata[k], v)
 		}
 	}
-	return nil
 }
 
 type timeoutUnit uint8
@@ -383,9 +379,6 @@ func newFramer(conn net.Conn) *framer {
 		writer: bufio.NewWriterSize(conn, http2IOBufSize),
 	}
 	f.fr = http2.NewFramer(f.writer, f.reader)
-	// Opt-in to Frame reuse API on framer to reduce garbage.
-	// Frames aren't safe to read from after a subsequent call to ReadFrame.
-	f.fr.SetReuseFrames()
 	f.fr.ReadMetaHeaders = hpack.NewDecoder(http2InitHeaderTableSize, nil)
 	return f
 }

go/vendor/google.golang.org/grpc/transport/pre_go16.go

+// +build !go1.6
+
+/*
+ * Copyright 2016, Google Inc.
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ *
+ *     * Redistributions of source code must retain the above copyright
+ * notice, this list of conditions and the following disclaimer.
+ *     * Redistributions in binary form must reproduce the above
+ * copyright notice, this list of conditions and the following disclaimer
+ * in the documentation and/or other materials provided with the
+ * distribution.
+ *     * Neither the name of Google Inc. nor the names of its
+ * contributors may be used to endorse or promote products derived from
+ * this software without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *
+ */
+
+package transport
+
+import (
+	"net"
+	"time"
+
+	"golang.org/x/net/context"
+)
+
+// dialContext connects to the address on the named network.
+func dialContext(ctx context.Context, network, address string) (net.Conn, error) {
+	var dialer net.Dialer
+	if deadline, ok := ctx.Deadline(); ok {
+		dialer.Timeout = deadline.Sub(time.Now())
+	}
+	return dialer.Dial(network, address)
+}

go/vendor/google.golang.org/grpc/transport/transport.go

@@ -45,13 +45,10 @@ import (
 	"sync"
 
 	"golang.org/x/net/context"
-	"golang.org/x/net/http2"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/credentials"
-	"google.golang.org/grpc/keepalive"
 	"google.golang.org/grpc/metadata"
 	"google.golang.org/grpc/stats"
-	"google.golang.org/grpc/status"
 	"google.golang.org/grpc/tap"
 )
 
@@ -213,13 +210,9 @@ type Stream struct {
 	// true iff headerChan is closed. Used to avoid closing headerChan
 	// multiple times.
 	headerDone bool
-	// the status error received from the server.
-	status *status.Status
-	// rstStream indicates whether a RST_STREAM frame needs to be sent
-	// to the server to signify that this stream is closing.
-	rstStream bool
-	// rstError is the error that needs to be sent along with the RST_STREAM frame.
-	rstError http2.ErrCode
+	// the status received from the server.
+	statusCode codes.Code
+	statusDesc string
 }
 
 // RecvCompress returns the compression algorithm applied to the inbound
@@ -284,9 +277,14 @@ func (s *Stream) Method() string {
 	return s.method
 }
 
-// Status returns the status received from the server.
-func (s *Stream) Status() *status.Status {
-	return s.status
+// StatusCode returns statusCode received from the server.
+func (s *Stream) StatusCode() codes.Code {
+	return s.statusCode
+}
+
+// StatusDesc returns statusDesc received from the server.
+func (s *Stream) StatusDesc() string {
+	return s.statusDesc
 }
 
 // SetHeader sets the header metadata. This can be called multiple times.
@@ -333,20 +331,6 @@ func (s *Stream) Read(p []byte) (n int, err error) {
 	return
 }
 
-// finish sets the stream's state and status, and closes the done channel.
-// s.mu must be held by the caller.  st must always be non-nil.
-func (s *Stream) finish(st *status.Status) {
-	s.status = st
-	s.state = streamDone
-	close(s.done)
-}
-
-// GoString is implemented by Stream so context.String() won't
-// race when printing %#v.
-func (s *Stream) GoString() string {
-	return fmt.Sprintf("<stream: %p, %v>", s, s.method)
-}
-
 // The key to save transport.Stream in the context.
 type streamKey struct{}
 
@@ -378,8 +362,6 @@ type ServerConfig struct {
 	AuthInfo     credentials.AuthInfo
 	InTapHandle  tap.ServerInHandle
 	StatsHandler stats.Handler
-	KeepaliveParams keepalive.ServerParameters
-	KeepalivePolicy keepalive.EnforcementPolicy
 }
 
 // NewServerTransport creates a ServerTransport with conn or non-nil error
@@ -392,9 +374,6 @@ func NewServerTransport(protocol string, conn net.Conn, config *ServerConfig) (S
 type ConnectOptions struct {
 	// UserAgent is the application user agent.
 	UserAgent string
-	// Authority is the :authority pseudo-header to use. This field has no effect if
-	// TransportCredentials is set.
-	Authority string
 	// Dialer specifies how to dial a network address.
 	Dialer func(context.Context, string) (net.Conn, error)
 	// FailOnNonTempDialError specifies if gRPC fails on non-temporary dial errors.
@@ -403,8 +382,6 @@ type ConnectOptions struct {
 	PerRPCCredentials []credentials.PerRPCCredentials
 	// TransportCredentials stores the Authenticator required to setup a client connection.
 	TransportCredentials credentials.TransportCredentials
-	// KeepaliveParams stores the keepalive parameters.
-	KeepaliveParams keepalive.ClientParameters
 	// StatsHandler stores the handler for stats.
 	StatsHandler stats.Handler
 }
@@ -493,9 +470,6 @@ type ClientTransport interface {
 	// receives the draining signal from the server (e.g., GOAWAY frame in
 	// HTTP/2).
 	GoAway() <-chan struct{}
-
-	// GetGoAwayReason returns the reason why GoAway frame was received.
-	GetGoAwayReason() GoAwayReason
 }
 
 // ServerTransport is the common interface for all gRPC server-side transport
@@ -515,9 +489,10 @@ type ServerTransport interface {
 	// Write may not be called on all streams.
 	Write(s *Stream, data []byte, opts *Options) error
 
-	// WriteStatus sends the status of a stream to the client.  WriteStatus is
-	// the final call made on a stream and always occurs.
-	WriteStatus(s *Stream, st *status.Status) error
+	// WriteStatus sends the status of a stream to the client.
+	// WriteStatus is the final call made on a stream and always
+	// occurs.
+	WriteStatus(s *Stream, statusCode codes.Code, statusDesc string) error
 
 	// Close tears down the transport. Once it is called, the transport
 	// should not be accessed any more. All the pending streams and their
@@ -583,8 +558,6 @@ var (
 	ErrStreamDrain = streamErrorf(codes.Unavailable, "the server stops accepting new RPCs")
 )
 
-// TODO: See if we can replace StreamError with status package errors.
-
 // StreamError is an error that only affects one stream within a connection.
 type StreamError struct {
 	Code codes.Code
@@ -592,7 +565,7 @@ type StreamError struct {
 }
 
 func (e StreamError) Error() string {
-	return fmt.Sprintf("stream error: code = %s desc = %q", e.Code, e.Desc)
+	return fmt.Sprintf("stream error: code = %d desc = %q", e.Code, e.Desc)
 }
 
 // ContextErr converts the error from context package into a StreamError.
@@ -633,16 +606,3 @@ func wait(ctx context.Context, done, goAway, closing <-chan struct{}, proceed <-
 		return i, nil
 	}
 }
-
-// GoAwayReason contains the reason for the GoAway frame received.
-type GoAwayReason uint8
-
-const (
-	// Invalid indicates that no GoAway frame is received.
-	Invalid GoAwayReason = 0
-	// NoReason is the default value when GoAway frame is received.
-	NoReason GoAwayReason = 1
-	// TooManyPings indicates that a GoAway frame with ErrCodeEnhanceYourCalm
-	// was recieved and that the debug data said "too_many_pings".
-	TooManyPings GoAwayReason = 2
-)

go/vendor/vendor.json

@@ -132,10 +132,10 @@
 			"revisionTime": "2017-04-04T13:20:09Z"
 		},
 		{
-			"checksumSHA1": "tidJMmntKTZuU196aiLojkULL+g=",
+			"checksumSHA1": "epHwh7hDQSYzDowPIbw8vnLzPS0=",
 			"path": "google.golang.org/grpc",
-			"revision": "6d158dbf32084eac5fc0b9ea6f1feed214290ec6",
-			"revisionTime": "2017-04-12T06:39:30Z"
+			"revision": "50955793b0183f9de69bd78e2ec251cf20aab121",
+			"revisionTime": "2017-01-11T19:10:52Z"
 		},
 		{
 			"checksumSHA1": "08icuA15HRkdYCt6H+Cs90RPQsY=",
@@ -204,10 +204,10 @@
 			"revisionTime": "2017-04-12T06:39:30Z"
 		},
 		{
-			"checksumSHA1": "WMlN+OrgFM70j2/AoMh6DM6NtK8=",
+			"checksumSHA1": "yHpUeGwKoqqwd3cbEp3lkcnvft0=",
 			"path": "google.golang.org/grpc/transport",
-			"revision": "6d158dbf32084eac5fc0b9ea6f1feed214290ec6",
-			"revisionTime": "2017-04-12T06:39:30Z"
+			"revision": "50955793b0183f9de69bd78e2ec251cf20aab121",
+			"revisionTime": "2017-01-11T19:10:52Z"
 		},
 		{
 			"checksumSHA1": "fALlQNY1fM99NesfLJ50KguWsio=",