vendor/google.golang.org/grpc/transport/handler_server.go

   1 /*
   2  *
   3  * Copyright 2016 gRPC authors.
   4  *
   5  * Licensed under the Apache License, Version 2.0 (the "License");
   6  * you may not use this file except in compliance with the License.
   7  * You may obtain a copy of the License at
   8  *
   9  *     http://www.apache.org/licenses/LICENSE-2.0
  10  *
  11  * Unless required by applicable law or agreed to in writing, software
  12  * distributed under the License is distributed on an "AS IS" BASIS,
  13  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14  * See the License for the specific language governing permissions and
  15  * limitations under the License.
  16  *
  17  */
  18
  19 // This file is the implementation of a gRPC server using HTTP/2 which
  20 // uses the standard Go http2 Server implementation (via the
  21 // http.Handler interface), rather than speaking low-level HTTP/2
  22 // frames itself. It is the implementation of *grpc.Server.ServeHTTP.
  23
  24 package transport
  25
  26 import (
  27         "errors"
  28         "fmt"
  29         "io"
  30         "net"
  31         "net/http"
  32         "strings"
  33         "sync"
  34         "time"
  35
  36         "github.com/golang/protobuf/proto"
  37         "golang.org/x/net/context"
  38         "golang.org/x/net/http2"
  39         "google.golang.org/grpc/codes"
  40         "google.golang.org/grpc/credentials"
  41         "google.golang.org/grpc/metadata"
  42         "google.golang.org/grpc/peer"
  43         "google.golang.org/grpc/status"
  44 )
  45
  46 // NewServerHandlerTransport returns a ServerTransport handling gRPC
  47 // from inside an http.Handler. It requires that the http Server
  48 // supports HTTP/2.
  49 func NewServerHandlerTransport(w http.ResponseWriter, r *http.Request) (ServerTransport, error) {
  50         if r.ProtoMajor != 2 {
  51                 return nil, errors.New("gRPC requires HTTP/2")
  52         }
  53         if r.Method != "POST" {
  54                 return nil, errors.New("invalid gRPC request method")
  55         }
  56         if !validContentType(r.Header.Get("Content-Type")) {
  57                 return nil, errors.New("invalid gRPC request content-type")
  58         }
  59         if _, ok := w.(http.Flusher); !ok {
  60                 return nil, errors.New("gRPC requires a ResponseWriter supporting http.Flusher")
  61         }
  62         if _, ok := w.(http.CloseNotifier); !ok {
  63                 return nil, errors.New("gRPC requires a ResponseWriter supporting http.CloseNotifier")
  64         }
  65
  66         st := &serverHandlerTransport{
  67                 rw:       w,
  68                 req:      r,
  69                 closedCh: make(chan struct{}),
  70                 writes:   make(chan func()),
  71         }
  72
  73         if v := r.Header.Get("grpc-timeout"); v != "" {
  74                 to, err := decodeTimeout(v)
  75                 if err != nil {
  76                         return nil, streamErrorf(codes.Internal, "malformed time-out: %v", err)
  77                 }
  78                 st.timeoutSet = true
  79                 st.timeout = to
  80         }
  81
  82         var metakv []string
  83         if r.Host != "" {
  84                 metakv = append(metakv, ":authority", r.Host)
  85         }
  86         for k, vv := range r.Header {
  87                 k = strings.ToLower(k)
  88                 if isReservedHeader(k) && !isWhitelistedPseudoHeader(k) {
  89                         continue
  90                 }
  91                 for _, v := range vv {
  92                         v, err := decodeMetadataHeader(k, v)
  93                         if err != nil {
  94                                 return nil, streamErrorf(codes.InvalidArgument, "malformed binary metadata: %v", err)
  95                         }
  96                         metakv = append(metakv, k, v)
  97                 }
  98         }
  99         st.headerMD = metadata.Pairs(metakv...)
 100
 101         return st, nil
 102 }
 103
 104 // serverHandlerTransport is an implementation of ServerTransport
 105 // which replies to exactly one gRPC request (exactly one HTTP request),
 106 // using the net/http.Handler interface. This http.Handler is guaranteed
 107 // at this point to be speaking over HTTP/2, so it's able to speak valid
 108 // gRPC.
 109 type serverHandlerTransport struct {
 110         rw               http.ResponseWriter
 111         req              *http.Request
 112         timeoutSet       bool
 113         timeout          time.Duration
 114         didCommonHeaders bool
 115
 116         headerMD metadata.MD
 117
 118         closeOnce sync.Once
 119         closedCh  chan struct{} // closed on Close
 120
 121         // writes is a channel of code to run serialized in the
 122         // ServeHTTP (HandleStreams) goroutine. The channel is closed
 123         // when WriteStatus is called.
 124         writes chan func()
 125
 126         mu sync.Mutex
 127         // streamDone indicates whether WriteStatus has been called and writes channel
 128         // has been closed.
 129         streamDone bool
 130 }
 131
 132 func (ht *serverHandlerTransport) Close() error {
 133         ht.closeOnce.Do(ht.closeCloseChanOnce)
 134         return nil
 135 }
 136
 137 func (ht *serverHandlerTransport) closeCloseChanOnce() { close(ht.closedCh) }
 138
 139 func (ht *serverHandlerTransport) RemoteAddr() net.Addr { return strAddr(ht.req.RemoteAddr) }
 140
 141 // strAddr is a net.Addr backed by either a TCP "ip:port" string, or
 142 // the empty string if unknown.
 143 type strAddr string
 144
 145 func (a strAddr) Network() string {
 146         if a != "" {
 147                 // Per the documentation on net/http.Request.RemoteAddr, if this is
 148                 // set, it's set to the IP:port of the peer (hence, TCP):
 149                 // https://golang.org/pkg/net/http/#Request
 150                 //
 151                 // If we want to support Unix sockets later, we can
 152                 // add our own grpc-specific convention within the
 153                 // grpc codebase to set RemoteAddr to a different
 154                 // format, or probably better: we can attach it to the
 155                 // context and use that from serverHandlerTransport.RemoteAddr.
 156                 return "tcp"
 157         }
 158         return ""
 159 }
 160
 161 func (a strAddr) String() string { return string(a) }
 162
 163 // do runs fn in the ServeHTTP goroutine.
 164 func (ht *serverHandlerTransport) do(fn func()) error {
 165         // Avoid a panic writing to closed channel. Imperfect but maybe good enough.
 166         select {
 167         case <-ht.closedCh:
 168                 return ErrConnClosing
 169         default:
 170                 select {
 171                 case ht.writes <- fn:
 172                         return nil
 173                 case <-ht.closedCh:
 174                         return ErrConnClosing
 175                 }
 176         }
 177 }
 178
 179 func (ht *serverHandlerTransport) WriteStatus(s *Stream, st *status.Status) error {
 180         ht.mu.Lock()
 181         if ht.streamDone {
 182                 ht.mu.Unlock()
 183                 return nil
 184         }
 185         ht.streamDone = true
 186         ht.mu.Unlock()
 187         err := ht.do(func() {
 188                 ht.writeCommonHeaders(s)
 189
 190                 // And flush, in case no header or body has been sent yet.
 191                 // This forces a separation of headers and trailers if this is the
 192                 // first call (for example, in end2end tests's TestNoService).
 193                 ht.rw.(http.Flusher).Flush()
 194
 195                 h := ht.rw.Header()
 196                 h.Set("Grpc-Status", fmt.Sprintf("%d", st.Code()))
 197                 if m := st.Message(); m != "" {
 198                         h.Set("Grpc-Message", encodeGrpcMessage(m))
 199                 }
 200
 201                 if p := st.Proto(); p != nil && len(p.Details) > 0 {
 202                         stBytes, err := proto.Marshal(p)
 203                         if err != nil {
 204                                 // TODO: return error instead, when callers are able to handle it.
 205                                 panic(err)
 206                         }
 207
 208                         h.Set("Grpc-Status-Details-Bin", encodeBinHeader(stBytes))
 209                 }
 210
 211                 if md := s.Trailer(); len(md) > 0 {
 212                         for k, vv := range md {
 213                                 // Clients don't tolerate reading restricted headers after some non restricted ones were sent.
 214                                 if isReservedHeader(k) {
 215                                         continue
 216                                 }
 217                                 for _, v := range vv {
 218                                         // http2 ResponseWriter mechanism to send undeclared Trailers after
 219                                         // the headers have possibly been written.
 220                                         h.Add(http2.TrailerPrefix+k, encodeMetadataHeader(k, v))
 221                                 }
 222                         }
 223                 }
 224         })
 225         close(ht.writes)
 226         return err
 227 }
 228
 229 // writeCommonHeaders sets common headers on the first write
 230 // call (Write, WriteHeader, or WriteStatus).
 231 func (ht *serverHandlerTransport) writeCommonHeaders(s *Stream) {
 232         if ht.didCommonHeaders {
 233                 return
 234         }
 235         ht.didCommonHeaders = true
 236
 237         h := ht.rw.Header()
 238         h["Date"] = nil // suppress Date to make tests happy; TODO: restore
 239         h.Set("Content-Type", "application/grpc")
 240
 241         // Predeclare trailers we'll set later in WriteStatus (after the body).
 242         // This is a SHOULD in the HTTP RFC, and the way you add (known)
 243         // Trailers per the net/http.ResponseWriter contract.
 244         // See https://golang.org/pkg/net/http/#ResponseWriter
 245         // and https://golang.org/pkg/net/http/#example_ResponseWriter_trailers
 246         h.Add("Trailer", "Grpc-Status")
 247         h.Add("Trailer", "Grpc-Message")
 248         h.Add("Trailer", "Grpc-Status-Details-Bin")
 249
 250         if s.sendCompress != "" {
 251                 h.Set("Grpc-Encoding", s.sendCompress)
 252         }
 253 }
 254
 255 func (ht *serverHandlerTransport) Write(s *Stream, hdr []byte, data []byte, opts *Options) error {
 256         return ht.do(func() {
 257                 ht.writeCommonHeaders(s)
 258                 ht.rw.Write(hdr)
 259                 ht.rw.Write(data)
 260                 if !opts.Delay {
 261                         ht.rw.(http.Flusher).Flush()
 262                 }
 263         })
 264 }
 265
 266 func (ht *serverHandlerTransport) WriteHeader(s *Stream, md metadata.MD) error {
 267         return ht.do(func() {
 268                 ht.writeCommonHeaders(s)
 269                 h := ht.rw.Header()
 270                 for k, vv := range md {
 271                         // Clients don't tolerate reading restricted headers after some non restricted ones were sent.
 272                         if isReservedHeader(k) {
 273                                 continue
 274                         }
 275                         for _, v := range vv {
 276                                 v = encodeMetadataHeader(k, v)
 277                                 h.Add(k, v)
 278                         }
 279                 }
 280                 ht.rw.WriteHeader(200)
 281                 ht.rw.(http.Flusher).Flush()
 282         })
 283 }
 284
 285 func (ht *serverHandlerTransport) HandleStreams(startStream func(*Stream), traceCtx func(context.Context, string) context.Context) {
 286         // With this transport type there will be exactly 1 stream: this HTTP request.
 287
 288         var ctx context.Context
 289         var cancel context.CancelFunc
 290         if ht.timeoutSet {
 291                 ctx, cancel = context.WithTimeout(context.Background(), ht.timeout)
 292         } else {
 293                 ctx, cancel = context.WithCancel(context.Background())
 294         }
 295
 296         // requestOver is closed when either the request's context is done
 297         // or the status has been written via WriteStatus.
 298         requestOver := make(chan struct{})
 299
 300         // clientGone receives a single value if peer is gone, either
 301         // because the underlying connection is dead or because the
 302         // peer sends an http2 RST_STREAM.
 303         clientGone := ht.rw.(http.CloseNotifier).CloseNotify()
 304         go func() {
 305                 select {
 306                 case <-requestOver:
 307                         return
 308                 case <-ht.closedCh:
 309                 case <-clientGone:
 310                 }
 311                 cancel()
 312         }()
 313
 314         req := ht.req
 315
 316         s := &Stream{
 317                 id:           0, // irrelevant
 318                 requestRead:  func(int) {},
 319                 cancel:       cancel,
 320                 buf:          newRecvBuffer(),
 321                 st:           ht,
 322                 method:       req.URL.Path,
 323                 recvCompress: req.Header.Get("grpc-encoding"),
 324         }
 325         pr := &peer.Peer{
 326                 Addr: ht.RemoteAddr(),
 327         }
 328         if req.TLS != nil {
 329                 pr.AuthInfo = credentials.TLSInfo{State: *req.TLS}
 330         }
 331         ctx = metadata.NewIncomingContext(ctx, ht.headerMD)
 332         ctx = peer.NewContext(ctx, pr)
 333         s.ctx = newContextWithStream(ctx, s)
 334         s.trReader = &transportReader{
 335                 reader:        &recvBufferReader{ctx: s.ctx, recv: s.buf},
 336                 windowHandler: func(int) {},
 337         }
 338
 339         // readerDone is closed when the Body.Read-ing goroutine exits.
 340         readerDone := make(chan struct{})
 341         go func() {
 342                 defer close(readerDone)
 343
 344                 // TODO: minimize garbage, optimize recvBuffer code/ownership
 345                 const readSize = 8196
 346                 for buf := make([]byte, readSize); ; {
 347                         n, err := req.Body.Read(buf)
 348                         if n > 0 {
 349                                 s.buf.put(recvMsg{data: buf[:n:n]})
 350                                 buf = buf[n:]
 351                         }
 352                         if err != nil {
 353                                 s.buf.put(recvMsg{err: mapRecvMsgError(err)})
 354                                 return
 355                         }
 356                         if len(buf) == 0 {
 357                                 buf = make([]byte, readSize)
 358                         }
 359                 }
 360         }()
 361
 362         // startStream is provided by the *grpc.Server's serveStreams.
 363         // It starts a goroutine serving s and exits immediately.
 364         // The goroutine that is started is the one that then calls
 365         // into ht, calling WriteHeader, Write, WriteStatus, Close, etc.
 366         startStream(s)
 367
 368         ht.runStream()
 369         close(requestOver)
 370
 371         // Wait for reading goroutine to finish.
 372         req.Body.Close()
 373         <-readerDone
 374 }
 375
 376 func (ht *serverHandlerTransport) runStream() {
 377         for {
 378                 select {
 379                 case fn, ok := <-ht.writes:
 380                         if !ok {
 381                                 return
 382                         }
 383                         fn()
 384                 case <-ht.closedCh:
 385                         return
 386                 }
 387         }
 388 }
 389
 390 func (ht *serverHandlerTransport) Drain() {
 391         panic("Drain() is not implemented")
 392 }
 393
 394 // mapRecvMsgError returns the non-nil err into the appropriate
 395 // error value as expected by callers of *grpc.parser.recvMsg.
 396 // In particular, in can only be:
 397 //   * io.EOF
 398 //   * io.ErrUnexpectedEOF
 399 //   * of type transport.ConnectionError
 400 //   * of type transport.StreamError
 401 func mapRecvMsgError(err error) error {
 402         if err == io.EOF || err == io.ErrUnexpectedEOF {
 403                 return err
 404         }
 405         if se, ok := err.(http2.StreamError); ok {
 406                 if code, ok := http2ErrConvTab[se.Code]; ok {
 407                         return StreamError{
 408                                 Code: code,
 409                                 Desc: se.Error(),
 410                         }
 411                 }
 412         }
 413         return connectionErrorf(true, err, err.Error())
 414 }