code: func (h *portForwardProxy) run() {
  	h.Debugf("Waiting for port forward streams.")
  	for {
  		select {
  		case <-h.context.Done():
  			h.Debugf("Context is closing, returning.")
  			return
  		case <-h.sourceConn.CloseChan():
  			h.Debugf("Upgraded connection closed.")
  			return
  		case stream := <-h.streamChan:
  			requestID, err := h.requestID(stream)
  			if err != nil {
  				h.Warningf("Failed to parse request id: %v.", err)
  				return
  			}
  			streamType := stream.Headers().Get(StreamType)
  			h.Debugf("Received new stream %v of type %v.", requestID, streamType)
  
  			p, created := h.getStreamPair(requestID)
  			if created {
  				go h.monitorStreamPair(p, time.After(h.streamCreationTimeout))
  			}
  			if complete, err := p.add(stream); err != nil {
  				msg := fmt.Sprintf("error processing stream for request %s: %v", requestID, err)
  				p.printError(msg)
  			} else if complete {
  				go h.portForward(p)
  			}
  		}
  	}
  }
docs: // run is the main loop for the portForwardProxy. It processes new
  // streams, invoking portForward for each complete stream pair. The loop exits
  // when the httpstream.Connection is closed.