grpchan

README

gRPC Channels

This repo provides an abstraction for an RPC connection: the Channel. Implementations of Channel can provide alternate transports -- different from the standard HTTP/2-based transport provided by the google.golang.org/grpc package.

This can be useful for providing new transports, such as HTTP 1.1, web sockets, or (significantly) in-process channels for testing.

This repo also contains two such alternate transports: an HTTP 1.1 implementation of gRPC (which supports all stream kinds other than full-duplex bidi streams) and an in-process transport (which allows a process to dispatch handlers implemented in the same program without needing serialize and de-serialize messages over the loopback network interface).

In order to use channels with your proto-defined gRPC services, you need to use a protoc plugin included in this repo: protoc-gen-grpchan.

go install github.com/fullstorydev/grpchan/cmd/protoc-gen-grpchan

You use the plugin via a --grpchan_out parameter to protoc. Specify the same output directory to this parameter as you supply to --go_out. The plugin will then generate *.pb.grpchan.go files, alongside the *.pb.go files. These additional files contain additional methods that let you use the proto-defined service methods with alternate transports.

//go:generate protoc --go_out=plugins=grpc:. --grpchan_out=. my.proto

Documentation

Overview

Package grpchan provides an abstraction for a gRPC transport, called a Channel. The channel is more general than the concrete *grpc.ClientConn and *grpc.Server types provided by gRPC. It allows gRPC over alternate substrates and includes sub-packages that provide two such alternatives: in-process and HTTP 1.1.

The key types in this package are interfaces that are helpful abstractions and expose the same key operations as *grpc.ClientConn and *grpc.Server. They are grpchan.Channel and grpchan.ServiceRegistry, respectively. To really make use of these interfaces, you also need to use the grpchan plugin for protoc. It will generate methods for each RPC service that allow you to create client stubs and register server handlers with these interfaces. (The code generated by the standard Go plugin for protoc does not suffice since it requires the concrete types *grpc.ClientConn and *grpc.Server.)

Protoc Plugin

To use the protoc plugin, you need to first build it and make sure its location is in your PATH.

go install github.com/fullstorydev/grpchan/cmd/protoc-gen-grpchan
# If necessary, make sure its location is on your path like so:
#   export PATH=$PATH:$GOPATH/bin

When you invoke protoc, include a --grpchan_out parameter that indicates the same output directory as used for your --go_out parameter. Alongside the *.pb.go files generated, the grpchan plugin will also create *.pb.grpchan.go files.

The normal Go plugin (when emitted gRPC code) creates client stub factory and server registration functions for each RPC service defined in the proto source files compiled.

// A function with this signature is generated, for creating a client
// stub that uses the given connection to issue requests.
func New<ServerName>Client(cc *grpc.ClientConn) <ServerName>Client {
    return &<serverName>Client{cc}
}

// And a function with this signature is generated, for registering
// server handlers with the given server.
func Register<ServerName>Server(s *grpc.Server, srv <ServerName>Server) {
    s.RegisterService(&_<ServerName>_serviceDesc, srv)
}

The grpchan plugins produces similar named methods that accept interfaces:

func New<ServerName>ChannelClient(ch grpchan.Channel) <ServerName>Client {
    return &<serverName>ChannelClient{ch}
}

func RegisterHandler<ServerName>(sr grpchan.ServiceRegistry, srv <ServerName>Server) {
    s.RegisterService(&_<ServerName>_serviceDesc, srv)
}

A new transport can then be implemented by just implementing these two interfaces, grpchan.Channel for the client side and grpchan.ServiceRegistry for the server side.

These alternate methods also work just fine with *grpc.ClientConn and *grpc.Server as they implement the necessary interfaces.

Client-Side Channels

The client-side implementation of a transport is done with just the two methods in the Channel interface: one for unary RPCs and the other for streaming RPCs.

Note that when a unary interceptor is invoked for an RPC on a channel that is *not* a *grpc.ClientConn, the parameter of that type will be nil.

Not all client call options will make sense for all transports. This repo chooses to ignore call options that do not apply (as opposed to failing the RPC or panic'ing). However, several call options are likely important to support: those for accessing header and trailer metadata. The peer, per-RPC credentials, and message size limits are other options that are reasonably straight-forward to apply to other transports. But the other options (dealing with on-the-wire encoding, compression, etc) are less likely to be meaningful.

Server-Side Service Registries

The server-side implementation of a transport must be able to invoke method and stream handlers for a given service implementation. This is done by implementing the ServiceRegistry interface. When a service is registered, a service description is provided that includes access to method and stream handlers. When the transport receives requests for RPC operations, it in turn invokes these handlers. For streaming operations, it must also supply a grpc.ServerStream implementation, for exchanging messages on the stream.

Note that the server stream's context will need a custom implementation of the grpc.ServerTransportStream in it, too. Sadly, this interface is just different enough from grpc.ServerStream that they cannot be implemented by the same type. This is particularly necessary for unary calls since this is how a unary handler dictates what headers and trailers to send back to the client.

Index

• func InterceptServer(svcDesc *grpc.ServiceDesc, unaryInt grpc.UnaryServerInterceptor, ...) *grpc.ServiceDesc
• type Channel
  • func InterceptChannel(ch Channel, unaryInt grpc.UnaryClientInterceptor, ...) Channel
• type HandlerMap
  • func (r HandlerMap) ForEach(fn func(desc *grpc.ServiceDesc, svr interface{}))
  • func (r HandlerMap) QueryService(name string) (*grpc.ServiceDesc, interface{})
  • func (r HandlerMap) RegisterService(desc *grpc.ServiceDesc, h interface{})
• type ServiceRegistry
  • func WithInterceptor(reg ServiceRegistry, unaryInt grpc.UnaryServerInterceptor, ...) ServiceRegistry

Functions

func InterceptServer

func InterceptServer(svcDesc *grpc.ServiceDesc, unaryInt grpc.UnaryServerInterceptor, streamInt grpc.StreamServerInterceptor) *grpc.ServiceDesc

InterceptServer returns a new service description that will intercepts RPCs with the given interceptors. If both given interceptors are nil, returns svcDesc.

Types

type Channel

type Channel interface {
	// InvokeUnary executes a unary RPC, sending the given req message and populating
	// the given resp with the server's reply.
	Invoke(ctx context.Context, methodName string, req, resp interface{}, opts ...grpc.CallOption) error

	// InvokeStream executes a streaming RPC.
	NewStream(ctx context.Context, desc *grpc.StreamDesc, methodName string, opts ...grpc.CallOption) (grpc.ClientStream, error)
}

Channel is an abstraction of a GRPC transport. With corresponding generated code, it can provide an alternate transport to the standard HTTP/2-based one. For example, a Channel implementation could instead provide an HTTP 1.1-based transport, or an in-process transport.

func InterceptChannel

func InterceptChannel(ch Channel, unaryInt grpc.UnaryClientInterceptor, streamInt grpc.StreamClientInterceptor) Channel

InterceptChannel returns a new channel that intercepts RPCs with the given interceptors. If both given interceptors are nil, returns ch.

type HandlerMap

type HandlerMap map[string]service

HandlerMap is used to accumulate service handlers into a map. The handlers can be registered once in the map, and then re-used to configure multiple servers that should expose the same handlers. HandlerMap can also be used as the internal store of registered handlers for a server implementation.

func (HandlerMap) ForEach

func (r HandlerMap) ForEach(fn func(desc *grpc.ServiceDesc, svr interface{}))

ForEach calls the given function for each registered handler. The function is provided the service description, and the handler. This can be used to contribute all registered handlers to a server and means that applications can easily expose the same services and handlers via multiple channels after registering the handlers once, with the map:

// Register all handlers once with the map:
reg := channel.HandlerMap{}
// (these registration functions are generated)
foo.RegisterHandlerFooBar(newFooBarImpl())
fu.RegisterHandlerFuBaz(newFuBazImpl())

// Now we can re-use these handlers for multiple channels:
//   Normal gRPC
svr := grpc.NewServer()
reg.ForEach(svr.RegisterService)
//   In-process
ipch := &inprocgrpc.Channel{}
reg.ForEach(ipch.RegisterService)
//   And HTTP 1.1
httpgrpc.HandleServices(http.HandleFunc, "/rpc/", reg, nil, nil)

func (HandlerMap) QueryService

func (r HandlerMap) QueryService(name string) (*grpc.ServiceDesc, interface{})

QueryService returns the service descriptor and handler for the named service. If no handler has been registered for the named service, then nil, nil is returned.

func (HandlerMap) RegisterService

func (r HandlerMap) RegisterService(desc *grpc.ServiceDesc, h interface{})

RegisterService registers the given handler to be used for the given service. Only a single handler can be registered for a given service. And services are identified by their fully-qualified name (e.g. "package.name.Service").

type ServiceRegistry

type ServiceRegistry interface {
	// RegisterService registers the given handler to be used for the given
	// service. Only a single handler can be registered for a given service. And
	// services are identified by their fully-qualified name (e.g.
	// "package.name.Service"). Attempting to register the same service more
	// than once is an error that can panic.
	RegisterService(desc *grpc.ServiceDesc, srv interface{})
}

ServiceRegistry accumulates service definitions. Servers typically have this interface for accumulating the services they expose.

func WithInterceptor

func WithInterceptor(reg ServiceRegistry, unaryInt grpc.UnaryServerInterceptor, streamInt grpc.StreamServerInterceptor) ServiceRegistry

WithInterceptor returns a view of the given ServiceRegistry that will automatically apply the given interceptors to all registered services.