|
@@ -52,6 +52,7 @@ import (
|
|
|
"google.golang.org/grpc/codes"
|
|
"google.golang.org/grpc/codes"
|
|
|
"google.golang.org/grpc/credentials"
|
|
"google.golang.org/grpc/credentials"
|
|
|
"google.golang.org/grpc/grpclog"
|
|
"google.golang.org/grpc/grpclog"
|
|
|
|
|
+ "google.golang.org/grpc/internal"
|
|
|
"google.golang.org/grpc/metadata"
|
|
"google.golang.org/grpc/metadata"
|
|
|
"google.golang.org/grpc/transport"
|
|
"google.golang.org/grpc/transport"
|
|
|
)
|
|
)
|
|
@@ -247,7 +248,6 @@ func (s *Server) Serve(lis net.Listener) error {
|
|
|
delete(s.lis, lis)
|
|
delete(s.lis, lis)
|
|
|
s.mu.Unlock()
|
|
s.mu.Unlock()
|
|
|
}()
|
|
}()
|
|
|
- listenerAddr := lis.Addr()
|
|
|
|
|
for {
|
|
for {
|
|
|
rawConn, err := lis.Accept()
|
|
rawConn, err := lis.Accept()
|
|
|
if err != nil {
|
|
if err != nil {
|
|
@@ -258,13 +258,13 @@ func (s *Server) Serve(lis net.Listener) error {
|
|
|
}
|
|
}
|
|
|
// Start a new goroutine to deal with rawConn
|
|
// Start a new goroutine to deal with rawConn
|
|
|
// so we don't stall this Accept loop goroutine.
|
|
// so we don't stall this Accept loop goroutine.
|
|
|
- go s.handleRawConn(listenerAddr, rawConn)
|
|
|
|
|
|
|
+ go s.handleRawConn(rawConn)
|
|
|
}
|
|
}
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
// handleRawConn is run in its own goroutine and handles a just-accepted
|
|
// handleRawConn is run in its own goroutine and handles a just-accepted
|
|
|
// connection that has not had any I/O performed on it yet.
|
|
// connection that has not had any I/O performed on it yet.
|
|
|
-func (s *Server) handleRawConn(listenerAddr net.Addr, rawConn net.Conn) {
|
|
|
|
|
|
|
+func (s *Server) handleRawConn(rawConn net.Conn) {
|
|
|
conn, authInfo, err := s.useTransportAuthenticator(rawConn)
|
|
conn, authInfo, err := s.useTransportAuthenticator(rawConn)
|
|
|
if err != nil {
|
|
if err != nil {
|
|
|
s.mu.Lock()
|
|
s.mu.Lock()
|
|
@@ -284,7 +284,7 @@ func (s *Server) handleRawConn(listenerAddr net.Addr, rawConn net.Conn) {
|
|
|
s.mu.Unlock()
|
|
s.mu.Unlock()
|
|
|
|
|
|
|
|
if s.opts.useHandlerImpl {
|
|
if s.opts.useHandlerImpl {
|
|
|
- s.serveUsingHandler(listenerAddr, conn)
|
|
|
|
|
|
|
+ s.serveUsingHandler(conn)
|
|
|
} else {
|
|
} else {
|
|
|
s.serveNewHTTP2Transport(conn, authInfo)
|
|
s.serveNewHTTP2Transport(conn, authInfo)
|
|
|
}
|
|
}
|
|
@@ -340,29 +340,18 @@ var _ http.Handler = (*Server)(nil)
|
|
|
// method as one of the environment types.
|
|
// method as one of the environment types.
|
|
|
//
|
|
//
|
|
|
// conn is the *tls.Conn that's already been authenticated.
|
|
// conn is the *tls.Conn that's already been authenticated.
|
|
|
-func (s *Server) serveUsingHandler(listenerAddr net.Addr, conn net.Conn) {
|
|
|
|
|
|
|
+func (s *Server) serveUsingHandler(conn net.Conn) {
|
|
|
if !s.addConn(conn) {
|
|
if !s.addConn(conn) {
|
|
|
conn.Close()
|
|
conn.Close()
|
|
|
return
|
|
return
|
|
|
}
|
|
}
|
|
|
defer s.removeConn(conn)
|
|
defer s.removeConn(conn)
|
|
|
- connDone := make(chan struct{})
|
|
|
|
|
- hs := &http.Server{
|
|
|
|
|
- Handler: s,
|
|
|
|
|
- ConnState: func(c net.Conn, cs http.ConnState) {
|
|
|
|
|
- if cs == http.StateClosed {
|
|
|
|
|
- close(connDone)
|
|
|
|
|
- }
|
|
|
|
|
- },
|
|
|
|
|
- }
|
|
|
|
|
- if err := http2.ConfigureServer(hs, &http2.Server{
|
|
|
|
|
|
|
+ h2s := &http2.Server{
|
|
|
MaxConcurrentStreams: s.opts.maxConcurrentStreams,
|
|
MaxConcurrentStreams: s.opts.maxConcurrentStreams,
|
|
|
- }); err != nil {
|
|
|
|
|
- grpclog.Fatalf("grpc: http2.ConfigureServer: %v", err)
|
|
|
|
|
- return
|
|
|
|
|
}
|
|
}
|
|
|
- hs.Serve(&singleConnListener{addr: listenerAddr, conn: conn})
|
|
|
|
|
- <-connDone
|
|
|
|
|
|
|
+ h2s.ServeConn(conn, &http2.ServeConnOpts{
|
|
|
|
|
+ Handler: s,
|
|
|
|
|
+ })
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
|
func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
|
@@ -705,10 +694,18 @@ func (s *Server) Stop() {
|
|
|
s.mu.Unlock()
|
|
s.mu.Unlock()
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
-// TestingCloseConns closes all exiting transports but keeps s.lis accepting new
|
|
|
|
|
-// connections.
|
|
|
|
|
-// This is only for tests and is subject to removal.
|
|
|
|
|
-func (s *Server) TestingCloseConns() {
|
|
|
|
|
|
|
+func init() {
|
|
|
|
|
+ internal.TestingCloseConns = func(arg interface{}) {
|
|
|
|
|
+ arg.(*Server).testingCloseConns()
|
|
|
|
|
+ }
|
|
|
|
|
+ internal.TestingUseHandlerImpl = func(arg interface{}) {
|
|
|
|
|
+ arg.(*Server).opts.useHandlerImpl = true
|
|
|
|
|
+ }
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+// testingCloseConns closes all existing transports but keeps s.lis
|
|
|
|
|
+// accepting new connections.
|
|
|
|
|
+func (s *Server) testingCloseConns() {
|
|
|
s.mu.Lock()
|
|
s.mu.Lock()
|
|
|
for c := range s.conns {
|
|
for c := range s.conns {
|
|
|
c.Close()
|
|
c.Close()
|
|
@@ -717,13 +714,6 @@ func (s *Server) TestingCloseConns() {
|
|
|
s.mu.Unlock()
|
|
s.mu.Unlock()
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
-// TestingUseHandlerImpl enables the http.Handler-based server implementation.
|
|
|
|
|
-// It must be called before Serve and requires TLS credentials.
|
|
|
|
|
-// This is only for tests and is subject to removal.
|
|
|
|
|
-func (s *Server) TestingUseHandlerImpl() {
|
|
|
|
|
- s.opts.useHandlerImpl = true
|
|
|
|
|
-}
|
|
|
|
|
-
|
|
|
|
|
// SendHeader sends header metadata. It may be called at most once from a unary
|
|
// SendHeader sends header metadata. It may be called at most once from a unary
|
|
|
// RPC handler. The ctx is the RPC handler's Context or one derived from it.
|
|
// RPC handler. The ctx is the RPC handler's Context or one derived from it.
|
|
|
func SendHeader(ctx context.Context, md metadata.MD) error {
|
|
func SendHeader(ctx context.Context, md metadata.MD) error {
|
|
@@ -754,30 +744,3 @@ func SetTrailer(ctx context.Context, md metadata.MD) error {
|
|
|
}
|
|
}
|
|
|
return stream.SetTrailer(md)
|
|
return stream.SetTrailer(md)
|
|
|
}
|
|
}
|
|
|
-
|
|
|
|
|
-// singleConnListener is a net.Listener that yields a single conn.
|
|
|
|
|
-type singleConnListener struct {
|
|
|
|
|
- mu sync.Mutex
|
|
|
|
|
- addr net.Addr
|
|
|
|
|
- conn net.Conn // nil if done
|
|
|
|
|
-}
|
|
|
|
|
-
|
|
|
|
|
-func (ln *singleConnListener) Addr() net.Addr { return ln.addr }
|
|
|
|
|
-
|
|
|
|
|
-func (ln *singleConnListener) Close() error {
|
|
|
|
|
- ln.mu.Lock()
|
|
|
|
|
- defer ln.mu.Unlock()
|
|
|
|
|
- ln.conn = nil
|
|
|
|
|
- return nil
|
|
|
|
|
-}
|
|
|
|
|
-
|
|
|
|
|
-func (ln *singleConnListener) Accept() (net.Conn, error) {
|
|
|
|
|
- ln.mu.Lock()
|
|
|
|
|
- defer ln.mu.Unlock()
|
|
|
|
|
- c := ln.conn
|
|
|
|
|
- if c == nil {
|
|
|
|
|
- return nil, io.EOF
|
|
|
|
|
- }
|
|
|
|
|
- ln.conn = nil
|
|
|
|
|
- return c, nil
|
|
|
|
|
-}
|
|
|