remove transport draft, add grpc web and gateway support

service/command.go

@@ -12,9 +12,9 @@ var cmd = &cobra.Command{
 
 const (
 	serverAddress = "server_address"
-	
-	secure        = "secure"
-	reflect       = "reflect"
+
+	secure  = "secure"
+	reflect = "reflect"
 
 	caCert     = "ca_cert"
 	serverCert = "server_cert"

service/gateway.go

@@ -0,0 +1,29 @@
+package service
+
+import (
+	"net/http"
+
+	"github.com/grpc-ecosystem/grpc-gateway/v2/runtime"
+)
+
+var defaultGatewayOptions = []runtime.ServeMuxOption{
+	runtime.WithIncomingHeaderMatcher(func(s string) (string, bool) {
+		return s, true
+	}),
+}
+
+func (s *service) gateway(opts ...runtime.ServeMuxOption) error {
+	if !s.opts.Gateway() {
+		return nil
+	}
+	mux := runtime.NewServeMux(append(defaultGatewayOptions, opts...)...)
+	if err := s.opts.gateway(s.opts.ctx, mux, s.inproc); err != nil {
+		return err
+	}
+	if s.opts.gatewayPrefix != "" {
+		s.mux.Handle(s.opts.gatewayPrefix+"/", http.StripPrefix(s.opts.gatewayPrefix, mux))
+	} else {
+		s.mux.Handle("/", mux)
+	}
+	return nil
+}

service/options.go

@@ -7,13 +7,17 @@ import (
 	"fmt"
 	"io/ioutil"
 	"net"
+	"strings"
 
+	"github.com/grpc-ecosystem/grpc-gateway/v2/runtime"
+	"github.com/improbable-eng/grpc-web/go/grpcweb"
 	"github.com/jinzhu/gorm"
 	"go.uber.org/multierr"
 	"google.golang.org/grpc"
 
 	"go.linka.cloud/grpc/certs"
 	"go.linka.cloud/grpc/registry"
+	"go.linka.cloud/grpc/transport"
 	"go.linka.cloud/grpc/utils/addr"
 )
 
@@ -46,29 +50,51 @@ GLOBAL OPTIONS:
 	--db_path                 DB_PATH
 */
 
+type RegisterGatewayFunc func(ctx context.Context, mux *runtime.ServeMux, cc grpc.ClientConnInterface) error
+
 type Options interface {
 	Context() context.Context
 	Name() string
 	Version() string
 	Address() string
+
 	Reflection() bool
+
 	CACert() string
 	Cert() string
 	Key() string
 	TLSConfig() *tls.Config
 	Secure() bool
+
 	Registry() registry.Registry
+
 	DB() *gorm.DB
+
 	BeforeStart() []func() error
 	AfterStart() []func() error
 	BeforeStop() []func() error
 	AfterStop() []func() error
+
 	ServerOpts() []grpc.ServerOption
 	ServerInterceptors() []grpc.UnaryServerInterceptor
 	StreamServerInterceptors() []grpc.StreamServerInterceptor
+
 	ClientInterceptors() []grpc.UnaryClientInterceptor
 	StreamClientInterceptors() []grpc.StreamClientInterceptor
-	Defaults()
+
+	// TODO(adphi): CORS for http handler
+
+	GRPCWeb() bool
+	GRPCWebPrefix() string
+	GRPCWebOpts() []grpcweb.Option
+
+	Gateway() bool
+	GatewayPrefix() string
+	GatewayOpts() []runtime.ServeMuxOption
+
+	// TODO(adphi): metrics
+
+	Default()
 }
 
 func NewOptions() *options {
@@ -78,13 +104,16 @@ func NewOptions() *options {
 	}
 }
 
-func (o *options) Defaults() {
+func (o *options) Default() {
 	if o.ctx == nil {
 		o.ctx = context.Background()
 	}
 	if o.address == "" {
 		o.address = "0.0.0.0:0"
 	}
+	if o.transport == nil {
+		o.transport = &grpc.Server{}
+	}
 }
 
 type Option func(*options)
@@ -229,20 +258,60 @@ func WithSubscriberInterceptor(w ...interface{}) Option {
 	}
 }
 
-type options struct {
-	ctx        context.Context
-	name       string
-	version string
-	address    string
-	secure     bool
-	reflection bool
-	caCert     string
-	cert       string
-	key        string
-	tlsConfig  *tls.Config
-	db         *gorm.DB
+func WithGRPCWeb(b bool) Option {
+	return func(o *options) {
+		o.grpcWeb = b
+	}
+}
 
-	registry registry.Registry
+func WithGRPCWebPrefix(prefix string) Option {
+	return func(o *options) {
+		o.grpcWebPrefix = strings.TrimSuffix(prefix, "/")
+	}
+}
+
+func WithGRPCWebOpts(opts ...grpcweb.Option) Option {
+	return func(o *options) {
+		o.grpcWebOpts = opts
+	}
+}
+
+func WithGateway(fn RegisterGatewayFunc) Option {
+	return func(o *options) {
+		o.gateway = fn
+	}
+}
+
+func WithGatewayPrefix(prefix string) Option {
+	return func(o *options) {
+		o.gatewayPrefix = strings.TrimSuffix(prefix, "/")
+	}
+}
+
+func WithGatewayOpts(opts ...runtime.ServeMuxOption) Option {
+	return func(o *options) {
+		o.gatewayOpts = opts
+	}
+}
+
+type options struct {
+	ctx     context.Context
+	name    string
+	version string
+	address string
+
+	reflection bool
+
+	secure    bool
+	caCert    string
+	cert      string
+	key       string
+	tlsConfig *tls.Config
+
+	db *gorm.DB
+
+	transport transport.Transport
+	registry  registry.Registry
 
 	beforeStart []func() error
 	afterStart  []func() error
@@ -257,7 +326,15 @@ type options struct {
 	clientInterceptors       []grpc.UnaryClientInterceptor
 	streamClientInterceptors []grpc.StreamClientInterceptor
 
-	error   error
+	grpcWeb       bool
+	grpcWebOpts   []grpcweb.Option
+	grpcWebPrefix string
+
+	gateway     RegisterGatewayFunc
+	gatewayOpts []runtime.ServeMuxOption
+
+	error         error
+	gatewayPrefix string
 }
 
 func (o *options) Name() string {
@@ -344,6 +421,30 @@ func (o *options) StreamClientInterceptors() []grpc.StreamClientInterceptor {
 	return o.streamClientInterceptors
 }
 
+func (o *options) GRPCWeb() bool {
+	return o.grpcWeb
+}
+
+func (o *options) GRPCWebPrefix() string {
+	return o.grpcWebPrefix
+}
+
+func (o *options) GRPCWebOpts() []grpcweb.Option {
+	return o.grpcWebOpts
+}
+
+func (o *options) Gateway() bool {
+	return o.gateway != nil
+}
+
+func (o *options) GatewayPrefix() string {
+	return o.gatewayPrefix
+}
+
+func (o *options) GatewayOpts() []runtime.ServeMuxOption {
+	return o.gatewayOpts
+}
+
 func (o *options) parseTLSConfig() error {
 	if o.tlsConfig != nil {
 		return nil

service/register.go

@@ -0,0 +1,86 @@
+package service
+
+import (
+	"net"
+	"strings"
+	"time"
+
+	"go.linka.cloud/grpc/registry"
+	"go.linka.cloud/grpc/utils/addr"
+	"go.linka.cloud/grpc/utils/backoff"
+	net2 "go.linka.cloud/grpc/utils/net"
+)
+
+func (s *service) register() error {
+	const (
+		defaultRegisterInterval = time.Second * 30
+		defaultRegisterTTL      = time.Second * 90
+	)
+	regFunc := func(service *registry.Service) error {
+		var regErr error
+
+		for i := 0; i < 3; i++ {
+			// set the ttl
+			rOpts := []registry.RegisterOption{registry.RegisterTTL(defaultRegisterTTL)}
+			// attempt to register
+			if err := s.opts.Registry().Register(service, rOpts...); err != nil {
+				// set the error
+				regErr = err
+				// backoff then retry
+				time.Sleep(backoff.Do(i + 1))
+				continue
+			}
+			// success so nil error
+			regErr = nil
+			break
+		}
+
+		return regErr
+	}
+
+	var err error
+	var advt, host, port string
+
+	// // check the advertise address first
+	// // if it exists then use it, otherwise
+	// // use the address
+	// if len(config.Advertise) > 0 {
+	//	advt = config.Advertise
+	// } else {
+	advt = s.opts.address
+	// }
+
+	if cnt := strings.Count(advt, ":"); cnt >= 1 {
+		// ipv6 address in format [host]:port or ipv4 host:port
+		host, port, err = net.SplitHostPort(advt)
+		if err != nil {
+			return err
+		}
+	} else {
+		host = s.opts.address
+	}
+
+	addr, err := addr.Extract(host)
+	if err != nil {
+		return err
+	}
+
+	// register service
+	node := &registry.Node{
+		Id:      s.opts.name + "-" + s.id,
+		Address: net2.HostPort(addr, port),
+	}
+
+	s.regSvc = &registry.Service{
+		Name:    s.opts.name,
+		Version: s.opts.version,
+		Nodes:   []*registry.Node{node},
+	}
+
+	// register the service
+	if err := regFunc(s.regSvc); err != nil {
+		return err
+	}
+
+	return nil
+}

service/service.go

@@ -2,8 +2,10 @@ package service
 
 import (
 	"context"
+	"crypto/tls"
 	"fmt"
 	"net"
+	"net/http"
 	"os"
 	"os/signal"
 	"strings"
@@ -11,21 +13,19 @@ import (
 	"syscall"
 	"time"
 
+	"github.com/fullstorydev/grpchan/inprocgrpc"
 	"github.com/google/uuid"
 	grpcmiddleware "github.com/grpc-ecosystem/go-grpc-middleware"
 	"github.com/jinzhu/gorm"
 	"github.com/sirupsen/logrus"
+	"github.com/soheilhy/cmux"
 	"github.com/spf13/cobra"
 	"go.uber.org/multierr"
 	"google.golang.org/grpc"
-	"google.golang.org/grpc/credentials"
 	"google.golang.org/grpc/reflection"
 
 	"go.linka.cloud/grpc/registry"
 	"go.linka.cloud/grpc/registry/noop"
-	"go.linka.cloud/grpc/utils/addr"
-	"go.linka.cloud/grpc/utils/backoff"
-	net2 "go.linka.cloud/grpc/utils/net"
 )
 
 type Service interface {
@@ -45,14 +45,18 @@ func New(opts ...Option) (Service, error) {
 }
 
 type service struct {
-	cmd     *cobra.Command
-	opts    *options
-	cancel  context.CancelFunc
+	cmd    *cobra.Command
+	opts   *options
+	cancel context.CancelFunc
+
 	server  *grpc.Server
-	list    net.Listener
 	mu      sync.Mutex
 	running bool
 
+	mux    *http.ServeMux
+	// inproc Channel is used to serve grpc gateway
+	inproc *inprocgrpc.Channel
+
 	id     string
 	regSvc *registry.Service
 	closed chan struct{}
@@ -63,9 +67,11 @@ func newService(opts ...Option) (*service, error) {
 		return nil, err
 	}
 	s := &service{
-		opts: parseFlags(NewOptions()),
-		cmd:  cmd,
-		id:   uuid.New().String(),
+		opts:   parseFlags(NewOptions()),
+		cmd:    cmd,
+		id:     uuid.New().String(),
+		mux:    http.NewServeMux(),
+		inproc: &inprocgrpc.Channel{},
 	}
 	s.mu.Lock()
 	defer s.mu.Unlock()
@@ -87,11 +93,7 @@ func newService(opts ...Option) (*service, error) {
 	if s.opts.registry == nil {
 		s.opts.registry = noop.New()
 	}
-	var err error
-	s.list, err = net.Listen("tcp", s.opts.address)
-	if err != nil {
-		return nil, err
-	}
+
 	if err := s.opts.parseTLSConfig(); err != nil {
 		return nil, err
 	}
@@ -102,18 +104,24 @@ func newService(opts ...Option) (*service, error) {
 		}
 		return s.run()
 	}
+	ui := grpcmiddleware.ChainUnaryServer(s.opts.serverInterceptors...)
+	s.inproc = s.inproc.WithServerUnaryInterceptor(ui)
+
+	si := grpcmiddleware.ChainStreamServer(/*TODO(adphi): add to options*/)
+	s.inproc = s.inproc.WithServerStreamInterceptor(si)
+
 	gopts := []grpc.ServerOption{
-		grpc.UnaryInterceptor(
-			grpcmiddleware.ChainUnaryServer(s.opts.serverInterceptors...),
-		),
-	}
-	if s.opts.tlsConfig != nil {
-		gopts = append(gopts, grpc.Creds(credentials.NewTLS(s.opts.tlsConfig)))
+		grpc.StreamInterceptor(si),
+		grpc.UnaryInterceptor(ui),
 	}
 	s.server = grpc.NewServer(append(gopts, s.opts.serverOpts...)...)
 	if s.opts.reflection {
 		reflection.Register(s.server)
 	}
+	if err := s.gateway(s.opts.gatewayOpts...); err != nil {
+		return nil, err
+	}
+	// we do not configure grpc web here as the grpc handlers are not yet registered
 	return s, nil
 }
 
@@ -133,99 +141,69 @@ func (s *service) Cmd() *cobra.Command {
 	return s.cmd
 }
 
-func (s *service) register() error {
-	const (
-		defaultRegisterInterval = time.Second * 30
-		defaultRegisterTTL      = time.Second * 90
-	)
-	regFunc := func(service *registry.Service) error {
-		var regErr error
-
-		for i := 0; i < 3; i++ {
-			// set the ttl
-			rOpts := []registry.RegisterOption{registry.RegisterTTL(defaultRegisterTTL)}
-			// attempt to register
-			if err := s.opts.Registry().Register(service, rOpts...); err != nil {
-				// set the error
-				regErr = err
-				// backoff then retry
-				time.Sleep(backoff.Do(i + 1))
-				continue
-			}
-			// success so nil error
-			regErr = nil
-			break
-		}
-
-		return regErr
-	}
-
-	var err error
-	var advt, host, port string
-
-	//// check the advertise address first
-	//// if it exists then use it, otherwise
-	//// use the address
-	//if len(config.Advertise) > 0 {
-	//	advt = config.Advertise
-	//} else {
-		advt = s.opts.address
-	//}
-
-	if cnt := strings.Count(advt, ":"); cnt >= 1 {
-		// ipv6 address in format [host]:port or ipv4 host:port
-		host, port, err = net.SplitHostPort(advt)
-		if err != nil {
-			return err
-		}
-	} else {
-		host = s.opts.address
-	}
-
-	addr, err := addr.Extract(host)
-	if err != nil {
-		return err
-	}
-
-	// register service
-	node := &registry.Node{
-		Id:      s.opts.name + "-" + s.id,
-		Address: net2.HostPort(addr, port),
-	}
-
-	s.regSvc = &registry.Service{
-		Name: s.opts.name,
-		Version:   s.opts.version,
-		Nodes: []*registry.Node{node},
-	}
-
-	// register the service
-	if err := regFunc(s.regSvc); err != nil {
-		return err
-	}
-
-	return nil
-}
-
 func (s *service) run() error {
 	s.mu.Lock()
 	s.closed = make(chan struct{})
+
+	// configure grpc web now that we are ready to go
+	if err := s.grpcWeb(s.opts.grpcWebOpts...); err != nil {
+		return err
+	}
+
+	lis, err := net.Listen("tcp", s.opts.address)
+	if err != nil {
+		return err
+	}
+	if s.opts.tlsConfig != nil {
+		lis = tls.NewListener(lis, s.opts.tlsConfig)
+	}
+
+	s.opts.address = lis.Addr().String()
+
+	mux := cmux.New(lis)
+	mux.SetReadTimeout(5 * time.Second)
+
+	gLis := mux.MatchWithWriters(cmux.HTTP2MatchHeaderFieldSendSettings("content-type", "application/grpc"))
+	hList := mux.Match(cmux.Any())
+
 	for i := range s.opts.beforeStart {
 		if err := s.opts.beforeStart[i](); err != nil {
 			s.mu.Unlock()
 			return err
 		}
 	}
-	s.opts.address = s.list.Addr().String()
 
 	if err := s.register(); err != nil {
 		return err
 	}
 	s.running = true
 
-	errs := make(chan error)
+	errs := make(chan error, 3)
+
+	hServer := &http.Server{
+		Handler: s.mux,
+	}
+	if s.opts.Gateway() || s.opts.grpcWeb {
+		go func() {
+			errs <- hServer.Serve(hList)
+			hServer.Shutdown(s.opts.ctx)
+		}()
+	}
 	go func() {
-		errs <- s.server.Serve(s.list)
+		errs <- s.server.Serve(gLis)
+	}()
+
+	go func() {
+		if err := mux.Serve(); err != nil {
+			// TODO(adphi): find more elegant solution
+			if ignoreMuxError(err) {
+				errs <- nil
+				return
+			}
+			errs <- err
+			return
+		}
+		errs <- nil
 	}()
 	for i := range s.opts.afterStart {
 		if err := s.opts.afterStart[i](); err != nil {
@@ -242,7 +220,7 @@ func (s *service) run() error {
 		logrus.Warnf("received %v", sig)
 		return s.Close()
 	case err := <-errs:
-		if err != nil{
+		if err != nil {
 			logrus.Error(err)
 			return err
 		}
@@ -297,6 +275,7 @@ func (s *service) Stop() error {
 
 func (s *service) RegisterService(desc *grpc.ServiceDesc, impl interface{}) {
 	s.server.RegisterService(desc, impl)
+	s.inproc.RegisterService(desc, impl)
 }
 
 func (s *service) Close() error {
@@ -313,3 +292,11 @@ func (s *service) notify() <-chan os.Signal {
 	signal.Notify(sigs, syscall.SIGINT, syscall.SIGQUIT)
 	return sigs
 }
+
+func ignoreMuxError(err error) bool {
+	if err == nil {
+		return true
+	}
+	return strings.Contains(err.Error(), "use of closed network connection") ||
+		strings.Contains(err.Error(), "mux: server closed")
+}

service/web.go

@@ -0,0 +1,35 @@
+package service
+
+import (
+	"net/http"
+	"time"
+
+	"github.com/improbable-eng/grpc-web/go/grpcweb"
+)
+
+var defaultWebOptions = []grpcweb.Option{
+	grpcweb.WithWebsockets(true),
+	grpcweb.WithWebsocketOriginFunc(func(req *http.Request) bool {
+		return true
+	}),
+	grpcweb.WithCorsForRegisteredEndpointsOnly(false),
+	grpcweb.WithOriginFunc(func(origin string) bool {
+		return true
+	}),
+	grpcweb.WithWebsocketPingInterval(time.Second),
+}
+
+func (s *service) grpcWeb(opts ...grpcweb.Option) error {
+	if !s.opts.grpcWeb {
+		return nil
+	}
+	h := grpcweb.WrapServer(s.server, opts...)
+	for _, v := range grpcweb.ListGRPCResources(s.server) {
+		if s.opts.grpcWebPrefix != "" {
+			s.mux.Handle(s.opts.grpcWebPrefix+v, http.StripPrefix(s.opts.grpcWebPrefix, h))
+		} else {
+			s.mux.Handle(v, h)
+		}
+	}
+	return nil
+}