Repository: grpc/grpc-go
File: xds/internal/clients/grpctransport/grpc_transport.go (Go)

Code before:
		return nil, fmt.Errorf("ServerConfig's Extensions field cannot be nil for gRPC transport")
	}
	sce, ok := sc.Extensions.(ServerConfigExtension)
	if !ok {
		return nil, fmt.Errorf("ServerConfig Extensions field is %T, but must be %T", sc.Extensions, ServerConfigExtension{})
	}
	if sce.Credentials == nil {
		return nil, fmt.Errorf("ServerConfigExtensions's Credentials field cannot be nil for gRPC transport")
	}

	// TODO: Incorporate reference count map for existing transports and
	// deduplicate transports based on the provided ServerConfig so that
	// transport channel to same server can be shared between xDS and LRS
	// client.

	// Create a new gRPC client/channel for the server with the provided
	// credentials, server URI, and a byte codec to send and receive messages.
	// Also set a static keepalive configuration that is common across gRPC
	// language implementations.
	kpCfg := grpc.WithKeepaliveParams(keepalive.ClientParameters{
		Time:    5 * time.Minute,
		Timeout: 20 * time.Second,
	})
	cc, err := grpc.NewClient(sc.ServerURI, kpCfg, grpc.WithCredentialsBundle(sce.Credentials), grpc.WithDefaultCallOptions(grpc.ForceCodec(&byteCodec{})))
	if err != nil {
		return nil, fmt.Errorf("error creating grpc client for server uri %s, %v", sc.ServerURI, err)
	}

	return &grpcTransport{cc: cc}, nil
}

type grpcTransport struct {
	cc *grpc.ClientConn
}

// NewStream creates a new gRPC stream to the server for the specified method.
func (g *grpcTransport) NewStream(ctx context.Context, method string) (clients.Stream, error) {
	s, err := g.cc.NewStream(ctx, &grpc.StreamDesc{StreamName: method, ClientStreams: true, ServerStreams: true}, method)
	if err != nil {
		return nil, err
	}
	return &stream{stream: s}, nil
}

type stream struct {
	stream grpc.ClientStream
}

// Send sends a message to the server.
func (s *stream) Send(msg []byte) error {
	return s.stream.SendMsg(msg)

Review comment (bug):
This error string could be `grpctransport: failed to create transport to server %q: %v", sc.ServerURI, err`

Code after:
		return nil, fmt.Errorf("grpctransport: Extensions is not set in ServerConfig")
	}
	sce, ok := sc.Extensions.(ServerConfigExtension)
	if !ok {
		return nil, fmt.Errorf("grpctransport: Extensions field is %T, but must be %T in ServerConfig", sc.Extensions, ServerConfigExtension{})
	}
	if sce.Credentials == nil {
		return nil, fmt.Errorf("grptransport: Credentials field is not set in ServerConfigExtension")
	}

	// TODO: Incorporate reference count map for existing transports and
	// deduplicate transports based on the provided ServerConfig so that
	// transport channel to same server can be shared between xDS and LRS
	// client.

	// Create a new gRPC client/channel for the server with the provided
	// credentials, server URI, and a byte codec to send and receive messages.
	// Also set a static keepalive configuration that is common across gRPC
	// language implementations.
	kpCfg := grpc.WithKeepaliveParams(keepalive.ClientParameters{
		Time:    5 * time.Minute,
		Timeout: 20 * time.Second,
	})
	cc, err := grpc.NewClient(sc.ServerURI, kpCfg, grpc.WithCredentialsBundle(sce.Credentials), grpc.WithDefaultCallOptions(grpc.ForceCodec(&byteCodec{})))
	if err != nil {
		return nil, fmt.Errorf("grpctransport: failed to create transport to server %q: %v", sc.ServerURI, err)
	}

	return &grpcTransport{cc: cc}, nil
}

type grpcTransport struct {
	cc *grpc.ClientConn
}

// NewStream creates a new gRPC stream to the server for the specified method.
func (g *grpcTransport) NewStream(ctx context.Context, method string) (clients.Stream, error) {
	s, err := g.cc.NewStream(ctx, &grpc.StreamDesc{ClientStreams: true, ServerStreams: true}, method)
	if err != nil {
		return nil, err
	}
	return &stream{stream: s}, nil
}

// Close closes the gRPC channel to the server.
func (g *grpcTransport) Close() error {
	return g.cc.Close()
}

type stream struct {
	stream grpc.ClientStream