func newContextWithStream(ctx context.Context, stream *Stream) context.Context {
	return context.WithValue(ctx, streamKey{}, stream)
}
newContextWithStream creates a new context from ctx and attaches stream
to it.
package transport  
import (
	"bytes"
	"errors"
	"fmt"
	"io"
	"net"
	"sync"
	"golang.org/x/net/context"
	"golang.org/x/net/trace"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/credentials"
	"google.golang.org/grpc/metadata"
)
type recvMsg struct {
	data []byte
	err error
}
func (*recvMsg) item() {}
type item interface {
	item()
}
type recvBuffer struct {
	c       chan item
	mu      sync.Mutex
	backlog []item
}
func newRecvBuffer() *recvBuffer {
	b := &recvBuffer{
		c: make(chan item, 1),
	}
	return b
}
func (b *recvBuffer) put(r item) {
	b.mu.Lock()
	defer b.mu.Unlock()
	if len(b.backlog) == 0 {
		select {
		case b.c <- r:
			return
		default:
		}
	}
	b.backlog = append(b.backlog, r)
}
func (b *recvBuffer) load() {
	b.mu.Lock()
	defer b.mu.Unlock()
	if len(b.backlog) > 0 {
		select {
		case b.c <- b.backlog[0]:
			b.backlog = b.backlog[1:]
		default:
		}
	}
}
func (b *recvBuffer) get() <-chan item {
	return b.c
}
type recvBufferReader struct {
	ctx    context.Context
	goAway chan struct{}
	recv   *recvBuffer
	last   *bytes.Reader  
	err    error
}
func (r *recvBufferReader) Read(p []byte) (n int, err error) {
	if r.err != nil {
		return 0, r.err
	}
	defer func() { r.err = err }()
	if r.last != nil && r.last.Len() > 0 {
		return r.last.Read(p)
	}
	select {
	case <-r.ctx.Done():
		return 0, ContextErr(r.ctx.Err())
	case <-r.goAway:
		return 0, ErrStreamDrain
	case i := <-r.recv.get():
		r.recv.load()
		m := i.(*recvMsg)
		if m.err != nil {
			return 0, m.err
		}
		r.last = bytes.NewReader(m.data)
		return r.last.Read(p)
	}
}
type streamState uint8
const (
	streamActive    streamState = iota
	streamWriteDone              
	streamReadDone               
	streamDone                   
)
type Stream struct {
	id uint32
	st ServerTransport
	ctx    context.Context
	cancel context.CancelFunc
	done chan struct{}
	goAway chan struct{}
	method       string
	recvCompress string
	sendCompress string
	buf          *recvBuffer
	dec          io.Reader
	fc           *inFlow
	recvQuota    uint32
	updateQuota uint32
	windowHandler func(int)
	sendQuotaPool *quotaPool
	headerChan chan struct{}
	header metadata.MD
	trailer metadata.MD
	mu sync.RWMutex  
	headerOk bool
	state    streamState
	headerDone bool
	statusCode codes.Code
	statusDesc string
}
func (s *Stream) RecvCompress() string {
	return s.recvCompress
}
func (s *Stream) SetSendCompress(str string) {
	s.sendCompress = str
}
func (s *Stream) Done() <-chan struct{} {
	return s.done
}
func (s *Stream) GoAway() <-chan struct{} {
	return s.goAway
}
func (s *Stream) Header() (metadata.MD, error) {
	select {
	case <-s.ctx.Done():
		return nil, ContextErr(s.ctx.Err())
	case <-s.goAway:
		return nil, ErrStreamDrain
	case <-s.headerChan:
		return s.header.Copy(), nil
	}
}
func (s *Stream) Trailer() metadata.MD {
	s.mu.RLock()
	defer s.mu.RUnlock()
	return s.trailer.Copy()
}
func (s *Stream) ServerTransport() ServerTransport {
	return s.st
}
func (s *Stream) Context() context.Context {
	return s.ctx
}
func (s *Stream) TraceContext(tr trace.Trace) {
	s.ctx = trace.NewContext(s.ctx, tr)
}
func (s *Stream) Method() string {
	return s.method
}
func (s *Stream) StatusCode() codes.Code {
	return s.statusCode
}
func (s *Stream) StatusDesc() string {
	return s.statusDesc
}
var ErrIllegalTrailerSet = errors.New("transport: trailer has been set")
func (s *Stream) SetTrailer(md metadata.MD) error {
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.trailer != nil {
		return ErrIllegalTrailerSet
	}
	s.trailer = md.Copy()
	return nil
}
func (s *Stream) write(m recvMsg) {
	s.buf.put(&m)
}
func (s *Stream) Read(p []byte) (n int, err error) {
	n, err = s.dec.Read(p)
	if err != nil {
		return
	}
	s.windowHandler(n)
	return
}
type streamKey struct{}