func (p *PushStream) Finish() {
	p.writeHeader()
	end := new(frames.DATA)
	end.StreamID = p.streamID
	end.Data = []byte{}
	end.Flags = common.FLAG_FIN
	p.output <- end
	p.Close()
}

// Write is used to send data to the connection. This
// takes care of the windowing. Although data may be
// buffered, rather than actually sent, this is not
// visible to the caller.
func (f *flowControl) Write(data []byte) (int, error) {
	l := len(data)
	if l == 0 {
		return 0, nil
	}

	if f.buffer == nil || f.stream == nil {
		return 0, errors.New("Error: Stream closed.")
	}

	// Transfer window processing.
	f.CheckInitialWindow()
	if f.constrained {
		f.Flush()
	}

	f.Lock()
	var window uint32
	if f.transferWindow < 0 {
		window = 0
	} else {
		window = uint32(f.transferWindow)
	}

	constrained := false
	sending := uint32(len(data))
	if sending > window {
		sending = window
		constrained = true
	}

	f.sent += sending
	f.transferWindow -= int64(sending)

	if constrained {
		f.buffer = append(f.buffer, data[window:])
		data = data[:window]
		f.constrained = true
		debug.Printf("Stream %d is now constrained.\n", f.streamID)
	}
	f.Unlock()

	if len(data) == 0 {
		return l, nil
	}

	dataFrame := new(frames.DATA)
	dataFrame.StreamID = f.streamID
	dataFrame.Data = data

	f.output <- dataFrame
	return l, nil
}

// Flush is used to send buffered data to
// the connection, if the transfer window
// will allow. Flush does not guarantee
// that any or all buffered data will be
// sent with a single flush.
func (f *flowControl) Flush() {
	f.CheckInitialWindow()
	if !f.constrained || f.transferWindow == 0 {
		return
	}

	out := make([]byte, 0, f.transferWindow)
	left := f.transferWindow
	for i := 0; i < len(f.buffer); i++ {
		if l := int64(len(f.buffer[i])); l <= left {
			out = append(out, f.buffer[i]...)
			left -= l
			f.buffer = f.buffer[1:]
		} else {
			out = append(out, f.buffer[i][:left]...)
			f.buffer[i] = f.buffer[i][left:]
			left = 0
		}

		if left == 0 {
			break
		}
	}

	f.transferWindow -= int64(len(out))

	if f.transferWindow > 0 {
		f.constrained = false
		debug.Printf("Stream %d is no longer constrained.\n", f.streamID)
	}

	dataFrame := new(frames.DATA)
	dataFrame.StreamID = f.streamID
	dataFrame.Data = out

	f.output <- dataFrame
}

// run is the main control path of
// the stream. It is prepared, the
// registered handler is called,
// and then the stream is cleaned
// up and closed.
func (s *ResponseStream) Run() error {
	// Catch any panics.
	defer func() {
		if v := recover(); v != nil {
			if s != nil && s.state != nil && !s.state.Closed() {
				log.Printf("Encountered stream error: %v (%[1]T)\n", v)
			}
		}
	}()

	// Make sure Request is prepared.
	if s.requestBody == nil || s.request.Body == nil {
		s.requestBody = new(bytes.Buffer)
		s.request.Body = &common.ReadCloser{s.requestBody}
	}

	// Wait until the full request has been received.
	<-s.ready

	/***************
	 *** HANDLER ***
	 ***************/
	s.handler.ServeHTTP(s, s.request)

	// Make sure any queued data has been sent.
	if err := s.flow.Wait(); err != nil {
		log.Println(err)
	}

	// Close the stream with a SYN_REPLY if
	// none has been sent, or an empty DATA
	// frame, if a SYN_REPLY has been sent
	// already.
	// If the stream is already closed at
	// this end, then nothing happens.
	if !s.unidirectional {
		if s.state.OpenHere() && !s.wroteHeader {
			s.header.Set(":status", "200")
			s.header.Set(":version", "HTTP/1.1")

			// Create the response SYN_REPLY.
			synReply := new(frames.SYN_REPLY)
			synReply.Flags = common.FLAG_FIN
			synReply.StreamID = s.streamID
			synReply.Header = make(http.Header)

			for name, values := range s.header {
				for _, value := range values {
					synReply.Header.Add(name, value)
				}
				s.header.Del(name)
			}

			s.output <- synReply
		} else if s.state.OpenHere() {
			// Create the DATA.
			data := new(frames.DATA)
			data.StreamID = s.streamID
			data.Flags = common.FLAG_FIN
			data.Data = []byte{}

			s.output <- data
		}
	}

	// Clean up state.
	s.state.CloseHere()

	if s.state.Closed() {
		return s.Close()
	}

	return nil
}

// send is run in a separate goroutine. It's used
// to ensure clear interleaving of frames and to
// provide assurances of priority and structure.
func (c *Conn) send() {
	// Catch any panics.
	defer func() {
		if v := recover(); v != nil {
			if !c.Closed() {
				log.Printf("Encountered send error: %v (%[1]T)\n", v)
			}
		}
	}()

	for i := 1; ; i++ {
		if i >= 5 {
			i = 0 // Once per 5 frames, pick randomly.
		}

		var frame common.Frame
		if i == 0 { // Ignore priority.
			frame = c.selectFrameToSend(false)
		} else { // Normal selection.
			frame = c.selectFrameToSend(true)
		}

		if frame == nil {
			c.Close()
			return
		}

		// Process connection-level flow control.
		if c.Subversion > 0 {
			c.connectionWindowLock.Lock()
			if frame, ok := frame.(*frames.DATA); ok {
				size := int64(len(frame.Data))
				constrained := false
				sending := size
				if sending > c.connectionWindowSize {
					sending = c.connectionWindowSize
					constrained = true
				}
				if sending < 0 {
					sending = 0
				}

				c.connectionWindowSize -= sending

				if constrained {
					// Chop off what we can send now.
					partial := new(frames.DATA)
					partial.Flags = frame.Flags
					partial.StreamID = frame.StreamID
					partial.Data = make([]byte, int(sending))
					copy(partial.Data, frame.Data[:sending])
					frame.Data = frame.Data[sending:]

					// Buffer this frame and try again.
					if c.dataBuffer == nil {
						c.dataBuffer = []*frames.DATA{frame}
					} else {
						buffer := make([]*frames.DATA, 1, len(c.dataBuffer)+1)
						buffer[0] = frame
						buffer = append(buffer, c.dataBuffer...)
						c.dataBuffer = buffer
					}

					frame = partial
				}
			}
			c.connectionWindowLock.Unlock()
		}

		// Compress any name/value header blocks.
		err := frame.Compress(c.compressor)
		if err != nil {
			log.Printf("Error in compression: %v (type %T).\n", err, frame)
			c.Close()
			return
		}

		debug.Printf("Sending %s:\n", frame.Name())
		debug.Println(frame)

		// Leave the specifics of writing to the
		// connection up to the frame.
		c.refreshWriteTimeout()
		if _, err = frame.WriteTo(c.conn); err != nil {
			c.handleReadWriteError(err)
			return
		}
	}
}

// Request is used to make a client request.
func (c *Conn) Request(request *http.Request, receiver common.Receiver, priority common.Priority) (common.Stream, error) {
	if c.Closed() {
		return nil, common.ErrConnClosed
	}
	c.goawayLock.Lock()
	goaway := c.goawayReceived || c.goawaySent
	c.goawayLock.Unlock()
	if goaway {
		return nil, common.ErrGoaway
	}

	if c.server != nil {
		return nil, errors.New("Error: Only clients can send requests.")
	}

	// Check stream limit would allow the new stream.
	if !c.requestStreamLimit.Add() {
		return nil, errors.New("Error: Max concurrent streams limit exceeded.")
	}

	if !priority.Valid(3) {
		return nil, errors.New("Error: Priority must be in the range 0 - 7.")
	}

	url := request.URL
	if url == nil || url.Scheme == "" || url.Host == "" {
		return nil, errors.New("Error: Incomplete path provided to resource.")
	}

	// Prepare the SYN_STREAM.
	path := url.Path
	if url.RawQuery != "" {
		path += "?" + url.RawQuery
	}
	if url.Fragment != "" {
		path += "#" + url.Fragment
	}
	if !strings.HasPrefix(path, "/") {
		path = "/" + path
	}

	host := url.Host
	if len(request.Host) > 0 {
		host = request.Host
	}
	syn := new(frames.SYN_STREAM)
	syn.Priority = priority
	syn.Header = request.Header
	syn.Header.Set(":method", request.Method)
	syn.Header.Set(":path", path)
	syn.Header.Set(":version", "HTTP/1.1")
	syn.Header.Set(":host", host)
	syn.Header.Set(":scheme", url.Scheme)

	// Prepare the request body, if any.
	body := make([]*frames.DATA, 0, 1)
	if request.Body != nil {
		buf := make([]byte, 32*1024)
		n, err := request.Body.Read(buf)
		if err != nil && err != io.EOF {
			return nil, err
		}
		total := n
		for n > 0 {
			data := new(frames.DATA)
			data.Data = make([]byte, n)
			copy(data.Data, buf[:n])
			body = append(body, data)
			n, err = request.Body.Read(buf)
			if err != nil && err != io.EOF {
				return nil, err
			}
			total += n
		}

		// Half-close the stream.
		if len(body) == 0 {
			syn.Flags = common.FLAG_FIN
		} else {
			syn.Header.Set("Content-Length", fmt.Sprint(total))
			body[len(body)-1].Flags = common.FLAG_FIN
		}
		request.Body.Close()
	} else {
		syn.Flags = common.FLAG_FIN
	}

	// Send.
	c.streamCreation.Lock()
	defer c.streamCreation.Unlock()

	c.lastRequestStreamIDLock.Lock()
	if c.lastRequestStreamID == 0 {
		c.lastRequestStreamID = 1
	} else {
		c.lastRequestStreamID += 2
	}
	syn.StreamID = c.lastRequestStreamID
	c.lastRequestStreamIDLock.Unlock()
//.........这里部分代码省略.........