func TestMain(m *testing.M) {
	flag.Parse() // needed for testing.Short()
	ctx := context.Background()
	testProjectID = testutil.ProjID()
	if testProjectID == "" || testing.Short() {
		integrationTest = false
		if testProjectID != "" {
			log.Print("Integration tests skipped in short mode (using fake instead)")
		}
		testProjectID = "PROJECT_ID"
		addr, err := ltesting.NewServer()
		if err != nil {
			log.Fatalf("creating fake server: %v", err)
		}
		newClient = func(ctx context.Context, projectID string) *Client {
			conn, err := grpc.Dial(addr, grpc.WithInsecure(), grpc.WithBlock())
			if err != nil {
				log.Fatalf("dialing %q: %v", addr, err)
			}
			c, err := NewClient(ctx, projectID, option.WithGRPCConn(conn))
			if err != nil {
				log.Fatalf("creating client for fake at %q: %v", addr, err)
			}
			return c
		}
	} else {
		integrationTest = true
		ts := testutil.TokenSource(ctx, logging.AdminScope)
		if ts == nil {
			log.Fatal("The project key must be set. See CONTRIBUTING.md for details")
		}
		log.Printf("running integration tests with project %s", testProjectID)
		newClient = func(ctx context.Context, projectID string) *Client {
			c, err := NewClient(ctx, projectID, option.WithTokenSource(ts),
				option.WithGRPCDialOption(grpc.WithBlock()))
			if err != nil {
				log.Fatalf("creating prod client: %v", err)
			}
			return c
		}
	}
	client = newClient(ctx, testProjectID)
	initMetrics(ctx)
	cleanup := initSinks(ctx)
	exit := m.Run()
	cleanup()
	client.Close()
	os.Exit(exit)
}

// add a service
func (p *service_pool) add_service(key, value string) {
	p.Lock()
	defer p.Unlock()
	// name check
	service_name := filepath.Dir(key)
	if p.enable_name_check && !p.known_names[service_name] {
		return
	}

	// try new service kind init
	if p.services[service_name] == nil {
		p.services[service_name] = &service{}
	}

	// create service connection
	service := p.services[service_name]
	if conn, err := grpc.Dial(value, grpc.WithBlock(), grpc.WithInsecure()); err == nil {
		service.clients = append(service.clients, client{key, conn})
		log.Println("service added:", key, "-->", value)
		for k := range p.callbacks[service_name] {
			select {
			case p.callbacks[service_name][k] <- key:
			default:
			}
		}
	} else {
		log.Println("did not connect:", key, "-->", value, "error:", err)
	}
}

func BenchmarkMain(tst *testing.B) {
	conn, err := grpc.Dial("localhost:4567", grpc.WithBlock())
	if err != nil {
		os.Exit(1)
	}
	defer conn.Close()

	client := pb.NewPingClient(conn)

	tst.ResetTimer()
	for n := 0; n < tst.N; n++ {
		ctx := context.Background()
		t := time.Now()
		req := pb.PingRequest{t.UnixNano()}
		_, err := client.Ping(ctx, &req)
		if err != nil {
			tst.FailNow()
		}

		/*
			t2 := time.Now()

			then := time.Unix(0, resp.TimeBack)

			reqdelay := then.Sub(t)
			log.Println("req ", reqdelay)

			repdelay := t2.Sub(then)
			log.Println("rep ", repdelay)
		*/
	}
}

func createFactoryFunc(addrFunc AddressFunc, timeout time.Duration) pools.Factory {
	return func() (pools.Resource, error) {
		addr := addrFunc()
		// create new client
		return New(addr, grpc.WithBlock(), grpc.WithTimeout(timeout))
	}
}

func GRPCClient(addr, token, caFile string) (*RemoteU2FClient, error) {
	var err error
	var tCreds credentials.TransportCredentials
	if caFile == "" {
		tCreds = credentials.NewClientTLSFromCert(nil, "")
	} else {
		tCreds, err = credentials.NewClientTLSFromFile(caFile, "")
		if err != nil {
			return nil, fmt.Errorf("error reading CA file: %s", err)
		}
	}

	t := oauth2.Token{
		AccessToken: token,
		TokenType:   "Bearer",
	}
	rpcCreds := oauth.NewOauthAccess(&t)

	conn, err := grpc.Dial(addr, grpc.WithTransportCredentials(tCreds),
		grpc.WithPerRPCCredentials(rpcCreds),
		grpc.WithBlock(),
		grpc.WithTimeout(30*time.Second))
	if err != nil {
		return nil, fmt.Errorf("error connecting to server: %s", err)
	}

	c := pb.NewRemoteU2FClient(conn)
	return &RemoteU2FClient{c}, nil
}

// Dial establishes a connection for a given endpoint using the client's config
func (c *Client) Dial(endpoint string) (*grpc.ClientConn, error) {
	opts := []grpc.DialOption{
		grpc.WithBlock(),
		grpc.WithTimeout(c.cfg.DialTimeout),
	}
	if c.creds != nil {
		opts = append(opts, grpc.WithTransportCredentials(*c.creds))
	} else {
		opts = append(opts, grpc.WithInsecure())
	}

	proto := "tcp"
	if url, uerr := url.Parse(endpoint); uerr == nil && url.Scheme == "unix" {
		proto = "unix"
		// strip unix:// prefix so certs work
		endpoint = url.Host
	}
	f := func(a string, t time.Duration) (net.Conn, error) {
		select {
		case <-c.ctx.Done():
			return nil, c.ctx.Err()
		default:
		}
		return net.DialTimeout(proto, a, t)
	}
	opts = append(opts, grpc.WithDialer(f))

	conn, err := grpc.Dial(endpoint, opts...)
	if err != nil {
		return nil, err
	}
	return conn, nil
}

// SendSync sends a raft message and waits for an acknowledgement.
func (t *RaftTransport) SendSync(ctx context.Context, req *RaftMessageRequest) error {
	// Use the circuit breaker to fail fast if the breaker is open.
	// The same underlying connection is shared between sync and
	// async raft transports, so we use the same breaker.
	var resp *RaftMessageResponse
	nodeID := req.ToReplica.NodeID
	breaker := t.GetCircuitBreaker(nodeID)
	if err := breaker.Call(func() error {
		addr, err := t.resolver(nodeID)
		if err != nil {
			return err
		}
		conn, err := t.rpcContext.GRPCDial(addr.String(), grpc.WithBlock())
		if err != nil {
			return err
		}
		client := NewMultiRaftClient(conn)
		resp, err = client.RaftMessageSync(ctx, req)
		return err
	}, 0); err != nil {
		return err
	}

	switch val := resp.Union.GetValue().(type) {
	case *roachpb.Error:
		return val.GoError()
	case nil:
		return nil
	default:
		return errors.Errorf("unexpected response value %T %s", val, val)
	}
}

func TestDropRequestFailedNonFailFast(t *testing.T) {
	// Start a backend.
	beLis, err := net.Listen("tcp", "localhost:0")
	if err != nil {
		t.Fatalf("Failed to listen %v", err)
	}
	beAddr := strings.Split(beLis.Addr().String(), ":")
	bePort, err := strconv.Atoi(beAddr[1])
	backends := startBackends(t, besn, beLis)
	defer stopBackends(backends)

	// Start a load balancer.
	lbLis, err := net.Listen("tcp", "localhost:0")
	if err != nil {
		t.Fatalf("Failed to create the listener for the load balancer %v", err)
	}
	lbCreds := &serverNameCheckCreds{
		sn: lbsn,
	}
	lb := grpc.NewServer(grpc.Creds(lbCreds))
	if err != nil {
		t.Fatalf("Failed to generate the port number %v", err)
	}
	be := &lbpb.Server{
		IpAddress:        []byte(beAddr[0]),
		Port:             int32(bePort),
		LoadBalanceToken: lbToken,
		DropRequest:      true,
	}
	var bes []*lbpb.Server
	bes = append(bes, be)
	sl := &lbpb.ServerList{
		Servers: bes,
	}
	ls := newRemoteBalancer(sl)
	lbpb.RegisterLoadBalancerServer(lb, ls)
	go func() {
		lb.Serve(lbLis)
	}()
	defer func() {
		ls.stop()
		lb.Stop()
	}()
	creds := serverNameCheckCreds{
		expected: besn,
	}
	ctx, _ := context.WithTimeout(context.Background(), 10*time.Second)
	cc, err := grpc.DialContext(ctx, besn, grpc.WithBalancer(Balancer(&testNameResolver{
		addr: lbLis.Addr().String(),
	})), grpc.WithBlock(), grpc.WithTransportCredentials(&creds))
	if err != nil {
		t.Fatalf("Failed to dial to the backend %v", err)
	}
	helloC := hwpb.NewGreeterClient(cc)
	ctx, _ = context.WithTimeout(context.Background(), 10*time.Millisecond)
	if _, err := helloC.SayHello(ctx, &hwpb.HelloRequest{Name: "grpc"}, grpc.FailFast(false)); grpc.Code(err) != codes.DeadlineExceeded {
		t.Fatalf("%v.SayHello(_, _) = _, %v, want _, %s", helloC, err, codes.DeadlineExceeded)
	}
	cc.Close()
}

// connectAndProcess connects to the node and then processes the
// provided channel containing a queue of raft messages until there is
// an unrecoverable error with the underlying connection. A circuit
// breaker is used to allow fast failures in SendAsync which will drop
// incoming raft messages and report unreachable status to the raft group.
func (t *RaftTransport) connectAndProcess(
	nodeID roachpb.NodeID,
	ch chan *RaftMessageRequest,
	stats *raftTransportStats,
) {
	breaker := t.GetCircuitBreaker(nodeID)
	successes := breaker.Successes()
	consecFailures := breaker.ConsecFailures()
	if err := breaker.Call(func() error {
		addr, err := t.resolver(nodeID)
		if err != nil {
			return err
		}
		conn, err := t.rpcContext.GRPCDial(addr.String(), grpc.WithBlock())
		if err != nil {
			return err
		}
		client := NewMultiRaftClient(conn)
		ctx, cancel := context.WithCancel(context.TODO())
		defer cancel()
		stream, err := client.RaftMessage(ctx)
		if err != nil {
			return err
		}
		if successes == 0 || consecFailures > 0 {
			log.Infof(context.TODO(), "raft transport stream to node %d established", nodeID)
		}
		return t.processQueue(nodeID, ch, stats, stream)
	}, 0); err != nil {
		if consecFailures == 0 {
			log.Warningf(context.TODO(), "raft transport stream to node %d failed: %s", nodeID, err)
		}
		return
	}
}

// startBrokerClient starts an individual broker client.
// It takes as input the broker information.
func (p *Publisher) startBrokerClient(broker brokerInfo) {
	var opts []grpc.DialOption
	opts = append(opts, grpc.WithInsecure(), grpc.WithBlock())

	conn, err := grpc.Dial(broker.addr, opts...)
	if err != nil {
		fmt.Printf("Error while connecting to server: %v\n", err)
		return
	}
	defer conn.Close()

	client := pb.NewPubBrokerClient(conn)
	ch := p.addChannel(broker.id)

	for {
		pub := <-ch
		pub.MAC = common.CreatePublicationMAC(&pub, p.brokers[broker.id].key)

		// Handle publish request and response
		resp, err := client.Publish(context.Background(), &pub)
		if err != nil {
			fmt.Printf("Error publishing to %v, %v\n", broker.id, err)
			p.statusCh <- -1
			continue
		}

		p.statusCh <- resp.Status
	}
}

// DialTablet creates and initializes gRPCQueryClient.
func DialTablet(tablet *topodatapb.Tablet, timeout time.Duration) (tabletconn.TabletConn, error) {
	// create the RPC client
	addr := ""
	if grpcPort, ok := tablet.PortMap["grpc"]; ok {
		addr = netutil.JoinHostPort(tablet.Hostname, grpcPort)
	} else {
		addr = tablet.Hostname
	}
	opt, err := grpcutils.ClientSecureDialOption(*cert, *key, *ca, *name)
	if err != nil {
		return nil, err
	}
	opts := []grpc.DialOption{opt}
	if timeout > 0 {
		opts = append(opts, grpc.WithBlock(), grpc.WithTimeout(timeout))
	}
	cc, err := grpc.Dial(addr, opts...)
	if err != nil {
		return nil, err
	}
	c := queryservicepb.NewQueryClient(cc)

	result := &gRPCQueryClient{
		tablet: tablet,
		cc:     cc,
		c:      c,
	}

	return result, nil
}

// DialTablet creates and initializes gRPCQueryClient.
func DialTablet(ctx context.Context, tablet *topodatapb.Tablet, timeout time.Duration) (tabletconn.TabletConn, error) {
	// create the RPC client
	addr := netutil.JoinHostPort(tablet.Hostname, tablet.PortMap["grpc"])
	opt, err := grpcutils.ClientSecureDialOption(*cert, *key, *ca, *name)
	if err != nil {
		return nil, err
	}
	cc, err := grpc.Dial(addr, opt, grpc.WithBlock(), grpc.WithTimeout(timeout))
	if err != nil {
		return nil, err
	}
	c := queryservicepb.NewQueryClient(cc)

	result := &gRPCQueryClient{
		tablet: tablet,
		cc:     cc,
		c:      c,
		target: &querypb.Target{
			Keyspace:   tablet.Keyspace,
			Shard:      tablet.Shard,
			TabletType: tablet.Type,
		},
	}

	return result, nil
}

//invoke the EmptyCall RPC
func invokeEmptyCall(address string, dialOptions []grpc.DialOption) (*testpb.Empty, error) {

	//add DialOptions
	dialOptions = append(dialOptions, grpc.WithBlock())
	dialOptions = append(dialOptions, grpc.WithTimeout(timeout))
	//create GRPC client conn
	clientConn, err := grpc.Dial(address, dialOptions...)
	if err != nil {
		return nil, err
	}
	defer clientConn.Close()

	//create GRPC client
	client := testpb.NewTestServiceClient(clientConn)

	ctx := context.Background()
	ctx, cancel := context.WithTimeout(ctx, timeout)
	defer cancel()

	//invoke service
	empty, err := client.EmptyCall(ctx, new(testpb.Empty))
	if err != nil {
		return nil, err
	}

	return empty, nil
}

// GetBroadcastClient creates a simple instance of the BroadcastClient interface
func GetBroadcastClient() (BroadcastClient, error) {
	var orderer string
	if viper.GetBool("peer.committer.enabled") {
		orderer = viper.GetString("peer.committer.ledger.orderer")
	}

	if orderer == "" {
		return nil, fmt.Errorf("Can't get orderer address")
	}

	var opts []grpc.DialOption
	opts = append(opts, grpc.WithInsecure())
	opts = append(opts, grpc.WithTimeout(3*time.Second))
	opts = append(opts, grpc.WithBlock())

	conn, err := grpc.Dial(orderer, opts...)
	if err != nil {
		return nil, fmt.Errorf("Error connecting to %s due to %s", orderer, err)
	}
	client, err := ab.NewAtomicBroadcastClient(conn).Broadcast(context.TODO())
	if err != nil {
		conn.Close()
		return nil, fmt.Errorf("Error connecting to %s due to %s", orderer, err)
	}

	return &broadcastClient{conn: conn, client: client}, nil
}

// DialTablet creates and initializes gRPCQueryClient.
func DialTablet(ctx context.Context, endPoint *pbt.EndPoint, keyspace, shard string, timeout time.Duration) (tabletconn.TabletConn, error) {
	// create the RPC client
	addr := netutil.JoinHostPort(endPoint.Host, endPoint.PortMap["grpc"])
	cc, err := grpc.Dial(addr, grpc.WithBlock(), grpc.WithTimeout(timeout))
	if err != nil {
		return nil, err
	}
	c := pbs.NewQueryClient(cc)

	result := &gRPCQueryClient{
		endPoint: endPoint,
		cc:       cc,
		c:        c,
	}
	if keyspace != "" || shard != "" {
		gsir, err := c.GetSessionId(ctx, &pb.GetSessionIdRequest{
			Keyspace: keyspace,
			Shard:    shard,
		})
		if err != nil {
			cc.Close()
			return nil, err
		}
		if gsir.Error != nil {
			cc.Close()
			return nil, tabletErrorFromRPCError(gsir.Error)
		}
		result.sessionID = gsir.SessionId
	}

	return result, nil
}

// getClient returns a connection to the Suite containerd
func (cs *ContainerdSuite) getClient(socket string) error {
	// Parse proto://address form addresses.
	bindParts := strings.SplitN(socket, "://", 2)
	if len(bindParts) != 2 {
		return fmt.Errorf("bad bind address format %s, expected proto://address", socket)
	}

	// reset the logger for grpc to log to dev/null so that it does not mess with our stdio
	grpclog.SetLogger(log.New(ioutil.Discard, "", log.LstdFlags))
	dialOpts := []grpc.DialOption{grpc.WithInsecure()}
	dialOpts = append(dialOpts,
		grpc.WithDialer(func(addr string, timeout time.Duration) (net.Conn, error) {
			return net.DialTimeout(bindParts[0], bindParts[1], timeout)
		}),
		grpc.WithBlock(),
		grpc.WithTimeout(5*time.Second),
	)
	conn, err := grpc.Dial(socket, dialOpts...)
	if err != nil {
		return err
	}
	healthClient := grpc_health_v1.NewHealthClient(conn)
	if _, err := healthClient.Check(context.Background(), &grpc_health_v1.HealthCheckRequest{}); err != nil {
		return err
	}
	cs.grpcClient = types.NewAPIClient(conn)

	return nil
}

//newEventsClientConnectionWithAddress Returns a new grpc.ClientConn to the configured local PEER.
func newEventsClientConnectionWithAddress(peerAddress string) (*grpc.ClientConn, error) {
	var opts []grpc.DialOption
	if peer.TLSEnabled() {
		var sn string
		if viper.GetString("peer.tls.serverhostoverride") != "" {
			sn = viper.GetString("peer.tls.serverhostoverride")
		}
		var creds credentials.TransportAuthenticator
		if viper.GetString("peer.tls.cert.file") != "" {
			var err error
			creds, err = credentials.NewClientTLSFromFile(viper.GetString("peer.tls.cert.file"), 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))
	}
	opts = append(opts, grpc.WithTimeout(defaultTimeout))
	opts = append(opts, grpc.WithBlock())
	opts = append(opts, grpc.WithInsecure())

	return grpc.Dial(peerAddress, opts...)
}

func newPeerClientConnection() (*grpc.ClientConn, error) {
	var opts []grpc.DialOption
	if viper.GetBool("peer.tls.enabled") {
		var sn string
		if viper.GetString("peer.tls.serverhostoverride") != "" {
			sn = viper.GetString("peer.tls.serverhostoverride")
		}
		var creds credentials.TransportAuthenticator
		if viper.GetString("peer.tls.cert.file") != "" {
			var err error
			creds, err = credentials.NewClientTLSFromFile(viper.GetString("peer.tls.cert.file"), 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))
	}
	opts = append(opts, grpc.WithTimeout(1*time.Second))
	opts = append(opts, grpc.WithBlock())
	opts = append(opts, grpc.WithInsecure())
	conn, err := grpc.Dial(getPeerAddress(), opts...)
	if err != nil {
		return nil, err
	}
	return conn, err
}

// start dials the remote addr and commences gossip once connected. Upon exit,
// the client is sent on the disconnected channel. This method starts client
// processing in a goroutine and returns immediately.
func (c *client) start(g *Gossip, disconnected chan *client, ctx *rpc.Context, stopper *stop.Stopper) {
	stopper.RunWorker(func() {
		defer func() {
			disconnected <- c
		}()

		conn, err := ctx.GRPCDial(c.addr.String(), grpc.WithBlock())
		if err != nil {
			log.Errorf("failed to dial: %v", err)
			return
		}

		// Start gossiping.
		if err := c.gossip(g, NewGossipClient(conn), stopper); err != nil {
			if !grpcutil.IsClosedConnection(err) {
				g.mu.Lock()
				peerID := c.peerID
				g.mu.Unlock()
				if peerID != 0 {
					log.Infof("closing client to node %d (%s): %s", peerID, c.addr, err)
				} else {
					log.Infof("closing client to %s: %s", c.addr, err)
				}
			}
		}
	})
}

// RegisterGRPCGateway starts the gateway (i.e. reverse proxy) that proxies
// HTTP requests to the appropriate gRPC endpoints.
func (s *adminServer) RegisterGRPCGateway(serverCtx *Context) error {
	// Setup HTTP<->gRPC handlers.
	var opts []grpc.DialOption
	if serverCtx.Insecure {
		opts = append(opts, grpc.WithInsecure())
	} else {
		tlsConfig, err := serverCtx.GetClientTLSConfig()
		if err != nil {
			return err
		}
		opts = append(
			opts,
			// TODO(tamird): remove this timeout. It is currently necessary because
			// GRPC will not actually bail on a bad certificate error - it will just
			// retry indefinitely. See https://github.com/grpc/grpc-go/issues/622.
			grpc.WithTimeout(time.Second),
			grpc.WithBlock(),
			grpc.WithTransportCredentials(credentials.NewTLS(tlsConfig)),
		)
	}
	if err := RegisterAdminHandlerFromEndpoint(s.gwCtx, s.gwMux, serverCtx.Addr, opts); err != nil {
		return util.Errorf("error constructing grpc-gateway: %s. are your certificates valid?", err)
	}

	// Pass all requests for gRPC-based API endpoints to the gateway mux.
	s.ServeMux.Handle(apiEndpoint, s.gwMux)
	return nil
}