func MethodFromServerStream(stream ServerStream) (string, bool) {
	s, ok := transport.StreamFromContext(stream.Context())
	if !ok {
		return "", ok
	}
	return s.Method(), ok
}
MethodFromServerStream returns the method string for the input stream.
The returned string is in the format of "/service/method".
package grpc
import (
	"errors"
	"io"
	"sync"
	"time"
	"golang.org/x/net/context"
	"golang.org/x/net/trace"
	"google.golang.org/grpc/balancer"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/encoding"
	"google.golang.org/grpc/metadata"
	"google.golang.org/grpc/peer"
	"google.golang.org/grpc/stats"
	"google.golang.org/grpc/status"
	"google.golang.org/grpc/transport"
)
type StreamHandler func(srv interface{}, stream ServerStream) error
type StreamDesc struct {
	StreamName string
	Handler    StreamHandler
	ServerStreams bool
	ClientStreams bool
}
type Stream interface {
	Context() context.Context
	SendMsg(m interface{}) error
	RecvMsg(m interface{}) error
}
type ClientStream interface {
	Header() (metadata.MD, error)
	Trailer() metadata.MD
	CloseSend() error
	Stream
}
func (cc *ClientConn) NewStream(ctx context.Context, desc *StreamDesc, method string, opts ...CallOption) (ClientStream, error) {
	if cc.dopts.streamInt != nil {
		return cc.dopts.streamInt(ctx, desc, cc, method, newClientStream, opts...)
	}
	return newClientStream(ctx, desc, cc, method, opts...)
}
func NewClientStream(ctx context.Context, desc *StreamDesc, cc *ClientConn, method string, opts ...CallOption) (ClientStream, error) {
	return cc.NewStream(ctx, desc, method, opts...)
}
func newClientStream(ctx context.Context, desc *StreamDesc, cc *ClientConn, method string, opts ...CallOption) (_ ClientStream, err error) {
	var (
		t      transport.ClientTransport
		s      *transport.Stream
		done   func(balancer.DoneInfo)
		cancel context.CancelFunc
	)
	c := defaultCallInfo()
	mc := cc.GetMethodConfig(method)
	if mc.WaitForReady != nil {
		c.failFast = !*mc.WaitForReady
	}
	if mc.Timeout != nil && *mc.Timeout >= 0 {
		ctx, cancel = context.WithTimeout(ctx, *mc.Timeout)
		defer func() {
			if err != nil {
				cancel()
			}
		}()
	}
	opts = append(cc.dopts.callOptions, opts...)
	for _, o := range opts {
		if err := o.before(c); err != nil {
			return nil, toRPCErr(err)
		}
	}
	c.maxSendMessageSize = getMaxSize(mc.MaxReqSize, c.maxSendMessageSize, defaultClientMaxSendMessageSize)
	c.maxReceiveMessageSize = getMaxSize(mc.MaxRespSize, c.maxReceiveMessageSize, defaultClientMaxReceiveMessageSize)
	callHdr := &transport.CallHdr{
		Host:   cc.authority,
		Method: method,
		Flush: desc.ClientStreams,
	}
	var cp Compressor
	var comp encoding.Compressor
	if ct := c.compressorType; ct != "" {
		callHdr.SendCompress = ct
		if ct != encoding.Identity {
			comp = encoding.GetCompressor(ct)
			if comp == nil {
				return nil, status.Errorf(codes.Internal, "grpc: Compressor is not installed for requested grpc-encoding %q", ct)
			}
		}
	} else if cc.dopts.cp != nil {
		callHdr.SendCompress = cc.dopts.cp.Type()
		cp = cc.dopts.cp
	}
	if c.creds != nil {
		callHdr.Creds = c.creds
	}
	var trInfo traceInfo
	if EnableTracing {
		trInfo.tr = trace.New("grpc.Sent."+methodFamily(method), method)
		trInfo.firstLine.client = true
		if deadline, ok := ctx.Deadline(); ok {
			trInfo.firstLine.deadline = deadline.Sub(time.Now())
		}
		trInfo.tr.LazyLog(&trInfo.firstLine, false)
		ctx = trace.NewContext(ctx, trInfo.tr)
		defer func() {
			if err != nil {
				trInfo.tr.LazyPrintf("RPC: [%v]", err)
				trInfo.tr.SetError()
				trInfo.tr.Finish()
			}
		}()
	}
	ctx = newContextWithRPCInfo(ctx, c.failFast)
	sh := cc.dopts.copts.StatsHandler
	if sh != nil {
		ctx = sh.TagRPC(ctx, &stats.RPCTagInfo{FullMethodName: method, FailFast: c.failFast})
		begin := &stats.Begin{
			Client:    true,
			BeginTime: time.Now(),
			FailFast:  c.failFast,
		}
		sh.HandleRPC(ctx, begin)
		defer func() {
			if err != nil {
				end := &stats.End{
					Client: true,
					Error:  err,
				}
				sh.HandleRPC(ctx, end)
			}
		}()
	}
	for {
		select {
		case <-ctx.Done():
			return nil, toRPCErr(ctx.Err())
		default:
		}
		t, done, err = cc.getTransport(ctx, c.failFast)
		if err != nil {
			return nil, err
		}
		s, err = t.NewStream(ctx, callHdr)
		if err != nil {
			if done != nil {
				doneInfo := balancer.DoneInfo{Err: err}
				if _, ok := err.(transport.ConnectionError); ok {
					doneInfo.BytesSent = true
				}
				done(doneInfo)
				done = nil
			}
			if !c.failFast {
				continue
			}
			return nil, toRPCErr(err)
		}
		break
	}
	if peer, ok := peer.FromContext(s.Context()); ok {
		c.peer = peer
	}
	cs := &clientStream{
		opts:   opts,
		c:      c,
		desc:   desc,
		codec:  cc.dopts.codec,
		cp:     cp,
		dc:     cc.dopts.dc,
		comp:   comp,
		cancel: cancel,
		done: done,
		t:    t,
		s:    s,
		p:    &parser{r: s},
		tracing: EnableTracing,
		trInfo:  trInfo,
		statsCtx:     ctx,
		statsHandler: cc.dopts.copts.StatsHandler,
	}
	go func() {
		select {
		case <-t.Error():
		case <-cc.ctx.Done():
			cs.finish(ErrClientConnClosing)
			cs.closeTransportStream(ErrClientConnClosing)
		case <-s.Done():
			cs.finish(s.Status().Err())
			cs.closeTransportStream(nil)
		case <-s.GoAway():
			cs.finish(errConnDrain)
			cs.closeTransportStream(errConnDrain)
		case <-s.Context().Done():
			err := s.Context().Err()
			cs.finish(err)
			cs.closeTransportStream(transport.ContextErr(err))
		}
	}()
	return cs, nil
}
type clientStream struct {
	opts []CallOption
	c    *callInfo
	t    transport.ClientTransport
	s    *transport.Stream
	p    *parser
	desc *StreamDesc
	codec     Codec
	cp        Compressor
	dc        Decompressor
	comp      encoding.Compressor
	decomp    encoding.Compressor
	decompSet bool
	cancel context.CancelFunc
	tracing bool  
	mu       sync.Mutex
	done     func(balancer.DoneInfo)
	closed   bool
	finished bool
	trInfo traceInfo
	statsCtx     context.Context
	statsHandler stats.Handler
}
func (cs *clientStream) Context() context.Context {
	return cs.s.Context()
}
func (cs *clientStream) Header() (metadata.MD, error) {
	m, err := cs.s.Header()
	if err != nil {
		if _, ok := err.(transport.ConnectionError); !ok {
			cs.closeTransportStream(err)
		}
	}
	return m, err
}
func (cs *clientStream) Trailer() metadata.MD {
	return cs.s.Trailer()
}
func (cs *clientStream) SendMsg(m interface{}) (err error) {
	if cs.tracing {
		cs.mu.Lock()
		if cs.trInfo.tr != nil {
			cs.trInfo.tr.LazyLog(&payload{sent: true, msg: m}, true)
		}
		cs.mu.Unlock()
	}
	defer func() {
		if err != nil {
			cs.finish(err)
		}
		if err == nil {
			return
		}
		if err == io.EOF {
			if !cs.desc.ClientStreams && cs.desc.ServerStreams {
				err = nil
			}
			return
		}
		if _, ok := err.(transport.ConnectionError); !ok {
			cs.closeTransportStream(err)
		}
		err = toRPCErr(err)
	}()
	var outPayload *stats.OutPayload
	if cs.statsHandler != nil {
		outPayload = &stats.OutPayload{
			Client: true,
		}
	}
	hdr, data, err := encode(cs.codec, m, cs.cp, outPayload, cs.comp)
	if err != nil {
		return err
	}
	if cs.c.maxSendMessageSize == nil {
		return status.Errorf(codes.Internal, "callInfo maxSendMessageSize field uninitialized(nil)")
	}
	if len(data) > *cs.c.maxSendMessageSize {
		return status.Errorf(codes.ResourceExhausted, "trying to send message larger than max (%d vs. %d)", len(data), *cs.c.maxSendMessageSize)
	}
	err = cs.t.Write(cs.s, hdr, data, &transport.Options{Last: false})
	if err == nil && outPayload != nil {
		outPayload.SentTime = time.Now()
		cs.statsHandler.HandleRPC(cs.statsCtx, outPayload)
	}
	return err
}
func (cs *clientStream) RecvMsg(m interface{}) (err error) {
	var inPayload *stats.InPayload
	if cs.statsHandler != nil {
		inPayload = &stats.InPayload{
			Client: true,
		}
	}
	if cs.c.maxReceiveMessageSize == nil {
		return status.Errorf(codes.Internal, "callInfo maxReceiveMessageSize field uninitialized(nil)")
	}
	if !cs.decompSet {
		if ct := cs.s.RecvCompress(); ct != "" && ct != encoding.Identity {
			if cs.dc == nil || cs.dc.Type() != ct {
				cs.dc = nil
				cs.decomp = encoding.GetCompressor(ct)
			}
		} else {
			cs.dc = nil
		}
		cs.decompSet = true
	}
	err = recv(cs.p, cs.codec, cs.s, cs.dc, m, *cs.c.maxReceiveMessageSize, inPayload, cs.decomp)
	defer func() {
		if err != nil {
			cs.finish(err)
		}
	}()
	if err == nil {
		if cs.tracing {
			cs.mu.Lock()
			if cs.trInfo.tr != nil {
				cs.trInfo.tr.LazyLog(&payload{sent: false, msg: m}, true)
			}
			cs.mu.Unlock()
		}
		if inPayload != nil {
			cs.statsHandler.HandleRPC(cs.statsCtx, inPayload)
		}
		if !cs.desc.ClientStreams || cs.desc.ServerStreams {
			return
		}
		if cs.c.maxReceiveMessageSize == nil {
			return status.Errorf(codes.Internal, "callInfo maxReceiveMessageSize field uninitialized(nil)")
		}
		err = recv(cs.p, cs.codec, cs.s, cs.dc, m, *cs.c.maxReceiveMessageSize, nil, cs.decomp)
		cs.closeTransportStream(err)
		if err == nil {
			return toRPCErr(errors.New("grpc: client streaming protocol violation: get <nil>, want <EOF>"))
		}
		if err == io.EOF {
			if se := cs.s.Status().Err(); se != nil {
				return se
			}
			cs.finish(err)
			return nil
		}
		return toRPCErr(err)
	}
	if _, ok := err.(transport.ConnectionError); !ok {
		cs.closeTransportStream(err)
	}
	if err == io.EOF {
		if statusErr := cs.s.Status().Err(); statusErr != nil {
			return statusErr
		}
		return
	}
	return toRPCErr(err)
}
func (cs *clientStream) CloseSend() (err error) {
	err = cs.t.Write(cs.s, nil, nil, &transport.Options{Last: true})
	defer func() {
		if err != nil {
			cs.finish(err)
		}
	}()
	if err == nil || err == io.EOF {
		return nil
	}
	if _, ok := err.(transport.ConnectionError); !ok {
		cs.closeTransportStream(err)
	}
	err = toRPCErr(err)
	return
}
func (cs *clientStream) closeTransportStream(err error) {
	cs.mu.Lock()
	if cs.closed {
		cs.mu.Unlock()
		return
	}
	cs.closed = true
	cs.mu.Unlock()
	cs.t.CloseStream(cs.s, err)
}
func (cs *clientStream) finish(err error) {
	cs.mu.Lock()
	defer cs.mu.Unlock()
	if cs.finished {
		return
	}
	cs.finished = true
	defer func() {
		if cs.cancel != nil {
			cs.cancel()
		}
	}()
	for _, o := range cs.opts {
		o.after(cs.c)
	}
	if cs.done != nil {
		cs.done(balancer.DoneInfo{
			Err:           err,
			BytesSent:     true,
			BytesReceived: cs.s.BytesReceived(),
		})
		cs.done = nil
	}
	if cs.statsHandler != nil {
		end := &stats.End{
			Client:  true,
			EndTime: time.Now(),
		}
		if err != io.EOF {
			end.Error = toRPCErr(err)
		}
		cs.statsHandler.HandleRPC(cs.statsCtx, end)
	}
	if !cs.tracing {
		return
	}
	if cs.trInfo.tr != nil {
		if err == nil || err == io.EOF {
			cs.trInfo.tr.LazyPrintf("RPC: [OK]")
		} else {
			cs.trInfo.tr.LazyPrintf("RPC: [%v]", err)
			cs.trInfo.tr.SetError()
		}
		cs.trInfo.tr.Finish()
		cs.trInfo.tr = nil
	}
}
type ServerStream interface {
	SetHeader(metadata.MD) error
	SendHeader(metadata.MD) error
	SetTrailer(metadata.MD)
	Stream
}
type serverStream struct {
	t     transport.ServerTransport
	s     *transport.Stream
	p     *parser
	codec Codec
	cp     Compressor
	dc     Decompressor
	comp   encoding.Compressor
	decomp encoding.Compressor
	maxReceiveMessageSize int
	maxSendMessageSize    int
	trInfo                *traceInfo
	statsHandler stats.Handler
	mu sync.Mutex  
}
func (ss *serverStream) Context() context.Context {
	return ss.s.Context()
}
func (ss *serverStream) SetHeader(md metadata.MD) error {
	if md.Len() == 0 {
		return nil
	}
	return ss.s.SetHeader(md)
}
func (ss *serverStream) SendHeader(md metadata.MD) error {
	return ss.t.WriteHeader(ss.s, md)
}
func (ss *serverStream) SetTrailer(md metadata.MD) {
	if md.Len() == 0 {
		return
	}
	ss.s.SetTrailer(md)
	return
}
func (ss *serverStream) SendMsg(m interface{}) (err error) {
	defer func() {
		if ss.trInfo != nil {
			ss.mu.Lock()
			if ss.trInfo.tr != nil {
				if err == nil {
					ss.trInfo.tr.LazyLog(&payload{sent: true, msg: m}, true)
				} else {
					ss.trInfo.tr.LazyLog(&fmtStringer{"%v", []interface{}{err}}, true)
					ss.trInfo.tr.SetError()
				}
			}
			ss.mu.Unlock()
		}
		if err != nil && err != io.EOF {
			st, _ := status.FromError(toRPCErr(err))
			ss.t.WriteStatus(ss.s, st)
		}
	}()
	var outPayload *stats.OutPayload
	if ss.statsHandler != nil {
		outPayload = &stats.OutPayload{}
	}
	hdr, data, err := encode(ss.codec, m, ss.cp, outPayload, ss.comp)
	if err != nil {
		return err
	}
	if len(data) > ss.maxSendMessageSize {
		return status.Errorf(codes.ResourceExhausted, "trying to send message larger than max (%d vs. %d)", len(data), ss.maxSendMessageSize)
	}
	if err := ss.t.Write(ss.s, hdr, data, &transport.Options{Last: false}); err != nil {
		return toRPCErr(err)
	}
	if outPayload != nil {
		outPayload.SentTime = time.Now()
		ss.statsHandler.HandleRPC(ss.s.Context(), outPayload)
	}
	return nil
}
func (ss *serverStream) RecvMsg(m interface{}) (err error) {
	defer func() {
		if ss.trInfo != nil {
			ss.mu.Lock()
			if ss.trInfo.tr != nil {
				if err == nil {
					ss.trInfo.tr.LazyLog(&payload{sent: false, msg: m}, true)
				} else if err != io.EOF {
					ss.trInfo.tr.LazyLog(&fmtStringer{"%v", []interface{}{err}}, true)
					ss.trInfo.tr.SetError()
				}
			}
			ss.mu.Unlock()
		}
		if err != nil && err != io.EOF {
			st, _ := status.FromError(toRPCErr(err))
			ss.t.WriteStatus(ss.s, st)
		}
	}()
	var inPayload *stats.InPayload
	if ss.statsHandler != nil {
		inPayload = &stats.InPayload{}
	}
	if err := recv(ss.p, ss.codec, ss.s, ss.dc, m, ss.maxReceiveMessageSize, inPayload, ss.decomp); err != nil {
		if err == io.EOF {
			return err
		}
		if err == io.ErrUnexpectedEOF {
			err = status.Errorf(codes.Internal, io.ErrUnexpectedEOF.Error())
		}
		return toRPCErr(err)
	}
	if inPayload != nil {
		ss.statsHandler.HandleRPC(ss.s.Context(), inPayload)
	}
	return nil
}