func doPerRPCCreds(tc testpb.TestServiceClient) {
	jsonKey := getServiceAccountJSONKey()
	pl := newPayload(testpb.PayloadType_COMPRESSABLE, largeReqSize)
	req := &testpb.SimpleRequest{
		ResponseType:   testpb.PayloadType_COMPRESSABLE.Enum(),
		ResponseSize:   proto.Int32(int32(largeRespSize)),
		Payload:        pl,
		FillUsername:   proto.Bool(true),
		FillOauthScope: proto.Bool(true),
	}
	token := getToken()
	kv := map[string]string{"authorization": token.TokenType + " " + token.AccessToken}
	ctx := metadata.NewContext(context.Background(), metadata.MD{"authorization": []string{kv["authorization"]}})
	reply, err := tc.UnaryCall(ctx, req)
	if err != nil {
		grpclog.Fatal("/TestService/UnaryCall RPC failed: ", err)
	}
	user := reply.GetUsername()
	scope := reply.GetOauthScope()
	if !strings.Contains(string(jsonKey), user) {
		grpclog.Fatalf("Got user name %q which is NOT a substring of %q.", user, jsonKey)
	}
	if !strings.Contains(*oauthScope, scope) {
		grpclog.Fatalf("Got OAuth scope %q which is NOT a substring of %q.", scope, *oauthScope)
	}
	grpclog.Println("PerRPCCreds done")
}

func doClientStreaming(tc testpb.TestServiceClient) {
	stream, err := tc.StreamingInputCall(context.Background())
	if err != nil {
		grpclog.Fatalf("%v.StreamingInputCall(_) = _, %v", tc, err)
	}
	var sum int
	for _, s := range reqSizes {
		pl := newPayload(testpb.PayloadType_COMPRESSABLE, s)
		req := &testpb.StreamingInputCallRequest{
			Payload: pl,
		}
		if err := stream.Send(req); err != nil {
			grpclog.Fatalf("%v.Send(%v) = %v", stream, req, err)
		}
		sum += s
		grpclog.Printf("Sent a request of size %d, aggregated size %d", s, sum)

	}
	reply, err := stream.CloseAndRecv()
	if err != nil {
		grpclog.Fatalf("%v.CloseAndRecv() got error %v, want %v", stream, err, nil)
	}
	if reply.GetAggregatedPayloadSize() != int32(sum) {
		grpclog.Fatalf("%v.CloseAndRecv().GetAggregatePayloadSize() = %v; want %v", stream, reply.GetAggregatedPayloadSize(), sum)
	}
	grpclog.Println("ClientStreaming done")
}

//Start Start
func Start(port int, tls bool, certFile, keyFile string, ch chan string) {

	log.Println("demo start")
	sid = utils.Randstr20()

	s := new(grpcService)
	lis, err := net.Listen("tcp", fmt.Sprintf(":%d", port))
	if err != nil {
		grpclog.Fatalf("failed to listen: %v", err)
	}
	var opts []grpc.ServerOption
	if tls {
		creds, err := credentials.NewServerTLSFromFile(certFile, keyFile)
		if err != nil {
			grpclog.Fatalf("Failed to generate credentials %v", err)
		}
		opts = []grpc.ServerOption{grpc.Creds(creds)}
	}

	grpcServer := grpc.NewServer(opts...)
	demo.RegisterRPCServer(grpcServer, s)
	go func() {
		<-ch
		grpcServer.Stop()
	}()
	grpcServer.Serve(lis)

	log.Println("demo end")
}

func main() {
	flag.Parse()
	lis, err := net.Listen("tcp", *serverAddr)
	if err != nil {
		grpclog.Fatalf("failed to listen: %v", err)
	}
	var opts []grpc.ServerOption
	if *tls {
		creds, err := credentials.NewServerTLSFromFile(*certFile, *keyFile)
		if err != nil {
			grpclog.Fatalf("Failed to generate credentials %v", err)
		}
		opts = append(opts, grpc.Creds(creds))
	}
	grpcServer := grpc.NewServer(opts...)
	oidcClient, err := util.GetOIDCClient(*clientID, *clientSecret, *discovery, *redirectURL)
	if err != nil {
		grpclog.Fatalf("unable to get oidc client: %s", err)
	}
	s, err := server.NewRoloServer(oidcClient, *policyFile)
	if err != nil {
		grpclog.Fatalln("unable to create ca from parent:", err)
	}
	pb.RegisterRoloServer(grpcServer, s)
	grpclog.Println("serving at", *serverAddr)
	grpcServer.Serve(lis)
}

// RegisterService register a service and its implementation to the gRPC
// server. Called from the IDL generated code. This must be called before
// invoking Serve.
func (s *Server) RegisterService(sd *ServiceDesc, ss interface{}) {
	s.mu.Lock()
	defer s.mu.Unlock()
	// Does some sanity checks.
	if _, ok := s.m[sd.ServiceName]; ok {
		grpclog.Fatalf("grpc: Server.RegisterService found duplicate service registration for %q", sd.ServiceName)
	}
	ht := reflect.TypeOf(sd.HandlerType).Elem()
	st := reflect.TypeOf(ss)
	if !st.Implements(ht) {
		grpclog.Fatalf("grpc: Server.RegisterService found the handler of type %v that does not satisfy %v", st, ht)
	}
	srv := &service{
		server: ss,
		md:     make(map[string]*MethodDesc),
		sd:     make(map[string]*StreamDesc),
	}
	for i := range sd.Methods {
		d := &sd.Methods[i]
		srv.md[d.MethodName] = d
	}
	for i := range sd.Streams {
		d := &sd.Streams[i]
		srv.sd[d.StreamName] = d
	}
	s.m[sd.ServiceName] = srv
}

// add_songs add one or more songs to DB.
func add_songs(client pb.SongSrvClient, SongsObj []pb.SongObj, client_id int) {
	grpclog.Printf("CLIENT-%v: Call to Adds Function with list of %v SongObj\n", client_id, len(SongsObj))
	songs := uniq_id(SongsObj, client_id)
	stream, err := client.Adds(context.Background())
	if err != nil {
		grpclog.Fatalf("%v.add_songs(_) = _, %v", client, err)
	}
	waitc := make(chan struct{})
	go func() {
		for {
			in, err := stream.Recv()
			if err == io.EOF {
				// read done.
				close(waitc)
				return
			}
			if err != nil {
				grpclog.Fatalf("CLIENT-%v: Failed to add song Id %v: %v", client_id, in.Id, err)
			}
			grpclog.Printf("CLIENT-%v: Successful add song id %v\n", client_id, in.Id)
		}
	}()
	for _, song := range songs {
		if err := stream.Send(&song); err != nil {
			grpclog.Fatalf("CLIENT-%v: Failed to send a song id %v : %v", client_id, song.Id, err)
		}
	}
	stream.CloseSend()
	<-waitc
	grpclog.Printf("\n")
}

// DoStatusCodeAndMessage checks that the status code is propagated back to the client.
func DoStatusCodeAndMessage(tc testpb.TestServiceClient) {
	var code int32 = 2
	msg := "test status message"
	expectedErr := grpc.Errorf(codes.Code(code), msg)
	respStatus := &testpb.EchoStatus{
		Code:    proto.Int32(code),
		Message: proto.String(msg),
	}
	// Test UnaryCall.
	req := &testpb.SimpleRequest{
		ResponseStatus: respStatus,
	}
	if _, err := tc.UnaryCall(context.Background(), req); err == nil || err.Error() != expectedErr.Error() {
		grpclog.Fatalf("%v.UnaryCall(_, %v) = _, %v, want _, %v", tc, req, err, expectedErr)
	}
	// Test FullDuplexCall.
	stream, err := tc.FullDuplexCall(context.Background())
	if err != nil {
		grpclog.Fatalf("%v.FullDuplexCall(_) = _, %v, want <nil>", tc, err)
	}
	streamReq := &testpb.StreamingOutputCallRequest{
		ResponseStatus: respStatus,
	}
	if err := stream.Send(streamReq); err != nil {
		grpclog.Fatalf("%v.Send(%v) = %v, want <nil>", stream, streamReq, err)
	}
	if err := stream.CloseSend(); err != nil {
		grpclog.Fatalf("%v.CloseSend() = %v, want <nil>", stream, err)
	}
	if _, err = stream.Recv(); err.Error() != expectedErr.Error() {
		grpclog.Fatalf("%v.Recv() returned error %v, want %v", stream, err, expectedErr)
	}
}

func (s *testServer) UnaryCall(ctx context.Context, in *testpb.SimpleRequest) (*testpb.SimpleResponse, error) {
	md, ok := metadata.FromContext(ctx)
	if ok {
		if err := grpc.SendHeader(ctx, md); err != nil {
			grpclog.Fatalf("grpc.SendHeader(%v, %v) = %v, want %v", ctx, md, err, nil)
		}
		grpc.SetTrailer(ctx, md)
	}
	if s.security != "" {
		// Check Auth info
		authInfo, ok := credentials.FromContext(ctx)
		if !ok {
			grpclog.Fatalf("Failed to get AuthInfo from ctx.")
		}
		var authType string
		switch info := authInfo.(type) {
		case credentials.TLSInfo:
			authType = info.AuthType()
		default:
			grpclog.Fatalf("Unknown AuthInfo type")
		}
		if authType != s.security {
			grpclog.Fatalf("Wrong auth type: got %q, want %q", authType, s.security)
		}
	}

	// Simulate some service delay.
	time.Sleep(time.Second)
	return &testpb.SimpleResponse{
		Payload: newPayload(in.GetResponseType(), in.GetResponseSize()),
	}, nil
}

func main() {
	// 2. 解析flags
	flag.Parse()

	// 3. 监听端口
	lis, err := net.Listen("tcp", fmt.Sprintf(":%d", *port))
	if err != nil {
		grpclog.Fatalf("failed to listen: %v", err)
	}

	// 4. 指定TLS
	var opts []grpc.ServerOption
	if *tls {
		creds, err := credentials.NewServerTLSFromFile(*certFile, *keyFile)
		if err != nil {
			grpclog.Fatalf("Failed to generate credentials %v", err)
		}
		opts = []grpc.ServerOption{grpc.Creds(creds)}
	}

	// 关联Server和Handler
	grpcServer := grpc.NewServer(opts...)
	pb.RegisterRouteGuideServer(grpcServer, newServer())
	grpcServer.Serve(lis)
}

func doCancelAfterFirstResponse(tc testpb.TestServiceClient) {
	ctx, cancel := context.WithCancel(context.Background())
	stream, err := tc.FullDuplexCall(ctx)
	if err != nil {
		grpclog.Fatalf("%v.FullDuplexCall(_) = _, %v", tc, err)
	}
	respParam := []*testpb.ResponseParameters{
		{
			Size: proto.Int32(31415),
		},
	}
	pl := newPayload(testpb.PayloadType_COMPRESSABLE, 27182)
	req := &testpb.StreamingOutputCallRequest{
		ResponseType:       testpb.PayloadType_COMPRESSABLE.Enum(),
		ResponseParameters: respParam,
		Payload:            pl,
	}
	if err := stream.Send(req); err != nil {
		grpclog.Fatalf("%v.Send(%v) = %v", stream, req, err)
	}
	if _, err := stream.Recv(); err != nil {
		grpclog.Fatalf("%v.Recv() = %v", stream, err)
	}
	cancel()
	if _, err := stream.Recv(); grpc.Code(err) != codes.Canceled {
		grpclog.Fatalf("%v compleled with error code %d, want %d", stream, grpc.Code(err), codes.Canceled)
	}
	grpclog.Println("CancelAfterFirstResponse done")
}

func printMetrics(client metricspb.MetricsServiceClient, totalOnly bool) {
	stream, err := client.GetAllGauges(context.Background(), &metricspb.EmptyMessage{})
	if err != nil {
		grpclog.Fatalf("failed to call GetAllGuages: %v", err)
	}

	var (
		overallQPS int64
		rpcStatus  error
	)
	for {
		gaugeResponse, err := stream.Recv()
		if err != nil {
			rpcStatus = err
			break
		}
		if _, ok := gaugeResponse.GetValue().(*metricspb.GaugeResponse_LongValue); !ok {
			panic(fmt.Sprintf("gauge %s is not a long value", gaugeResponse.Name))
		}
		v := gaugeResponse.GetLongValue()
		if !totalOnly {
			grpclog.Printf("%s: %d", gaugeResponse.Name, v)
		}
		overallQPS += v
	}
	if rpcStatus != io.EOF {
		grpclog.Fatalf("failed to finish server streaming: %v", rpcStatus)
	}
	grpclog.Printf("overall qps: %d", overallQPS)
}

// runRecordRoute sends a sequence of points to server and expects to get a RouteSummary from server.
func runRecordRoute(client pb.RouteGuideClient) {
	// Create a random number of random points
	r := rand.New(rand.NewSource(time.Now().UnixNano()))
	pointCount := int(r.Int31n(100)) + 2 // Traverse at least two points
	var points []*pb.Point
	for i := 0; i < pointCount; i++ {
		points = append(points, randomPoint(r))
	}
	grpclog.Printf("Traversing %d points.", len(points))

	// 通过Stream的方式将数据发送到服务器
	stream, err := client.RecordRoute(context.Background())
	if err != nil {
		grpclog.Fatalf("%v.RecordRoute(_) = _, %v", client, err)
	}
	for _, point := range points {
		if err := stream.Send(point); err != nil {
			grpclog.Fatalf("%v.Send(%v) = %v", stream, point, err)
		}
	}

	// 一口气发送完毕
	reply, err := stream.CloseAndRecv()
	if err != nil {
		grpclog.Fatalf("%v.CloseAndRecv() got error %v, want %v", stream, err, nil)
	}
	grpclog.Printf("Route summary: %v", reply)
}

// DoCall performs an RPC with given stub and request and response sizes.
func DoCall(hc testClient, reqSize, respSize int) {
	pl := testpb.NewPayload(testpb.PayloadType_COMPRESSABLE, reqSize)
	req := &testpb.SimpleRequest{
		ResponseType: pl.Type,
		ResponseSize: int32(respSize),
		Payload:      pl,
	}
	buf, err := json.Marshal(req)
	if err != nil {
		grpclog.Fatalf("json marshal failed %v: %v", req, err)
	}
	sr := &testpb.SimpleResponse{}
	resp, err := hc.Post("http://"+hc.addr+"/", "test/json", bytes.NewBuffer(buf))
	if err != nil {
		grpclog.Fatalf("post failed %v: %v", resp, err)
	}
	defer resp.Body.Close()
	body, err := ioutil.ReadAll(resp.Body)
	if err != nil {
		grpclog.Fatalf("read failed %v: %v", resp, err)
	}
	err = json.Unmarshal(body, sr)
	if err != nil {
		grpclog.Fatalf("json unmarshal failed %v: %v", resp, err)
	}
}

// loadFeatures loads features from a JSON file.
func (s *routeGuideServer) loadFeatures(filePath string) {
	file, err := ioutil.ReadFile(filePath)
	if err != nil {
		grpclog.Fatalf("Failed to load default features: %v", err)
	}
	if err := json.Unmarshal(file, &s.savedFeatures); err != nil {
		grpclog.Fatalf("Failed to load default features: %v", err)
	}
}

func demoConn(gwid string) {
	var opts []grpc.DialOption
	var err error
	if demoTLS {
		var sn string
		if demoOver != "" {
			sn = demoOver
		}
		var creds credentials.TransportCredentials
		if demoFile != "" {

			creds, err = credentials.NewClientTLSFromFile(demoFile, sn)
			if err != nil {
				grpclog.Fatalf("Failed to create TLS credentials %v", err)
			}
		} else {
			creds = credentials.NewClientTLSFromCert(nil, sn)
		}
		opts = append(opts, grpc.WithTransportCredentials(creds))
	} else {
		opts = append(opts, grpc.WithInsecure())
	}
	conn, err := grpc.Dial(demoAddr, opts...)
	if err != nil {
		grpclog.Fatalf("fail to dial: %v", err)
	}
	//defer demoCliConn.Close()

	demoClient = demo.NewRPCClient(conn)

	go func() {
		for {

			ctx, _ := context.WithTimeout(context.Background(), 1*time.Second)

			r, err := demoClient.GWSRPCCHECK(ctx, &demo.GWSRPCCHECKSEND{GWSRPCID: gwid})
			if err != nil {
				//grpclog.Printf("demoClient %v: ", err)
				time.Sleep(100 * time.Millisecond)
				continue
			}

			if r.GWSRPCID != demoID {
				demoID = r.GWSRPCID
				h.reinitall <- "demo"
			}

			time.Sleep(100 * time.Millisecond)

		}

	}()

}

// DoCancelAfterBegin cancels the RPC after metadata has been sent but before payloads are sent.
func DoCancelAfterBegin(tc testpb.TestServiceClient) {
	ctx, cancel := context.WithCancel(metadata.NewContext(context.Background(), testMetadata))
	stream, err := tc.StreamingInputCall(ctx)
	if err != nil {
		grpclog.Fatalf("%v.StreamingInputCall(_) = _, %v", tc, err)
	}
	cancel()
	_, err = stream.CloseAndRecv()
	if grpc.Code(err) != codes.Canceled {
		grpclog.Fatalf("%v.CloseAndRecv() got error code %d, want %d", stream, grpc.Code(err), codes.Canceled)
	}
}

func getToken() *oauth2.Token {
	jsonKey := getServiceAccountJSONKey()
	config, err := google.JWTConfigFromJSON(jsonKey, *oauthScope)
	if err != nil {
		grpclog.Fatalf("Failed to get the config: %v", err)
	}
	token, err := config.TokenSource(context.Background()).Token()
	if err != nil {
		grpclog.Fatalf("Failed to get the token: %v", err)
	}
	return token
}

// DoEmptyStream sets up a bi-directional streaming with zero message.
func DoEmptyStream(tc testpb.TestServiceClient) {
	stream, err := tc.FullDuplexCall(context.Background())
	if err != nil {
		grpclog.Fatalf("%v.FullDuplexCall(_) = _, %v", tc, err)
	}
	if err := stream.CloseSend(); err != nil {
		grpclog.Fatalf("%v.CloseSend() got %v, want %v", stream, err, nil)
	}
	if _, err := stream.Recv(); err != io.EOF {
		grpclog.Fatalf("%v failed to complete the empty stream test: %v", stream, err)
	}
}

func runStream(b *testing.B, maxConcurrentCalls int) {
	s := stats.AddStats(b, 38)
	b.StopTimer()
	target, stopper := StartServer("localhost:0")
	defer stopper()
	conn := NewClientConn(target)
	tc := testpb.NewTestServiceClient(conn)

	// Warm up connection.
	stream, err := tc.StreamingCall(context.Background())
	if err != nil {
		grpclog.Fatalf("%v.StreamingCall(_) = _, %v", tc, err)
	}
	for i := 0; i < 10; i++ {
		streamCaller(tc, stream)
	}

	ch := make(chan int, maxConcurrentCalls*4)
	var (
		mu sync.Mutex
		wg sync.WaitGroup
	)
	wg.Add(maxConcurrentCalls)

	// Distribute the b.N calls over maxConcurrentCalls workers.
	for i := 0; i < maxConcurrentCalls; i++ {
		go func() {
			stream, err := tc.StreamingCall(context.Background())
			if err != nil {
				grpclog.Fatalf("%v.StreamingCall(_) = _, %v", tc, err)
			}
			for _ = range ch {
				start := time.Now()
				streamCaller(tc, stream)
				elapse := time.Since(start)
				mu.Lock()
				s.Add(elapse)
				mu.Unlock()
			}
			wg.Done()
		}()
	}
	b.StartTimer()
	for i := 0; i < b.N; i++ {
		ch <- i
	}
	b.StopTimer()
	close(ch)
	wg.Wait()
	conn.Close()
}

func main() {
	flag.Parse()
	go func() {
		lis, err := net.Listen("tcp", ":0")
		if err != nil {
			grpclog.Fatalf("Failed to listen: %v", err)
		}
		grpclog.Println("Client profiling address: ", lis.Addr().String())
		if err := http.Serve(lis, nil); err != nil {
			grpclog.Fatalf("Failed to serve: %v", err)
		}
	}()
	closeLoop()
}