upgrade.go

Documentation: k8s.io/apimachinery/pkg/util/httpstream/spdy

     1  /*
     2  Copyright 2015 The Kubernetes Authors.
     3  
     4  Licensed under the Apache License, Version 2.0 (the "License");
     5  you may not use this file except in compliance with the License.
     6  You may obtain a copy of the License at
     7  
     8      http://www.apache.org/licenses/LICENSE-2.0
     9  
    10  Unless required by applicable law or agreed to in writing, software
    11  distributed under the License is distributed on an "AS IS" BASIS,
    12  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    13  See the License for the specific language governing permissions and
    14  limitations under the License.
    15  */
    16  
    17  package spdy
    18  
    19  import (
    20  	"bufio"
    21  	"fmt"
    22  	"io"
    23  	"net"
    24  	"net/http"
    25  	"strings"
    26  	"sync/atomic"
    27  	"time"
    28  
    29  	"k8s.io/apimachinery/pkg/util/httpstream"
    30  	"k8s.io/apimachinery/pkg/util/runtime"
    31  )
    32  
    33  const HeaderSpdy31 = "SPDY/3.1"
    34  
    35  // responseUpgrader knows how to upgrade HTTP responses. It
    36  // implements the httpstream.ResponseUpgrader interface.
    37  type responseUpgrader struct {
    38  	pingPeriod time.Duration
    39  }
    40  
    41  // connWrapper is used to wrap a hijacked connection and its bufio.Reader. All
    42  // calls will be handled directly by the underlying net.Conn with the exception
    43  // of Read and Close calls, which will consider data in the bufio.Reader. This
    44  // ensures that data already inside the used bufio.Reader instance is also
    45  // read.
    46  type connWrapper struct {
    47  	net.Conn
    48  	closed    int32
    49  	bufReader *bufio.Reader
    50  }
    51  
    52  func (w *connWrapper) Read(b []byte) (n int, err error) {
    53  	if atomic.LoadInt32(&w.closed) == 1 {
    54  		return 0, io.EOF
    55  	}
    56  	return w.bufReader.Read(b)
    57  }
    58  
    59  func (w *connWrapper) Close() error {
    60  	err := w.Conn.Close()
    61  	atomic.StoreInt32(&w.closed, 1)
    62  	return err
    63  }
    64  
    65  // NewResponseUpgrader returns a new httpstream.ResponseUpgrader that is
    66  // capable of upgrading HTTP responses using SPDY/3.1 via the
    67  // spdystream package.
    68  func NewResponseUpgrader() httpstream.ResponseUpgrader {
    69  	return NewResponseUpgraderWithPings(0)
    70  }
    71  
    72  // NewResponseUpgraderWithPings returns a new httpstream.ResponseUpgrader that
    73  // is capable of upgrading HTTP responses using SPDY/3.1 via the spdystream
    74  // package.
    75  //
    76  // If pingPeriod is non-zero, for each incoming connection a background
    77  // goroutine will send periodic Ping frames to the server. Use this to keep
    78  // idle connections through certain load balancers alive longer.
    79  func NewResponseUpgraderWithPings(pingPeriod time.Duration) httpstream.ResponseUpgrader {
    80  	return responseUpgrader{pingPeriod: pingPeriod}
    81  }
    82  
    83  // UpgradeResponse upgrades an HTTP response to one that supports multiplexed
    84  // streams. newStreamHandler will be called synchronously whenever the
    85  // other end of the upgraded connection creates a new stream.
    86  func (u responseUpgrader) UpgradeResponse(w http.ResponseWriter, req *http.Request, newStreamHandler httpstream.NewStreamHandler) httpstream.Connection {
    87  	connectionHeader := strings.ToLower(req.Header.Get(httpstream.HeaderConnection))
    88  	upgradeHeader := strings.ToLower(req.Header.Get(httpstream.HeaderUpgrade))
    89  	if !strings.Contains(connectionHeader, strings.ToLower(httpstream.HeaderUpgrade)) || !strings.Contains(upgradeHeader, strings.ToLower(HeaderSpdy31)) {
    90  		errorMsg := fmt.Sprintf("unable to upgrade: missing upgrade headers in request: %#v", req.Header)
    91  		http.Error(w, errorMsg, http.StatusBadRequest)
    92  		return nil
    93  	}
    94  
    95  	hijacker, ok := w.(http.Hijacker)
    96  	if !ok {
    97  		errorMsg := "unable to upgrade: unable to hijack response"
    98  		http.Error(w, errorMsg, http.StatusInternalServerError)
    99  		return nil
   100  	}
   101  
   102  	w.Header().Add(httpstream.HeaderConnection, httpstream.HeaderUpgrade)
   103  	w.Header().Add(httpstream.HeaderUpgrade, HeaderSpdy31)
   104  	w.WriteHeader(http.StatusSwitchingProtocols)
   105  
   106  	conn, bufrw, err := hijacker.Hijack()
   107  	if err != nil {
   108  		runtime.HandleError(fmt.Errorf("unable to upgrade: error hijacking response: %v", err))
   109  		return nil
   110  	}
   111  
   112  	connWithBuf := &connWrapper{Conn: conn, bufReader: bufrw.Reader}
   113  	spdyConn, err := NewServerConnectionWithPings(connWithBuf, newStreamHandler, u.pingPeriod)
   114  	if err != nil {
   115  		runtime.HandleError(fmt.Errorf("unable to upgrade: error creating SPDY server connection: %v", err))
   116  		return nil
   117  	}
   118  
   119  	return spdyConn
   120  }
   121
View as plain text