How to use Close method of grpc Package

Best K6 code snippet using grpc.Close

forwarder_grpc.pb.go

Source:forwarder_grpc.pb.go

...26//27// For semantics around ctx use and closing/ending streaming RPCs, please refer to https://pkg.go.dev/google.golang.org/grpc/?tab=doc#ClientConn.NewStream.28type ForwarderClient interface {29	Open(ctx context.Context, in *OpenRequest, opts ...grpc.CallOption) (*OpenResponse, error)30	Close(ctx context.Context, in *CloseRequest, opts ...grpc.CallOption) (*CloseResponse, error)31	ListByRemote(ctx context.Context, in *ListByRemoteRequest, opts ...grpc.CallOption) (*ListByRemoteResponse, error)32}33type forwarderClient struct {34	cc grpc.ClientConnInterface35}36func NewForwarderClient(cc grpc.ClientConnInterface) ForwarderClient {37	return &forwarderClient{cc}38}39func (c *forwarderClient) Open(ctx context.Context, in *OpenRequest, opts ...grpc.CallOption) (*OpenResponse, error) {40	out := new(OpenResponse)41	err := c.cc.Invoke(ctx, "/pb.Forwarder/Open", in, out, opts...)42	if err != nil {43		return nil, err44	}45	return out, nil46}47func (c *forwarderClient) Close(ctx context.Context, in *CloseRequest, opts ...grpc.CallOption) (*CloseResponse, error) {48	out := new(CloseResponse)49	err := c.cc.Invoke(ctx, "/pb.Forwarder/Close", in, out, opts...)50	if err != nil {51		return nil, err52	}53	return out, nil54}55func (c *forwarderClient) ListByRemote(ctx context.Context, in *ListByRemoteRequest, opts ...grpc.CallOption) (*ListByRemoteResponse, error) {56	out := new(ListByRemoteResponse)57	err := c.cc.Invoke(ctx, "/pb.Forwarder/ListByRemote", in, out, opts...)58	if err != nil {59		return nil, err60	}61	return out, nil62}63// ForwarderServer is the server API for Forwarder service.64// All implementations must embed UnimplementedForwarderServer65// for forward compatibility66type ForwarderServer interface {67	Open(context.Context, *OpenRequest) (*OpenResponse, error)68	Close(context.Context, *CloseRequest) (*CloseResponse, error)69	ListByRemote(context.Context, *ListByRemoteRequest) (*ListByRemoteResponse, error)70	mustEmbedUnimplementedForwarderServer()71}72// UnimplementedForwarderServer must be embedded to have forward compatible implementations.73type UnimplementedForwarderServer struct {74}75func (*UnimplementedForwarderServer) Open(context.Context, *OpenRequest) (*OpenResponse, error) {76	return nil, status.Errorf(codes.Unimplemented, "method Open not implemented")77}78func (*UnimplementedForwarderServer) Close(context.Context, *CloseRequest) (*CloseResponse, error) {79	return nil, status.Errorf(codes.Unimplemented, "method Close not implemented")80}81func (*UnimplementedForwarderServer) ListByRemote(context.Context, *ListByRemoteRequest) (*ListByRemoteResponse, error) {82	return nil, status.Errorf(codes.Unimplemented, "method ListByRemote not implemented")83}84func (*UnimplementedForwarderServer) mustEmbedUnimplementedForwarderServer() {}85func RegisterForwarderServer(s *grpc.Server, srv ForwarderServer) {86	s.RegisterService(&_Forwarder_serviceDesc, srv)87}88func _Forwarder_Open_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {89	in := new(OpenRequest)90	if err := dec(in); err != nil {91		return nil, err92	}93	if interceptor == nil {94		return srv.(ForwarderServer).Open(ctx, in)95	}96	info := &grpc.UnaryServerInfo{97		Server:     srv,98		FullMethod: "/pb.Forwarder/Open",99	}100	handler := func(ctx context.Context, req interface{}) (interface{}, error) {101		return srv.(ForwarderServer).Open(ctx, req.(*OpenRequest))102	}103	return interceptor(ctx, in, info, handler)104}105func _Forwarder_Close_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {106	in := new(CloseRequest)107	if err := dec(in); err != nil {108		return nil, err109	}110	if interceptor == nil {111		return srv.(ForwarderServer).Close(ctx, in)112	}113	info := &grpc.UnaryServerInfo{114		Server:     srv,115		FullMethod: "/pb.Forwarder/Close",116	}117	handler := func(ctx context.Context, req interface{}) (interface{}, error) {118		return srv.(ForwarderServer).Close(ctx, req.(*CloseRequest))119	}120	return interceptor(ctx, in, info, handler)121}122func _Forwarder_ListByRemote_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {123	in := new(ListByRemoteRequest)124	if err := dec(in); err != nil {125		return nil, err126	}127	if interceptor == nil {128		return srv.(ForwarderServer).ListByRemote(ctx, in)129	}130	info := &grpc.UnaryServerInfo{131		Server:     srv,132		FullMethod: "/pb.Forwarder/ListByRemote",133	}134	handler := func(ctx context.Context, req interface{}) (interface{}, error) {135		return srv.(ForwarderServer).ListByRemote(ctx, req.(*ListByRemoteRequest))136	}137	return interceptor(ctx, in, info, handler)138}139var _Forwarder_serviceDesc = grpc.ServiceDesc{140	ServiceName: "pb.Forwarder",141	HandlerType: (*ForwarderServer)(nil),142	Methods: []grpc.MethodDesc{143		{144			MethodName: "Open",145			Handler:    _Forwarder_Open_Handler,146		},147		{148			MethodName: "Close",149			Handler:    _Forwarder_Close_Handler,150		},151		{152			MethodName: "ListByRemote",153			Handler:    _Forwarder_ListByRemote_Handler,154		},155	},156	Streams:  []grpc.StreamDesc{},157	Metadata: "forwarder.proto",158}...

client.go

Source:client.go

...42	grpcInitialWindowSize     = 1 << 3043	grpcInitialConnWindowSize = 1 << 3044)45// Client is a client that sends RPC.46// It should not be used after calling Close().47type Client interface {48	// Close should release all data.49	Close() error50	// SendRequest sends Request.51	SendRequest(ctx context.Context, addr string, req *tikvrpc.Request, timeout time.Duration) (*tikvrpc.Response, error)52}53type connArray struct {54	// The target host.55	target string56	index uint3257	v     []*grpc.ClientConn58	done  chan struct{}59}60func newConnArray(maxSize uint, addr string, idleNotify *uint32) (*connArray, error) {61	a := &connArray{62		index: 0,63		v:     make([]*grpc.ClientConn, maxSize),64		done:  make(chan struct{}),65	}66	if err := a.Init(addr, idleNotify); err != nil {67		return nil, err68	}69	return a, nil70}71func (a *connArray) Init(addr string, idleNotify *uint32) error {72	a.target = addr73	opt := grpc.WithInsecure()74	var (75		unaryInterceptor  grpc.UnaryClientInterceptor76		streamInterceptor grpc.StreamClientInterceptor77	)78	for i := range a.v {79		ctx, cancel := context.WithTimeout(context.Background(), dialTimeout)80		conn, err := grpc.DialContext(81			ctx,82			addr,83			opt,84			grpc.WithInitialWindowSize(grpcInitialWindowSize),85			grpc.WithInitialConnWindowSize(grpcInitialConnWindowSize),86			grpc.WithUnaryInterceptor(unaryInterceptor),87			grpc.WithStreamInterceptor(streamInterceptor),88			grpc.WithDefaultCallOptions(grpc.MaxCallRecvMsgSize(MaxRecvMsgSize)),89			grpc.WithConnectParams(grpc.ConnectParams{90				Backoff: backoff.Config{91					BaseDelay:  100 * time.Millisecond, // Default was 1s.92					Multiplier: 1.6,                    // Default93					Jitter:     0.2,                    // Default94					MaxDelay:   3 * time.Second,        // Default was 120s.95				},96				MinConnectTimeout: dialTimeout,97			}),98			grpc.WithKeepaliveParams(keepalive.ClientParameters{99				Time:                time.Duration(10) * time.Second,100				Timeout:             time.Duration(3) * time.Second,101				PermitWithoutStream: true,102			}),103		)104		cancel()105		if err != nil {106			// Cleanup if the initialization fails.107			a.Close()108			return errors.Trace(err)109		}110		a.v[i] = conn111	}112	return nil113}114func (a *connArray) Get() *grpc.ClientConn {115	next := atomic.AddUint32(&a.index, 1) % uint32(len(a.v))116	return a.v[next]117}118func (a *connArray) Close() {119	for i, c := range a.v {120		if c != nil {121			err := c.Close()122			terror.Log(errors.Trace(err))123			a.v[i] = nil124		}125	}126	close(a.done)127}128// rpcClient is RPC client struct.129// TODO: Add flow control between RPC clients in TiDB ond RPC servers in TiKV.130// Since we use shared client connection to communicate to the same TiKV, it's possible131// that there are too many concurrent requests which overload the service of TiKV.132type rpcClient struct {133	sync.RWMutex134	conns map[string]*connArray135	idleNotify uint32136	// Periodically check whether there is any connection that is idle and then close and remove these idle connections.137	// Implement background cleanup.138	isClosed bool139}140func newRPCClient() *rpcClient {141	return &rpcClient{142		conns: make(map[string]*connArray),143	}144}145func (c *rpcClient) getConnArray(addr string) (*connArray, error) {146	c.RLock()147	if c.isClosed {148		c.RUnlock()149		return nil, errors.Errorf("rpcClient is closed")150	}151	array, ok := c.conns[addr]152	c.RUnlock()153	if !ok {154		var err error155		array, err = c.createConnArray(addr)156		if err != nil {157			return nil, err158		}159	}160	return array, nil161}162func (c *rpcClient) createConnArray(addr string) (*connArray, error) {163	c.Lock()164	defer c.Unlock()165	array, ok := c.conns[addr]166	if !ok {167		var err error168		array, err = newConnArray(4, addr, &c.idleNotify)169		if err != nil {170			return nil, err171		}172		c.conns[addr] = array173	}174	return array, nil175}176func (c *rpcClient) closeConns() {177	c.Lock()178	if !c.isClosed {179		c.isClosed = true180		// close all connections181		for _, array := range c.conns {182			array.Close()183		}184	}185	c.Unlock()186}187// SendRequest sends a Request to server and receives Response.188func (c *rpcClient) SendRequest(ctx context.Context, addr string, req *tikvrpc.Request, timeout time.Duration) (*tikvrpc.Response, error) {189	connArray, err := c.getConnArray(addr)190	if err != nil {191		return nil, errors.Trace(err)192	}193	clientConn := connArray.Get()194	client := tinykvpb.NewTinyKvClient(clientConn)195	ctx1, cancel := context.WithTimeout(ctx, timeout)196	defer cancel()197	return tikvrpc.CallRPC(ctx1, client, req)198}199func (c *rpcClient) Close() error {200	// TODO: add a unit test for SendRequest After Closed201	c.closeConns()202	return nil203}...

bidi.go

Source:bidi.go

...72	}73	if err != nil {74		return err75	}76	defer conn.Close()77	session, err := yamux.Client(conn, nil)78	if err != nil {79		return err80	}81	defer session.Close()82	// now that we have a connection, create both clients & servers83	// setup client84	yDialer.SetSession(session)85	// start grpc server in a separate goroutine. this will exit when the86	// underlying session (conn) closes and clean itself up.87	go grpcServer.Serve(session)88	// return when the conn closes so we can try reconnecting89	<-session.CloseChan()90	return nil91}92// Listen starts the server side of the yamux channel for bidi grpc.93func Listen(lis net.Listener, grpcServer *grpc.Server) (clientChan chan *grpc.ClientConn, shutdownChan chan int, err error) {94	// create channels95	clientChan = make(chan *grpc.ClientConn)96	shutdownChan = make(chan int)97	go func() {98		// start listenLoop - blocks until some error occurs99		listenLoop(lis, grpcServer, clientChan)100		close(shutdownChan)101	}()102	return clientChan, shutdownChan, nil103}104// listenLoop accepts new connections and sets up a yamux channel and grpc client on it.105func listenLoop(lis net.Listener, grpcServer *grpc.Server, clientChan chan *grpc.ClientConn) {106	for {107		// accept a new connection and set up a yamux session on it108		conn, err := lis.Accept()109		if err != nil {110			return // chan is prob shut down111		}112		// server session will be used to multiplex both clients & servers113		session, err := yamux.Server(conn, nil)114		if err != nil {115			conn.Close() // close conn and retry116			continue117		}118		// start grpc server using yamux session (which implements net.Listener)119		go grpcServer.Serve(session)120		// create new client connection and dialer for this session121		dialer := NewYamuxDialer()122		dialer.SetSession(session)123		gconn, _ := grpc.Dial("localhost:50000", grpc.WithInsecure(), grpc.WithDialer(dialer.Dial))124		go func() {125			// wait for session close and close related gconn126			<-session.CloseChan()127			gconn.Close()128		}()129		go func() {130			// close session if gconn is closed for any reason131			state := gconn.GetState()132			for {133				switch state {134				case connectivity.Shutdown:135					gconn.Close()136					session.Close()137					return138				}139				gconn.WaitForStateChange(context.Background(), gconn.GetState())140				state = gconn.GetState()141			}142		}()143		clientChan <- gconn // publish gconn144	}145}...

Using AI Code Generation

1import (2func main() {3	conn, err := grpc.Dial("localhost:50051", grpc.WithInsecure())4	if err != nil {5		log.Fatalf("did not connect: %v", err)6	}7	defer conn.Close()8	ctx, cancel := context.WithTimeout(context.Background(), time.Second)9	defer cancel()10	r, err := conn.GetState(ctx, &pb.Empty{})11	if err != nil {12		log.Fatalf("could not greet: %v", err)13	}14	fmt.Println(r.GetState())15}16import (17func main() {18	conn, err := grpc.Dial("localhost:50051", grpc.WithInsecure())19	if err != nil {20		log.Fatalf("did not connect: %v", err)21	}22	defer conn.Close()23	ctx, cancel := context.WithTimeout(context.Background(), time.Second)24	defer cancel()25	r, err := conn.GetState(ctx, &pb.Empty{})26	if err != nil {27		log.Fatalf("could not greet: %v", err)28	}29	fmt.Println(r.GetState())30}31import (32func main() {33	conn, err := grpc.Dial("localhost:50051", grpc.WithInsecure())34	if err != nil {35		log.Fatalf("did not connect: %v", err)36	}37	defer conn.Close()38	ctx, cancel := context.WithTimeout(context.Background(), time.Second)39	defer cancel()40	r, err := conn.GetState(ctx, &pb.Empty{})41	if err != nil {42		log.Fatalf("could not greet: %v", err)43	}44	fmt.Println(r.GetState())45}46import (47func main() {

Using AI Code Generation

1import (2const (3type server struct{}4func (s *server) SayHello(ctx context.Context, in *pb.HelloRequest) (*pb.HelloReply, error) {5    return &pb.HelloReply{Message: "Hello " + in.Name}, nil6}7func main() {8    lis, err := net.Listen("tcp", port)9    if err != nil {10        log.Fatalf("failed to listen: %v", err)11    }12    s := grpc.NewServer()13    pb.RegisterGreeterServer(s, &server{})14    reflection.Register(s)15    if err := s.Serve(lis); err != nil {16        log.Fatalf("failed to serve: %v", err)17    }18}19import (20const (21type server struct{}22func (s *server) SayHello(ctx context.Context, in *pb.HelloRequest) (*pb.HelloReply, error) {23    return &pb.HelloReply{Message: "Hello " + in.Name}, nil24}25func main() {26    lis, err := net.Listen("tcp", port)27    if err != nil {28        log.Fatalf("failed to listen: %v", err)29    }30    s := grpc.NewServer()31    pb.RegisterGreeterServer(s, &server{})32    reflection.Register(s)33    if err := s.Serve(lis); err != nil {34        log.Fatalf("failed to serve: %v", err)35    }36}37import (

Using AI Code Generation

1import (2func main() {3	fmt.Println("Hello I'm a client")4	opts := grpc.WithInsecure()5	if tls {6		creds, sslErr := credentials.NewClientTLSFromFile(certFile, "")7		if sslErr != nil {8			log.Fatalf("Could not load CA trust certificate: %v", sslErr)9		}10		opts = grpc.WithTransportCredentials(creds)11	}12	cc, err := grpc.Dial("localhost:50051", opts)13	if err != nil {14		log.Fatalf("Could not connect: %v", err)15	}16	defer cc.Close()17	c := NewGreetServiceClient(cc)18	doBidiStreaming(c)19}20func doUnary(c GreetServiceClient) {21	fmt.Println("Starting to do a Unary RPC...")22	req := &GreetRequest{23		Greeting: &Greeting{24		},25	}26	res, err := c.Greet(context.Background(), req)27	if err != nil {28		log.Fatalf("error while calling Greet RPC: %v", err)29	}30	log.Printf("Response from Greet: %v", res.Result)31}32func doServerStreaming(c GreetServiceClient) {33	fmt.Println("Starting to do a Server Streaming RPC...")34	req := &GreetManyTimesRequest{35		Greeting: &Greeting{36		},37	}38	resStream, err := c.GreetManyTimes(context.Background(), req)39	if err != nil {40		log.Fatalf("error while calling GreetManyTimes RPC: %v", err)41	}42	for {43		msg, err := resStream.Recv()44		if err == io.EOF {45		}

Using AI Code Generation

1import (2func main() {3	creds, err := credentials.NewClientTLSFromFile("cert.pem", "localhost")4	if err != nil {5		log.Fatalf("could not load tls cert: %s", err)6	}7	conn, err := grpc.Dial("localhost:8080", grpc.WithTransportCredentials(creds))8	if err != nil {9		log.Fatalf("did not connect: %s", err)10	}11	defer conn.Close()12	client := NewPingClient(conn)13	ctx := context.Background()14	p, ok := peer.FromContext(ctx)15	if ok {16		fmt.Println("peer from context", p.Addr)17	}18	_, err = client.Ping(ctx, &PingMessage{Greeting: "Hello"})19	if err != nil {20		log.Fatalf("error when calling Ping: %s", err)21	}22	_, err = client.Ping(ctx, &PingMessage{Greeting: "Hello"})23	if err != nil {24		log.Fatalf("error when calling Ping: %s", err)25	}26	_, err = client.Ping(ctx, &PingMessage{Greeting: "Hello"})27	if err != nil {28		log.Fatalf("error when calling Ping: %s", err)29	}30}31import (32func main() {33	creds, err := credentials.NewClientTLSFromFile("cert.pem", "localhost")

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.