linka-cloud/grpc-toolkit
2
0
Fork 0
mirror of https://github.com/linka-cloud/grpc.git synced 2025-06-22 09:12:28 +00:00
Code Issues Packages Projects Releases Wiki Activity

add h2c/http2 support with WithoutCmux option

Browse Source 
Signed-off-by: Adphi <philippe.adrien.nousse@gmail.com>
This commit is contained in:
Adphi
2023-09-16 16:21:51 +02:00
parent f455c9994c
commit c0c19683cf
3 changed files with 99 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files

94
service/service.go
View File

@ -21,6 +21,8 @@ import (
"github.com/rs/cors"
"github.com/soheilhy/cmux"
"go.uber.org/multierr"
"golang.org/x/net/http2"
"golang.org/x/net/http2/h2c"
"golang.org/x/sync/errgroup"
"google.golang.org/grpc"
"google.golang.org/grpc/health"
@ -170,12 +172,6 @@ func (s *service) start() (*errgroup.Group, error) {
s.opts.address = s.opts.lis.Addr().String()
}
mux := cmux.New(s.opts.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 {
return nil, err
@ -204,34 +200,23 @@ func (s *service) start() (*errgroup.Group, error) {
}
}
g, ctx := errgroup.WithContext(s.opts.ctx)
if s.opts.mux != nil {
hServer := &http.Server{
Handler: alice.New(s.opts.middlewares...).Then(cors.New(s.opts.cors).Handler(s.opts.mux)),
}
g.Go(func() error {
defer hServer.Shutdown(ctx)
return hServer.Serve(hList)
})
fn := s.runWithCmux
if s.opts.withoutCmux {
fn = s.runWithoutCmux
}
g.Go(func() error {
return s.server.Serve(gLis)
})
g.Go(func() error {
return ignoreMuxError(mux.Serve())
})
g, ctx := errgroup.WithContext(s.opts.ctx)
if err := fn(ctx, g); err != nil {
return nil, err
}
if s.healthServer != nil {
for k := range s.services {
s.healthServer.SetServingStatus(k, grpc_health_v1.HealthCheckResponse_SERVING)
}
defer func() {
if s.healthServer != nil {
for k := range s.services {
s.healthServer.SetServingStatus(k, grpc_health_v1.HealthCheckResponse_NOT_SERVING)
}
for k := range s.services {
s.healthServer.SetServingStatus(k, grpc_health_v1.HealthCheckResponse_NOT_SERVING)
}
}()
}
@ -268,6 +253,63 @@ func (s *service) run() error {
}
}
func (s *service) runWithoutCmux(ctx context.Context, g *errgroup.Group) error {
if s.opts.mux != nil {
handler := alice.New(s.opts.middlewares...).Then(cors.New(s.opts.cors).Handler(s.opts.mux))
hServer := &http.Server{
Handler: h2c.NewHandler(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.ProtoMajor == 2 && r.Header.Get("Content-Type") == "application/grpc" {
s.server.ServeHTTP(w, r)
} else {
handler.ServeHTTP(w, r)
}
}), &http2.Server{}),
TLSConfig: s.opts.tlsConfig,
}
if err := http2.ConfigureServer(hServer, &http2.Server{}); err != nil {
return err
}
g.Go(func() error {
defer hServer.Shutdown(ctx)
return hServer.Serve(s.opts.lis)
})
} else {
g.Go(func() error {
return s.server.Serve(s.opts.lis)
})
}
return nil
}
func (s *service) runWithCmux(ctx context.Context, g *errgroup.Group) error {
mux := cmux.New(s.opts.lis)
mux.SetReadTimeout(5 * time.Second)
gLis := mux.MatchWithWriters(cmux.HTTP2MatchHeaderFieldSendSettings("content-type", "application/grpc"))
hList := mux.Match(cmux.Any())
if s.opts.mux != nil {
hServer := &http.Server{
Handler: alice.New(s.opts.middlewares...).Then(cors.New(s.opts.cors).Handler(s.opts.mux)),
TLSConfig: s.opts.tlsConfig,
}
g.Go(func() error {
defer hServer.Shutdown(ctx)
return hServer.Serve(hList)
})
}
g.Go(func() error {
return s.server.Serve(gLis)
})
g.Go(func() error {
return ignoreMuxError(mux.Serve())
})
return nil
}
func (s *service) Serve(lis net.Listener) error {
s.mu.Lock()
s.opts.lis = lis