opensource/grpc-go
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge branch 'master' of https://github.com/grpc/grpc-go

Browse Source
This commit is contained in:
yangzhouhan
2015-07-24 13:49:10 -07:00
parent 6cfd2022af c8555c5a9c
commit 3b4ecfaa42
7 changed files with 192 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
clientconn.go
View File

@ -114,9 +114,14 @@ func WithDialer(f func(addr string, timeout time.Duration) (net.Conn, error)) Di
}
}
// WithUserAgent returns a DialOption that specifies a user agent string for all the RPCs.
func WithUserAgent(s string) DialOption {
return func(o *dialOptions) {
o.copts.UserAgent = s
}
}
// Dial creates a client connection the given target.
// TODO(zhaoq): Have an option to make Dial return immediately without waiting
// for connection to complete.
func Dial(target string, opts ...DialOption) (*ClientConn, error) {
if target == "" {
return nil, ErrUnspecTarget

115
credentials/oauth/oauth.go Normal file
View File

@ -0,0 +1,115 @@
/*
*
* Copyright 2015, 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 oauth implements gRPC credentials using OAuth.
package oauth
import (
"fmt"
"io/ioutil"
"golang.org/x/net/context"
"golang.org/x/oauth2"
"golang.org/x/oauth2/google"
"golang.org/x/oauth2/jwt"
"google.golang.org/grpc/credentials"
)
// TokenSource supplies credentials from an oauth2.TokenSource.
type TokenSource struct {
oauth2.TokenSource
}
// GetRequestMetadata gets the request metadata as a map from a TokenSource.
func (ts TokenSource) GetRequestMetadata(ctx context.Context) (map[string]string, error) {
token, err := ts.Token()
if err != nil {
return nil, err
}
return map[string]string{
"authorization": token.TokenType + " " + token.AccessToken,
}, nil
}
// NewComputeEngine constructs the credentials that fetches access tokens from
// Google Compute Engine (GCE)'s metadata server. It is only valid to use this
// if your program is running on a GCE instance.
// TODO(dsymonds): Deprecate and remove this.
func NewComputeEngine() credentials.Credentials {
return TokenSource{google.ComputeTokenSource("")}
}
// serviceAccount represents credentials via JWT signing key.
type serviceAccount struct {
config *jwt.Config
}
func (s serviceAccount) GetRequestMetadata(ctx context.Context) (map[string]string, error) {
token, err := s.config.TokenSource(ctx).Token()
if err != nil {
return nil, err
}
return map[string]string{
"authorization": token.TokenType + " " + token.AccessToken,
}, nil
}
// NewServiceAccountFromKey constructs the credentials using the JSON key slice
// from a Google Developers service account.
func NewServiceAccountFromKey(jsonKey []byte, scope ...string) (credentials.Credentials, error) {
config, err := google.JWTConfigFromJSON(jsonKey, scope...)
if err != nil {
return nil, err
}
return serviceAccount{config: config}, nil
}
// NewServiceAccountFromFile constructs the credentials using the JSON key file
// of a Google Developers service account.
func NewServiceAccountFromFile(keyFile string, scope ...string) (credentials.Credentials, error) {
jsonKey, err := ioutil.ReadFile(keyFile)
if err != nil {
return nil, fmt.Errorf("credentials: failed to read the service account key file: %v", err)
}
return NewServiceAccountFromKey(jsonKey, scope...)
}
// NewApplicationDefault returns "Application Default Credentials". For more
// detail, see https://developers.google.com/accounts/docs/application-default-credentials.
func NewApplicationDefault(ctx context.Context, scope ...string) (credentials.Credentials, error) {
t, err := google.DefaultTokenSource(ctx, scope...)
if err != nil {
return nil, err
}
return TokenSource{t}, nil
}

65
test/end2end_test.go
View File

@ -62,15 +62,20 @@ var (
"key1": "value1",
"key2": "value2",
}
testAppUA = "myApp/1.0"
)
type testServer struct {
}
func (s *testServer) EmptyCall(ctx context.Context, in *testpb.Empty) (*testpb.Empty, error) {
if _, ok := metadata.FromContext(ctx); ok {
// For testing purpose, returns an error if there is attached metadata.
return nil, grpc.Errorf(codes.DataLoss, "got extra metadata")
if md, ok := metadata.FromContext(ctx); ok {
// For testing purpose, returns an error if there is attached metadata other than
// the user agent set by the client application.
if _, ok := md["user-agent"]; !ok {
return nil, grpc.Errorf(codes.DataLoss, "got extra metadata")
}
grpc.SendHeader(ctx, metadata.Pairs("ua", md["user-agent"]))
}
return new(testpb.Empty), nil
}
@ -285,7 +290,7 @@ func listTestEnv() []env {
return []env{env{"tcp", nil, ""}, env{"tcp", nil, "tls"}, env{"unix", unixDialer, ""}, env{"unix", unixDialer, "tls"}}
}
func setUp(hs *health.HealthServer, maxStream uint32, e env) (s *grpc.Server, cc *grpc.ClientConn) {
func setUp(hs *health.HealthServer, maxStream uint32, ua string, e env) (s *grpc.Server, cc *grpc.ClientConn) {
sopts := []grpc.ServerOption{grpc.MaxConcurrentStreams(maxStream)}
la := ":0"
switch e.network {
@ -325,9 +330,9 @@ func setUp(hs *health.HealthServer, maxStream uint32, e env) (s *grpc.Server, cc
if err != nil {
grpclog.Fatalf("Failed to create credentials %v", err)
}
cc, err = grpc.Dial(addr, grpc.WithTransportCredentials(creds), grpc.WithDialer(e.dialer))
cc, err = grpc.Dial(addr, grpc.WithTransportCredentials(creds), grpc.WithDialer(e.dialer), grpc.WithUserAgent(ua))
} else {
cc, err = grpc.Dial(addr, grpc.WithDialer(e.dialer))
cc, err = grpc.Dial(addr, grpc.WithDialer(e.dialer), grpc.WithUserAgent(ua))
}
if err != nil {
grpclog.Fatalf("Dial(%q) = %v", addr, err)
@ -347,7 +352,7 @@ func TestTimeoutOnDeadServer(t *testing.T) {
}
func testTimeoutOnDeadServer(t *testing.T, e env) {
s, cc := setUp(nil, math.MaxUint32, e)
s, cc := setUp(nil, math.MaxUint32, "", e)
tc := testpb.NewTestServiceClient(cc)
s.Stop()
// Set -1 as the timeout to make sure if transportMonitor gets error
@ -379,7 +384,7 @@ func TestHealthCheckOnSuccess(t *testing.T) {
func testHealthCheckOnSuccess(t *testing.T, e env) {
hs := health.NewHealthServer()
hs.SetServingStatus("", "grpc.health.v1alpha.HealthCheck", 1)
s, cc := setUp(hs, math.MaxUint32, e)
s, cc := setUp(hs, math.MaxUint32, "", e)
defer tearDown(s, cc)
if _, err := healthCheck(1*time.Second, cc, "grpc.health.v1alpha.HealthCheck"); err != nil {
t.Fatalf("HealthCheck/Check(_, _) = _, %v, want _, <nil>", err)
@ -395,7 +400,7 @@ func TestHealthCheckOnFailure(t *testing.T) {
func testHealthCheckOnFailure(t *testing.T, e env) {
hs := health.NewHealthServer()
hs.SetServingStatus("", "grpc.health.v1alpha.HealthCheck", 1)
s, cc := setUp(hs, math.MaxUint32, e)
s, cc := setUp(hs, math.MaxUint32, "", e)
defer tearDown(s, cc)
if _, err := healthCheck(0*time.Second, cc, "grpc.health.v1alpha.HealthCheck"); err != grpc.Errorf(codes.DeadlineExceeded, "context deadline exceeded") {
t.Fatalf("HealthCheck/Check(_, _) = _, %v, want _, error code %d", err, codes.DeadlineExceeded)
@ -409,7 +414,7 @@ func TestHealthCheckOff(t *testing.T) {
}
func testHealthCheckOff(t *testing.T, e env) {
s, cc := setUp(nil, math.MaxUint32, e)
s, cc := setUp(nil, math.MaxUint32, "", e)
defer tearDown(s, cc)
if _, err := healthCheck(1*time.Second, cc, ""); err != grpc.Errorf(codes.Unimplemented, "unknown service grpc.health.v1alpha.HealthCheck") {
t.Fatalf("HealthCheck/Check(_, _) = _, %v, want _, error code %d", err, codes.Unimplemented)
@ -424,7 +429,7 @@ func TestHealthCheckServingStatus(t *testing.T) {
func testHealthCheckServingStatus(t *testing.T, e env) {
hs := health.NewHealthServer()
s, cc := setUp(hs, math.MaxUint32, e)
s, cc := setUp(hs, math.MaxUint32, "", e)
defer tearDown(s, cc)
if _, err := healthCheck(1*time.Second, cc, "grpc.health.v1alpha.HealthCheck"); err != grpc.Errorf(codes.NotFound, "unknown service") {
t.Fatalf("HealthCheck/Check(_, _) = _, %v, want _, error code %d", err, codes.NotFound)
@ -448,20 +453,24 @@ func testHealthCheckServingStatus(t *testing.T, e env) {
}
func TestEmptyUnary(t *testing.T) {
func TestEmptyUnaryWithUserAgent(t *testing.T) {
for _, e := range listTestEnv() {
testEmptyUnary(t, e)
testEmptyUnaryWithUserAgent(t, e)
}
}
func testEmptyUnary(t *testing.T, e env) {
s, cc := setUp(nil, math.MaxUint32, e)
func testEmptyUnaryWithUserAgent(t *testing.T, e env) {
s, cc := setUp(nil, math.MaxUint32, testAppUA, e)
tc := testpb.NewTestServiceClient(cc)
defer tearDown(s, cc)
reply, err := tc.EmptyCall(context.Background(), &testpb.Empty{})
var header metadata.MD
reply, err := tc.EmptyCall(context.Background(), &testpb.Empty{}, grpc.Header(&header))
if err != nil || !proto.Equal(&testpb.Empty{}, reply) {
t.Fatalf("TestService/EmptyCall(_, _) = %v, %v, want %v, <nil>", reply, err, &testpb.Empty{})
}
if v, ok := header["ua"]; !ok || v != testAppUA {
t.Fatalf("header[\"ua\"] = %q, %t, want %q, true", v, ok, testAppUA)
}
}
func TestFailedEmptyUnary(t *testing.T) {
@ -471,7 +480,7 @@ func TestFailedEmptyUnary(t *testing.T) {
}
func testFailedEmptyUnary(t *testing.T, e env) {
s, cc := setUp(nil, math.MaxUint32, e)
s, cc := setUp(nil, math.MaxUint32, "", e)
tc := testpb.NewTestServiceClient(cc)
defer tearDown(s, cc)
ctx := metadata.NewContext(context.Background(), testMetadata)
@ -487,7 +496,7 @@ func TestLargeUnary(t *testing.T) {
}
func testLargeUnary(t *testing.T, e env) {
s, cc := setUp(nil, math.MaxUint32, e)
s, cc := setUp(nil, math.MaxUint32, "", e)
tc := testpb.NewTestServiceClient(cc)
defer tearDown(s, cc)
argSize := 271828
@ -515,7 +524,7 @@ func TestMetadataUnaryRPC(t *testing.T) {
}
func testMetadataUnaryRPC(t *testing.T, e env) {
s, cc := setUp(nil, math.MaxUint32, e)
s, cc := setUp(nil, math.MaxUint32, "", e)
tc := testpb.NewTestServiceClient(cc)
defer tearDown(s, cc)
argSize := 2718
@ -569,7 +578,7 @@ func TestRetry(t *testing.T) {
// TODO(zhaoq): Refactor to make this clearer and add more cases to test racy
// and error-prone paths.
func testRetry(t *testing.T, e env) {
s, cc := setUp(nil, math.MaxUint32, e)
s, cc := setUp(nil, math.MaxUint32, "", e)
tc := testpb.NewTestServiceClient(cc)
defer tearDown(s, cc)
var wg sync.WaitGroup
@ -599,7 +608,7 @@ func TestRPCTimeout(t *testing.T) {
// TODO(zhaoq): Have a better test coverage of timeout and cancellation mechanism.
func testRPCTimeout(t *testing.T, e env) {
s, cc := setUp(nil, math.MaxUint32, e)
s, cc := setUp(nil, math.MaxUint32, "", e)
tc := testpb.NewTestServiceClient(cc)
defer tearDown(s, cc)
argSize := 2718
@ -625,7 +634,7 @@ func TestCancel(t *testing.T) {
}
func testCancel(t *testing.T, e env) {
s, cc := setUp(nil, math.MaxUint32, e)
s, cc := setUp(nil, math.MaxUint32, "", e)
tc := testpb.NewTestServiceClient(cc)
defer tearDown(s, cc)
argSize := 2718
@ -657,7 +666,7 @@ func TestPingPong(t *testing.T) {
}
func testPingPong(t *testing.T, e env) {
s, cc := setUp(nil, math.MaxUint32, e)
s, cc := setUp(nil, math.MaxUint32, "", e)
tc := testpb.NewTestServiceClient(cc)
defer tearDown(s, cc)
stream, err := tc.FullDuplexCall(context.Background())
@ -708,7 +717,7 @@ func TestMetadataStreamingRPC(t *testing.T) {
}
func testMetadataStreamingRPC(t *testing.T, e env) {
s, cc := setUp(nil, math.MaxUint32, e)
s, cc := setUp(nil, math.MaxUint32, "", e)
tc := testpb.NewTestServiceClient(cc)
defer tearDown(s, cc)
ctx := metadata.NewContext(context.Background(), testMetadata)
@ -765,7 +774,7 @@ func TestServerStreaming(t *testing.T) {
}
func testServerStreaming(t *testing.T, e env) {
s, cc := setUp(nil, math.MaxUint32, e)
s, cc := setUp(nil, math.MaxUint32, "", e)
tc := testpb.NewTestServiceClient(cc)
defer tearDown(s, cc)
respParam := make([]*testpb.ResponseParameters, len(respSizes))
@ -817,7 +826,7 @@ func TestFailedServerStreaming(t *testing.T) {
}
func testFailedServerStreaming(t *testing.T, e env) {
s, cc := setUp(nil, math.MaxUint32, e)
s, cc := setUp(nil, math.MaxUint32, "", e)
tc := testpb.NewTestServiceClient(cc)
defer tearDown(s, cc)
respParam := make([]*testpb.ResponseParameters, len(respSizes))
@ -847,7 +856,7 @@ func TestClientStreaming(t *testing.T) {
}
func testClientStreaming(t *testing.T, e env) {
s, cc := setUp(nil, math.MaxUint32, e)
s, cc := setUp(nil, math.MaxUint32, "", e)
tc := testpb.NewTestServiceClient(cc)
defer tearDown(s, cc)
stream, err := tc.StreamingInputCall(context.Background())
@ -882,7 +891,7 @@ func TestExceedMaxStreamsLimit(t *testing.T) {
func testExceedMaxStreamsLimit(t *testing.T, e env) {
// Only allows 1 live stream per server transport.
s, cc := setUp(nil, 1, e)
s, cc := setUp(nil, 1, "", e)
tc := testpb.NewTestServiceClient(cc)
defer tearDown(s, cc)
// Perform a unary RPC to make sure the new settings were propagated to the client.

18
transport/http2_client.go
View File

@ -53,9 +53,10 @@ import (
// http2Client implements the ClientTransport interface with HTTP2.
type http2Client struct {
target string // server name/addr
conn net.Conn // underlying communication channel
nextID uint32 // the next stream ID to be used
target string // server name/addr
userAgent string
conn net.Conn // underlying communication channel
nextID uint32 // the next stream ID to be used
// writableChan synchronizes write access to the transport.
// A writer acquires the write lock by sending a value on writableChan
@ -158,10 +159,15 @@ func newHTTP2Client(addr string, opts *ConnectOptions) (_ ClientTransport, err e
return nil, ConnectionErrorf("transport: %v", err)
}
}
ua := primaryUA
if opts.UserAgent != "" {
ua += " " + opts.UserAgent
}
var buf bytes.Buffer
t := &http2Client{
target: addr,
conn: conn,
target: addr,
userAgent: ua,
conn: conn,
// The client initiated stream id is odd starting from 1.
nextID: 1,
writableChan: make(chan int, 1),
@ -273,7 +279,9 @@ func (t *http2Client) NewStream(ctx context.Context, callHdr *CallHdr) (_ *Strea
t.hEnc.WriteField(hpack.HeaderField{Name: ":path", Value: callHdr.Method})
t.hEnc.WriteField(hpack.HeaderField{Name: ":authority", Value: callHdr.Host})
t.hEnc.WriteField(hpack.HeaderField{Name: "content-type", Value: "application/grpc"})
t.hEnc.WriteField(hpack.HeaderField{Name: "user-agent", Value: t.userAgent})
t.hEnc.WriteField(hpack.HeaderField{Name: "te", Value: "trailers"})
if timeout > 0 {
t.hEnc.WriteField(hpack.HeaderField{Name: "grpc-timeout", Value: timeoutEncode(timeout)})
}

2
transport/http2_server.go
View File

@ -91,8 +91,6 @@ type http2Server struct {
func newHTTP2Server(conn net.Conn, maxStreams uint32) (_ ServerTransport, err error) {
framer := newFramer(conn)
// Send initial settings as connection preface to client.
// TODO(zhaoq): Have a better way to signal "no limit" because 0 is
// permitted in the HTTP2 spec.
var settings []http2.Setting
// TODO(zhaoq): Have a better way to signal "no limit" because 0 is
// permitted in the HTTP2 spec.

15
transport/http_util.go
View File

@ -39,6 +39,7 @@ import (
"io"
"net"
"strconv"
"strings"
"sync/atomic"
"time"
@ -50,6 +51,8 @@ import (
)
const (
// The primary user agent
primaryUA = "grpc-go/0.7"
// http2MaxFrameLen specifies the max length of a HTTP2 frame.
http2MaxFrameLen = 16384 // 16KB frame
// http://http2.github.io/http2-spec/#SettingValues
@ -128,8 +131,7 @@ func isReservedHeader(hdr string) bool {
"grpc-message",
"grpc-status",
"grpc-timeout",
"te",
"user-agent":
"te":
return true
default:
return false
@ -161,6 +163,15 @@ func newHPACKDecoder() *hpackDecoder {
d.state.method = f.Value
default:
if !isReservedHeader(f.Name) {
if f.Name == "user-agent" {
str := strings.Split(f.Value, " ")
if len(str) <= 1 {
// There is no application user agent string being set.
return
}
// Extract the application user agent string.
f.Value = str[1]
}
if d.state.mdata == nil {
d.state.mdata = make(map[string]string)
}

9
transport/transport.go
View File

@ -320,9 +320,14 @@ func NewServerTransport(protocol string, conn net.Conn, maxStreams uint32) (Serv
// ConnectOptions covers all relevant options for dialing a server.
type ConnectOptions struct {
Dialer func(string, time.Duration) (net.Conn, error)
// UserAgent is the application user agent.
UserAgent string
// Dialer specifies how to dial a network address.
Dialer func(string, time.Duration) (net.Conn, error)
// AuthOptions stores the credentials required to setup a client connection and/or issue RPCs.
AuthOptions []credentials.Credentials
Timeout time.Duration
// Timeout specifies the timeout for dialing a client connection.
Timeout time.Duration
}
// NewClientTransport establishes the transport with the required ConnectOptions