3 * Copyright 2016 gRPC authors.
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
9 * http://www.apache.org/licenses/LICENSE-2.0
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.
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.
37 "github.com/golang/protobuf/proto"
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/stats"
44 "google.golang.org/grpc/status"
47 // NewServerHandlerTransport returns a ServerTransport handling gRPC
48 // from inside an http.Handler. It requires that the http Server
50 func NewServerHandlerTransport(w http.ResponseWriter, r *http.Request, stats stats.Handler) (ServerTransport, error) {
51 if r.ProtoMajor != 2 {
52 return nil, errors.New("gRPC requires HTTP/2")
54 if r.Method != "POST" {
55 return nil, errors.New("invalid gRPC request method")
57 contentType := r.Header.Get("Content-Type")
58 // TODO: do we assume contentType is lowercase? we did before
59 contentSubtype, validContentType := contentSubtype(contentType)
60 if !validContentType {
61 return nil, errors.New("invalid gRPC request content-type")
63 if _, ok := w.(http.Flusher); !ok {
64 return nil, errors.New("gRPC requires a ResponseWriter supporting http.Flusher")
66 if _, ok := w.(http.CloseNotifier); !ok {
67 return nil, errors.New("gRPC requires a ResponseWriter supporting http.CloseNotifier")
70 st := &serverHandlerTransport{
73 closedCh: make(chan struct{}),
74 writes: make(chan func()),
75 contentType: contentType,
76 contentSubtype: contentSubtype,
80 if v := r.Header.Get("grpc-timeout"); v != "" {
81 to, err := decodeTimeout(v)
83 return nil, status.Errorf(codes.Internal, "malformed time-out: %v", err)
89 metakv := []string{"content-type", contentType}
91 metakv = append(metakv, ":authority", r.Host)
93 for k, vv := range r.Header {
94 k = strings.ToLower(k)
95 if isReservedHeader(k) && !isWhitelistedHeader(k) {
98 for _, v := range vv {
99 v, err := decodeMetadataHeader(k, v)
101 return nil, status.Errorf(codes.Internal, "malformed binary metadata: %v", err)
103 metakv = append(metakv, k, v)
106 st.headerMD = metadata.Pairs(metakv...)
111 // serverHandlerTransport is an implementation of ServerTransport
112 // which replies to exactly one gRPC request (exactly one HTTP request),
113 // using the net/http.Handler interface. This http.Handler is guaranteed
114 // at this point to be speaking over HTTP/2, so it's able to speak valid
116 type serverHandlerTransport struct {
117 rw http.ResponseWriter
120 timeout time.Duration
121 didCommonHeaders bool
126 closedCh chan struct{} // closed on Close
128 // writes is a channel of code to run serialized in the
129 // ServeHTTP (HandleStreams) goroutine. The channel is closed
130 // when WriteStatus is called.
133 // block concurrent WriteStatus calls
134 // e.g. grpc/(*serverStream).SendMsg/RecvMsg
135 writeStatusMu sync.Mutex
137 // we just mirror the request content-type
139 // we store both contentType and contentSubtype so we don't keep recreating them
140 // TODO make sure this is consistent across handler_server and http2_server
141 contentSubtype string
146 func (ht *serverHandlerTransport) Close() error {
147 ht.closeOnce.Do(ht.closeCloseChanOnce)
151 func (ht *serverHandlerTransport) closeCloseChanOnce() { close(ht.closedCh) }
153 func (ht *serverHandlerTransport) RemoteAddr() net.Addr { return strAddr(ht.req.RemoteAddr) }
155 // strAddr is a net.Addr backed by either a TCP "ip:port" string, or
156 // the empty string if unknown.
159 func (a strAddr) Network() string {
161 // Per the documentation on net/http.Request.RemoteAddr, if this is
162 // set, it's set to the IP:port of the peer (hence, TCP):
163 // https://golang.org/pkg/net/http/#Request
165 // If we want to support Unix sockets later, we can
166 // add our own grpc-specific convention within the
167 // grpc codebase to set RemoteAddr to a different
168 // format, or probably better: we can attach it to the
169 // context and use that from serverHandlerTransport.RemoteAddr.
175 func (a strAddr) String() string { return string(a) }
177 // do runs fn in the ServeHTTP goroutine.
178 func (ht *serverHandlerTransport) do(fn func()) error {
179 // Avoid a panic writing to closed channel. Imperfect but maybe good enough.
182 return ErrConnClosing
185 case ht.writes <- fn:
188 return ErrConnClosing
193 func (ht *serverHandlerTransport) WriteStatus(s *Stream, st *status.Status) error {
194 ht.writeStatusMu.Lock()
195 defer ht.writeStatusMu.Unlock()
197 err := ht.do(func() {
198 ht.writeCommonHeaders(s)
200 // And flush, in case no header or body has been sent yet.
201 // This forces a separation of headers and trailers if this is the
202 // first call (for example, in end2end tests's TestNoService).
203 ht.rw.(http.Flusher).Flush()
206 h.Set("Grpc-Status", fmt.Sprintf("%d", st.Code()))
207 if m := st.Message(); m != "" {
208 h.Set("Grpc-Message", encodeGrpcMessage(m))
211 if p := st.Proto(); p != nil && len(p.Details) > 0 {
212 stBytes, err := proto.Marshal(p)
214 // TODO: return error instead, when callers are able to handle it.
218 h.Set("Grpc-Status-Details-Bin", encodeBinHeader(stBytes))
221 if md := s.Trailer(); len(md) > 0 {
222 for k, vv := range md {
223 // Clients don't tolerate reading restricted headers after some non restricted ones were sent.
224 if isReservedHeader(k) {
227 for _, v := range vv {
228 // http2 ResponseWriter mechanism to send undeclared Trailers after
229 // the headers have possibly been written.
230 h.Add(http2.TrailerPrefix+k, encodeMetadataHeader(k, v))
236 if err == nil { // transport has not been closed
238 ht.stats.HandleRPC(s.Context(), &stats.OutTrailer{})
246 // writeCommonHeaders sets common headers on the first write
247 // call (Write, WriteHeader, or WriteStatus).
248 func (ht *serverHandlerTransport) writeCommonHeaders(s *Stream) {
249 if ht.didCommonHeaders {
252 ht.didCommonHeaders = true
255 h["Date"] = nil // suppress Date to make tests happy; TODO: restore
256 h.Set("Content-Type", ht.contentType)
258 // Predeclare trailers we'll set later in WriteStatus (after the body).
259 // This is a SHOULD in the HTTP RFC, and the way you add (known)
260 // Trailers per the net/http.ResponseWriter contract.
261 // See https://golang.org/pkg/net/http/#ResponseWriter
262 // and https://golang.org/pkg/net/http/#example_ResponseWriter_trailers
263 h.Add("Trailer", "Grpc-Status")
264 h.Add("Trailer", "Grpc-Message")
265 h.Add("Trailer", "Grpc-Status-Details-Bin")
267 if s.sendCompress != "" {
268 h.Set("Grpc-Encoding", s.sendCompress)
272 func (ht *serverHandlerTransport) Write(s *Stream, hdr []byte, data []byte, opts *Options) error {
273 return ht.do(func() {
274 ht.writeCommonHeaders(s)
277 ht.rw.(http.Flusher).Flush()
281 func (ht *serverHandlerTransport) WriteHeader(s *Stream, md metadata.MD) error {
282 err := ht.do(func() {
283 ht.writeCommonHeaders(s)
285 for k, vv := range md {
286 // Clients don't tolerate reading restricted headers after some non restricted ones were sent.
287 if isReservedHeader(k) {
290 for _, v := range vv {
291 v = encodeMetadataHeader(k, v)
295 ht.rw.WriteHeader(200)
296 ht.rw.(http.Flusher).Flush()
301 ht.stats.HandleRPC(s.Context(), &stats.OutHeader{})
307 func (ht *serverHandlerTransport) HandleStreams(startStream func(*Stream), traceCtx func(context.Context, string) context.Context) {
308 // With this transport type there will be exactly 1 stream: this HTTP request.
310 ctx := ht.req.Context()
311 var cancel context.CancelFunc
313 ctx, cancel = context.WithTimeout(ctx, ht.timeout)
315 ctx, cancel = context.WithCancel(ctx)
318 // requestOver is closed when either the request's context is done
319 // or the status has been written via WriteStatus.
320 requestOver := make(chan struct{})
322 // clientGone receives a single value if peer is gone, either
323 // because the underlying connection is dead or because the
324 // peer sends an http2 RST_STREAM.
325 clientGone := ht.rw.(http.CloseNotifier).CloseNotify()
340 requestRead: func(int) {},
342 buf: newRecvBuffer(),
344 method: req.URL.Path,
345 recvCompress: req.Header.Get("grpc-encoding"),
346 contentSubtype: ht.contentSubtype,
349 Addr: ht.RemoteAddr(),
352 pr.AuthInfo = credentials.TLSInfo{State: *req.TLS}
354 ctx = metadata.NewIncomingContext(ctx, ht.headerMD)
355 s.ctx = peer.NewContext(ctx, pr)
357 s.ctx = ht.stats.TagRPC(s.ctx, &stats.RPCTagInfo{FullMethodName: s.method})
358 inHeader := &stats.InHeader{
359 FullMethod: s.method,
360 RemoteAddr: ht.RemoteAddr(),
361 Compression: s.recvCompress,
363 ht.stats.HandleRPC(s.ctx, inHeader)
365 s.trReader = &transportReader{
366 reader: &recvBufferReader{ctx: s.ctx, ctxDone: s.ctx.Done(), recv: s.buf},
367 windowHandler: func(int) {},
370 // readerDone is closed when the Body.Read-ing goroutine exits.
371 readerDone := make(chan struct{})
373 defer close(readerDone)
375 // TODO: minimize garbage, optimize recvBuffer code/ownership
376 const readSize = 8196
377 for buf := make([]byte, readSize); ; {
378 n, err := req.Body.Read(buf)
380 s.buf.put(recvMsg{data: buf[:n:n]})
384 s.buf.put(recvMsg{err: mapRecvMsgError(err)})
388 buf = make([]byte, readSize)
393 // startStream is provided by the *grpc.Server's serveStreams.
394 // It starts a goroutine serving s and exits immediately.
395 // The goroutine that is started is the one that then calls
396 // into ht, calling WriteHeader, Write, WriteStatus, Close, etc.
402 // Wait for reading goroutine to finish.
407 func (ht *serverHandlerTransport) runStream() {
410 case fn, ok := <-ht.writes:
421 func (ht *serverHandlerTransport) IncrMsgSent() {}
423 func (ht *serverHandlerTransport) IncrMsgRecv() {}
425 func (ht *serverHandlerTransport) Drain() {
426 panic("Drain() is not implemented")
429 // mapRecvMsgError returns the non-nil err into the appropriate
430 // error value as expected by callers of *grpc.parser.recvMsg.
431 // In particular, in can only be:
433 // * io.ErrUnexpectedEOF
434 // * of type transport.ConnectionError
435 // * an error from the status package
436 func mapRecvMsgError(err error) error {
437 if err == io.EOF || err == io.ErrUnexpectedEOF {
440 if se, ok := err.(http2.StreamError); ok {
441 if code, ok := http2ErrConvTab[se.Code]; ok {
442 return status.Error(code, se.Error())
445 if strings.Contains(err.Error(), "body closed by handler") {
446 return status.Error(codes.Canceled, err.Error())
448 return connectionErrorf(true, err, err.Error())