func (s *Server) handleStream(t transport.ServerTransport, stream *transport.Stream) {
	sm := stream.Method()
	if sm != "" && sm[0] == '/' {
		sm = sm[1:]
	}
	pos := strings.LastIndex(sm, "/")
	if pos == -1 {
		if err := t.WriteStatus(stream, codes.InvalidArgument, fmt.Sprintf("malformed method name: %q", stream.Method())); err != nil {
			grpclog.Printf("grpc: Server.handleStream failed to write status: %v", err)
		}
		return
	}
	service := sm[:pos]
	method := sm[pos+1:]
	srv, ok := s.m[service]
	if !ok {
		if err := t.WriteStatus(stream, codes.Unimplemented, fmt.Sprintf("unknown service %v", service)); err != nil {
			grpclog.Printf("grpc: Server.handleStream failed to write status: %v", err)
		}
		return
	}
	// Unary RPC or Streaming RPC?
	if md, ok := srv.md[method]; ok {
		s.processUnaryRPC(t, stream, srv, md)
		return
	}
	if sd, ok := srv.sd[method]; ok {
		s.processStreamingRPC(t, stream, srv, sd)
		return
	}
	if err := t.WriteStatus(stream, codes.Unimplemented, fmt.Sprintf("unknown method %v", method)); err != nil {
		grpclog.Printf("grpc: Server.handleStream failed to write status: %v", err)
	}
}

func doClientStreaming(tc testpb.TestServiceClient) {
	stream, err := tc.StreamingInputCall(context.Background())
	if err != nil {
		grpclog.Fatalf("%v.StreamingInputCall(_) = _, %v", tc, err)
	}
	var sum int
	for _, s := range reqSizes {
		pl := newPayload(testpb.PayloadType_COMPRESSABLE, s)
		req := &testpb.StreamingInputCallRequest{
			Payload: pl,
		}
		if err := stream.Send(req); err != nil {
			grpclog.Fatalf("%v.Send(%v) = %v", stream, req, err)
		}
		sum += s
		grpclog.Printf("Sent a request of size %d, aggregated size %d", s, sum)

	}
	reply, err := stream.CloseAndRecv()
	if err != nil {
		grpclog.Fatalf("%v.CloseAndRecv() got error %v, want %v", stream, err, nil)
	}
	if reply.GetAggregatedPayloadSize() != int32(sum) {
		grpclog.Fatalf("%v.CloseAndRecv().GetAggregatePayloadSize() = %v; want %v", stream, reply.GetAggregatedPayloadSize(), sum)
	}
	grpclog.Println("ClientStreaming done")
}

// Dial creates a client connection the given target.
func Dial(target string, opts ...DialOption) (*ClientConn, error) {
	if target == "" {
		return nil, ErrUnspecTarget
	}
	cc := &ClientConn{
		target:       target,
		shutdownChan: make(chan struct{}),
	}
	for _, opt := range opts {
		opt(&cc.dopts)
	}
	if !cc.dopts.insecure {
		var ok bool
		for _, c := range cc.dopts.copts.AuthOptions {
			if _, ok := c.(credentials.TransportAuthenticator); !ok {
				continue
			}
			ok = true
		}
		if !ok {
			return nil, ErrNoTransportSecurity
		}
	} else {
		for _, c := range cc.dopts.copts.AuthOptions {
			if c.RequireTransportSecurity() {
				return nil, ErrCredentialsMisuse
			}
		}
	}
	colonPos := strings.LastIndex(target, ":")
	if colonPos == -1 {
		colonPos = len(target)
	}
	cc.authority = target[:colonPos]
	if cc.dopts.codec == nil {
		// Set the default codec.
		cc.dopts.codec = protoCodec{}
	}
	cc.stateCV = sync.NewCond(&cc.mu)
	if cc.dopts.block {
		if err := cc.resetTransport(false); err != nil {
			cc.Close()
			return nil, err
		}
		// Start to monitor the error status of transport.
		go cc.transportMonitor()
	} else {
		// Start a goroutine connecting to the server asynchronously.
		go func() {
			if err := cc.resetTransport(false); err != nil {
				grpclog.Printf("Failed to dial %s: %v; please retry.", target, err)
				cc.Close()
				return
			}
			go cc.transportMonitor()
		}()
	}
	return cc, nil
}

// printFeature gets the feature for the given point.
func printFeature(client pb.RouteGuideClient, point *pb.Point) {
	grpclog.Printf("Getting feature for point (%d, %d)", point.Latitude, point.Longitude)
	feature, err := client.GetFeature(context.Background(), point)
	if err != nil {
		grpclog.Fatalf("%v.GetFeatures(_) = _, %v: ", client, err)
	}
	grpclog.Println(feature)
}

func (t *http2Client) notifyError(err error) {
	t.mu.Lock()
	defer t.mu.Unlock()
	// make sure t.errorChan is closed only once.
	if t.state == reachable {
		t.state = unreachable
		close(t.errorChan)
		grpclog.Printf("transport: http2Client.notifyError got notified that the client transport was broken %v.", err)
	}
}

func newHPACKDecoder() *hpackDecoder {
	d := &hpackDecoder{}
	d.h = hpack.NewDecoder(http2InitHeaderTableSize, func(f hpack.HeaderField) {
		switch f.Name {
		case "content-type":
			if !strings.Contains(f.Value, "application/grpc") {
				d.err = StreamErrorf(codes.FailedPrecondition, "transport: received the unexpected header")
				return
			}
		case "grpc-status":
			code, err := strconv.Atoi(f.Value)
			if err != nil {
				d.err = StreamErrorf(codes.Internal, "transport: malformed grpc-status: %v", err)
				return
			}
			d.state.statusCode = codes.Code(code)
		case "grpc-message":
			d.state.statusDesc = f.Value
		case "grpc-timeout":
			d.state.timeoutSet = true
			var err error
			d.state.timeout, err = timeoutDecode(f.Value)
			if err != nil {
				d.err = StreamErrorf(codes.Internal, "transport: malformed time-out: %v", err)
				return
			}
		case ":path":
			d.state.method = f.Value
		default:
			if !isReservedHeader(f.Name) {
				if f.Name == "user-agent" {
					i := strings.LastIndex(f.Value, " ")
					if i == -1 {
						// There is no application user agent string being set.
						return
					}
					// Extract the application user agent string.
					f.Value = f.Value[:i]
				}
				if d.state.mdata == nil {
					d.state.mdata = make(map[string][]string)
				}
				k, v, err := metadata.DecodeKeyValue(f.Name, f.Value)
				if err != nil {
					grpclog.Printf("Failed to decode (%q, %q): %v", f.Name, f.Value, err)
					return
				}
				d.state.mdata[k] = append(d.state.mdata[k], v)
			}
		}
	})
	return d
}

// NewConn creates a Conn.
func NewConn(cc *ClientConn) (*Conn, error) {
	if cc.target == "" {
		return nil, ErrUnspecTarget
	}
	c := &Conn{
		target:       cc.target,
		dopts:        cc.dopts,
		resetChan:    make(chan int, 1),
		shutdownChan: make(chan struct{}),
	}
	if EnableTracing {
		c.events = trace.NewEventLog("grpc.ClientConn", c.target)
	}
	if !c.dopts.insecure {
		var ok bool
		for _, cd := range c.dopts.copts.AuthOptions {
			if _, ok := cd.(credentials.TransportAuthenticator); !ok {
				continue
			}
			ok = true
		}
		if !ok {
			return nil, ErrNoTransportSecurity
		}
	} else {
		for _, cd := range c.dopts.copts.AuthOptions {
			if cd.RequireTransportSecurity() {
				return nil, ErrCredentialsMisuse
			}
		}
	}
	c.stateCV = sync.NewCond(&c.mu)
	if c.dopts.block {
		if err := c.resetTransport(false); err != nil {
			c.Close()
			return nil, err
		}
		// Start to monitor the error status of transport.
		go c.transportMonitor()
	} else {
		// Start a goroutine connecting to the server asynchronously.
		go func() {
			if err := c.resetTransport(false); err != nil {
				grpclog.Printf("Failed to dial %s: %v; please retry.", c.target, err)
				c.Close()
				return
			}
			c.transportMonitor()
		}()
	}
	return c, nil
}

// reader runs as a separate goroutine in charge of reading data from network
// connection.
//
// TODO(zhaoq): currently one reader per transport. Investigate whether this is
// optimal.
// TODO(zhaoq): Check the validity of the incoming frame sequence.
func (t *http2Client) reader() {
	// Check the validity of server preface.
	frame, err := t.framer.readFrame()
	if err != nil {
		t.notifyError(err)
		return
	}
	sf, ok := frame.(*http2.SettingsFrame)
	if !ok {
		t.notifyError(err)
		return
	}
	t.handleSettings(sf)

	hDec := newHPACKDecoder()
	var curStream *Stream
	// loop to keep reading incoming messages on this transport.
	for {
		frame, err := t.framer.readFrame()
		if err != nil {
			t.notifyError(err)
			return
		}
		switch frame := frame.(type) {
		case *http2.HeadersFrame:
			// operateHeaders has to be invoked regardless the value of curStream
			// because the HPACK decoder needs to be updated using the received
			// headers.
			curStream, _ = t.getStream(frame)
			endStream := frame.Header().Flags.Has(http2.FlagHeadersEndStream)
			curStream = t.operateHeaders(hDec, curStream, frame, endStream)
		case *http2.ContinuationFrame:
			curStream = t.operateHeaders(hDec, curStream, frame, frame.HeadersEnded())
		case *http2.DataFrame:
			t.handleData(frame)
		case *http2.RSTStreamFrame:
			t.handleRSTStream(frame)
		case *http2.SettingsFrame:
			t.handleSettings(frame)
		case *http2.PingFrame:
			t.handlePing(frame)
		case *http2.GoAwayFrame:
			t.handleGoAway(frame)
		case *http2.WindowUpdateFrame:
			t.handleWindowUpdate(frame)
		default:
			grpclog.Printf("transport: http2Client.reader got unhandled frame type %v.", frame)
		}
	}
}

// runRecordRoute sends a sequence of points to server and expects to get a RouteSummary from server.
func runRecordRoute(client pb.RouteGuideClient) {
	// Create a random number of random points
	r := rand.New(rand.NewSource(time.Now().UnixNano()))
	pointCount := int(r.Int31n(100)) + 2 // Traverse at least two points
	var points []*pb.Point
	for i := 0; i < pointCount; i++ {
		points = append(points, randomPoint(r))
	}
	grpclog.Printf("Traversing %d points.", len(points))
	stream, err := client.RecordRoute(context.Background())
	if err != nil {
		grpclog.Fatalf("%v.RecordRoute(_) = _, %v", client, err)
	}
	for _, point := range points {
		if err := stream.Send(point); err != nil {
			grpclog.Fatalf("%v.Send(%v) = %v", stream, point, err)
		}
	}
	reply, err := stream.CloseAndRecv()
	if err != nil {
		grpclog.Fatalf("%v.CloseAndRecv() got error %v, want %v", stream, err, nil)
	}
	grpclog.Printf("Route summary: %v", reply)
}

// printFeatures lists all the features within the given bounding Rectangle.
func printFeatures(client pb.RouteGuideClient, rect *pb.Rectangle) {
	grpclog.Printf("Looking for features within %v", rect)
	stream, err := client.ListFeatures(context.Background(), rect)
	if err != nil {
		grpclog.Fatalf("%v.ListFeatures(_) = _, %v", client, err)
	}
	for {
		feature, err := stream.Recv()
		if err == io.EOF {
			break
		}
		if err != nil {
			grpclog.Fatalf("%v.ListFeatures(_) = _, %v", client, err)
		}
		grpclog.Println(feature)
	}
}

func (cc *Conn) reconnect() bool {
	cc.mu.Lock()
	if cc.state == Shutdown {
		// cc.Close() has been invoked.
		cc.mu.Unlock()
		return false
	}
	cc.state = TransientFailure
	cc.stateCV.Broadcast()
	cc.mu.Unlock()
	if err := cc.resetTransport(true); err != nil {
		// The ClientConn is closing.
		cc.mu.Lock()
		cc.printf("transport exiting: %v", err)
		cc.mu.Unlock()
		grpclog.Printf("grpc: Conn.transportMonitor exits due to: %v", err)
		return false
	}
	return true
}

// Run in a goroutine to track the error in transport and create the
// new transport if an error happens. It returns when the channel is closing.
func (cc *ClientConn) transportMonitor() {
	for {
		select {
		// shutdownChan is needed to detect the teardown when
		// the ClientConn is idle (i.e., no RPC in flight).
		case <-cc.shutdownChan:
			return
		case <-cc.transport.Error():
			cc.mu.Lock()
			cc.state = TransientFailure
			cc.stateCV.Broadcast()
			cc.mu.Unlock()
			if err := cc.resetTransport(true); err != nil {
				// The ClientConn is closing.
				grpclog.Printf("grpc: ClientConn.transportMonitor exits due to: %v", err)
				return
			}
			continue
		}
	}
}

func (t *http2Server) handleData(f *http2.DataFrame) {
	// Select the right stream to dispatch.
	s, ok := t.getStream(f)
	if !ok {
		return
	}
	size := len(f.Data())
	if size > 0 {
		if err := s.fc.onData(uint32(size)); err != nil {
			if _, ok := err.(ConnectionError); ok {
				grpclog.Printf("transport: http2Server %v", err)
				t.Close()
				return
			}
			t.closeStream(s)
			t.controlBuf.put(&resetStream{s.id, http2.ErrCodeFlowControl})
			return
		}
		// TODO(bradfitz, zhaoq): A copy is required here because there is no
		// guarantee f.Data() is consumed before the arrival of next frame.
		// Can this copy be eliminated?
		data := make([]byte, size)
		copy(data, f.Data())
		s.write(recvMsg{data: data})
	}
	if f.Header().Flags.Has(http2.FlagDataEndStream) {
		// Received the end of stream from the client.
		s.mu.Lock()
		if s.state != streamDone {
			if s.state == streamWriteDone {
				s.state = streamDone
			} else {
				s.state = streamReadDone
			}
		}
		s.mu.Unlock()
		s.write(recvMsg{err: io.EOF})
	}
}

// controller running in a separate goroutine takes charge of sending control
// frames (e.g., window update, reset stream, setting, etc.) to the server.
func (t *http2Server) controller() {
	for {
		select {
		case i := <-t.controlBuf.get():
			t.controlBuf.load()
			select {
			case <-t.writableChan:
				switch i := i.(type) {
				case *windowUpdate:
					t.framer.writeWindowUpdate(true, i.streamID, i.increment)
				case *settings:
					if i.ack {
						t.framer.writeSettingsAck(true)
						t.applySettings(i.ss)
					} else {
						t.framer.writeSettings(true, i.ss...)
					}
				case *resetStream:
					t.framer.writeRSTStream(true, i.streamID, i.code)
				case *flushIO:
					t.framer.flushWrite()
				case *ping:
					// TODO(zhaoq): Ack with all-0 data now. will change to some
					// meaningful content when this is actually in use.
					t.framer.writePing(true, i.ack, [8]byte{})
				default:
					grpclog.Printf("transport: http2Server.controller got unexpected item type %v\n", i)
				}
				t.writableChan <- 0
				continue
			case <-t.shutdownChan:
				return
			}
		case <-t.shutdownChan:
			return
		}
	}
}

// controller running in a separate goroutine takes charge of sending control
// frames (e.g., window update, reset stream, setting, etc.) to the server.
func (t *http2Server) controller() {
	for {
		select {
		case i := <-t.controlBuf.get():
			t.controlBuf.load()
			select {
			case <-t.writableChan:
				switch i := i.(type) {
				case *windowUpdate:
					t.framer.writeWindowUpdate(true, i.streamID, i.increment)
				case *settings:
					if i.ack {
						t.framer.writeSettingsAck(true)
						t.applySettings(i.ss)
					} else {
						t.framer.writeSettings(true, i.ss...)
					}
				case *resetStream:
					t.framer.writeRSTStream(true, i.streamID, i.code)
				case *flushIO:
					t.framer.flushWrite()
				case *ping:
					t.framer.writePing(true, i.ack, i.data)
				default:
					grpclog.Printf("transport: http2Server.controller got unexpected item type %v\n", i)
				}
				t.writableChan <- 0
				continue
			case <-t.shutdownChan:
				return
			}
		case <-t.shutdownChan:
			return
		}
	}
}

// runRouteChat receives a sequence of route notes, while sending notes for various locations.
func runRouteChat(client pb.RouteGuideClient) {
	notes := []*pb.RouteNote{
		{&pb.Point{0, 1}, "First message"},
		{&pb.Point{0, 2}, "Second message"},
		{&pb.Point{0, 3}, "Third message"},
		{&pb.Point{0, 1}, "Fourth message"},
		{&pb.Point{0, 2}, "Fifth message"},
		{&pb.Point{0, 3}, "Sixth message"},
	}
	stream, err := client.RouteChat(context.Background())
	if err != nil {
		grpclog.Fatalf("%v.RouteChat(_) = _, %v", client, err)
	}
	waitc := make(chan struct{})
	go func() {
		for {
			in, err := stream.Recv()
			if err == io.EOF {
				// read done.
				close(waitc)
				return
			}
			if err != nil {
				grpclog.Fatalf("Failed to receive a note : %v", err)
			}
			grpclog.Printf("Got message %s at point(%d, %d)", in.Message, in.Location.Latitude, in.Location.Longitude)
		}
	}()
	for _, note := range notes {
		if err := stream.Send(note); err != nil {
			grpclog.Fatalf("Failed to send a note: %v", err)
		}
	}
	stream.CloseSend()
	<-waitc
}

func (cc *Conn) resetTransport(closeTransport bool) error {
	var retries int
	start := time.Now()
	for {
		cc.mu.Lock()
		cc.printf("connecting")
		if cc.state == Shutdown {
			// cc.Close() has been invoked.
			cc.mu.Unlock()
			return ErrClientConnClosing
		}
		cc.state = Connecting
		cc.stateCV.Broadcast()
		cc.mu.Unlock()
		if closeTransport {
			cc.transport.Close()
		}
		// Adjust timeout for the current try.
		copts := cc.dopts.copts
		if copts.Timeout < 0 {
			cc.Close()
			return ErrClientConnTimeout
		}
		if copts.Timeout > 0 {
			copts.Timeout -= time.Since(start)
			if copts.Timeout <= 0 {
				cc.Close()
				return ErrClientConnTimeout
			}
		}
		sleepTime := backoff(retries)
		timeout := sleepTime
		if timeout < minConnectTimeout {
			timeout = minConnectTimeout
		}
		if copts.Timeout == 0 || copts.Timeout > timeout {
			copts.Timeout = timeout
		}
		connectTime := time.Now()
		addr, err := cc.dopts.picker.PickAddr()
		var newTransport transport.ClientTransport
		if err == nil {
			newTransport, err = transport.NewClientTransport(addr, &copts)
		}
		if err != nil {
			cc.mu.Lock()
			if cc.state == Shutdown {
				// cc.Close() has been invoked.
				cc.mu.Unlock()
				return ErrClientConnClosing
			}
			cc.errorf("transient failure: %v", err)
			cc.state = TransientFailure
			cc.stateCV.Broadcast()
			if cc.ready != nil {
				close(cc.ready)
				cc.ready = nil
			}
			cc.mu.Unlock()
			sleepTime -= time.Since(connectTime)
			if sleepTime < 0 {
				sleepTime = 0
			}
			// Fail early before falling into sleep.
			if cc.dopts.copts.Timeout > 0 && cc.dopts.copts.Timeout < sleepTime+time.Since(start) {
				cc.mu.Lock()
				cc.errorf("connection timeout")
				cc.mu.Unlock()
				cc.Close()
				return ErrClientConnTimeout
			}
			closeTransport = false
			time.Sleep(sleepTime)
			retries++
			grpclog.Printf("grpc: Conn.resetTransport failed to create client transport: %v; Reconnecting to %q", err, cc.target)
			continue
		}
		cc.mu.Lock()
		cc.printf("ready")
		if cc.state == Shutdown {
			// cc.Close() has been invoked.
			cc.mu.Unlock()
			newTransport.Close()
			return ErrClientConnClosing
		}
		cc.state = Ready
		cc.stateCV.Broadcast()
		cc.transport = newTransport
		if cc.ready != nil {
			close(cc.ready)
			cc.ready = nil
		}
		cc.mu.Unlock()
		return nil
	}
}

// HandleStreams receives incoming streams using the given handler. This is
// typically run in a separate goroutine.
func (t *http2Server) HandleStreams(handle func(*Stream)) {
	// Check the validity of client preface.
	preface := make([]byte, len(clientPreface))
	if _, err := io.ReadFull(t.conn, preface); err != nil {
		grpclog.Printf("transport: http2Server.HandleStreams failed to receive the preface from client: %v", err)
		t.Close()
		return
	}
	if !bytes.Equal(preface, clientPreface) {
		grpclog.Printf("transport: http2Server.HandleStreams received bogus greeting from client: %q", preface)
		t.Close()
		return
	}

	frame, err := t.framer.readFrame()
	if err != nil {
		grpclog.Printf("transport: http2Server.HandleStreams failed to read frame: %v", err)
		t.Close()
		return
	}
	sf, ok := frame.(*http2.SettingsFrame)
	if !ok {
		grpclog.Printf("transport: http2Server.HandleStreams saw invalid preface type %T from client", frame)
		t.Close()
		return
	}
	t.handleSettings(sf)

	hDec := newHPACKDecoder()
	var curStream *Stream
	var wg sync.WaitGroup
	defer wg.Wait()
	for {
		frame, err := t.framer.readFrame()
		if err != nil {
			t.Close()
			return
		}
		switch frame := frame.(type) {
		case *http2.HeadersFrame:
			id := frame.Header().StreamID
			if id%2 != 1 || id <= t.maxStreamID {
				// illegal gRPC stream id.
				grpclog.Println("transport: http2Server.HandleStreams received an illegal stream id: ", id)
				t.Close()
				break
			}
			t.maxStreamID = id
			buf := newRecvBuffer()
			fc := &inFlow{
				limit: initialWindowSize,
				conn:  t.fc,
			}
			curStream = &Stream{
				id:  frame.Header().StreamID,
				st:  t,
				buf: buf,
				fc:  fc,
			}
			endStream := frame.Header().Flags.Has(http2.FlagHeadersEndStream)
			curStream = t.operateHeaders(hDec, curStream, frame, endStream, handle, &wg)
		case *http2.ContinuationFrame:
			curStream = t.operateHeaders(hDec, curStream, frame, false, handle, &wg)
		case *http2.DataFrame:
			t.handleData(frame)
		case *http2.RSTStreamFrame:
			t.handleRSTStream(frame)
		case *http2.SettingsFrame:
			t.handleSettings(frame)
		case *http2.PingFrame:
			t.handlePing(frame)
		case *http2.WindowUpdateFrame:
			t.handleWindowUpdate(frame)
		case *http2.GoAwayFrame:
			break
		default:
			grpclog.Printf("transport: http2Server.HandleStreams found unhandled frame type %v.", frame)
		}
	}
}

// operateHeader takes action on the decoded headers. It returns the current
// stream if there are remaining headers on the wire (in the following
// Continuation frame).
func (t *http2Server) operateHeaders(hDec *hpackDecoder, s *Stream, frame headerFrame, endStream bool, handle func(*Stream), wg *sync.WaitGroup) (pendingStream *Stream) {
	defer func() {
		if pendingStream == nil {
			hDec.state = decodeState{}
		}
	}()
	endHeaders, err := hDec.decodeServerHTTP2Headers(frame)
	if s == nil {
		// s has been closed.
		return nil
	}
	if err != nil {
		grpclog.Printf("transport: http2Server.operateHeader found %v", err)
		if se, ok := err.(StreamError); ok {
			t.controlBuf.put(&resetStream{s.id, statusCodeConvTab[se.Code]})
		}
		return nil
	}
	if endStream {
		// s is just created by the caller. No lock needed.
		s.state = streamReadDone
	}
	if !endHeaders {
		return s
	}
	t.mu.Lock()
	if t.state != reachable {
		t.mu.Unlock()
		return nil
	}
	if uint32(len(t.activeStreams)) >= t.maxStreams {
		t.mu.Unlock()
		t.controlBuf.put(&resetStream{s.id, http2.ErrCodeRefusedStream})
		return nil
	}
	s.sendQuotaPool = newQuotaPool(int(t.streamSendQuota))
	t.activeStreams[s.id] = s
	t.mu.Unlock()
	s.windowHandler = func(n int) {
		t.updateWindow(s, uint32(n))
	}
	if hDec.state.timeoutSet {
		s.ctx, s.cancel = context.WithTimeout(context.TODO(), hDec.state.timeout)
	} else {
		s.ctx, s.cancel = context.WithCancel(context.TODO())
	}
	// Attach Auth info if there is any.
	if t.authInfo != nil {
		s.ctx = credentials.NewContext(s.ctx, t.authInfo)
	}
	// Cache the current stream to the context so that the server application
	// can find out. Required when the server wants to send some metadata
	// back to the client (unary call only).
	s.ctx = newContextWithStream(s.ctx, s)
	// Attach the received metadata to the context.
	if len(hDec.state.mdata) > 0 {
		s.ctx = metadata.NewContext(s.ctx, hDec.state.mdata)
	}

	s.dec = &recvBufferReader{
		ctx:  s.ctx,
		recv: s.buf,
	}
	s.method = hDec.state.method

	wg.Add(1)
	go func() {
		handle(s)
		wg.Done()
	}()
	return nil
}

func (s *Server) handleStream(t transport.ServerTransport, stream *transport.Stream, trInfo *traceInfo) {
	sm := stream.Method()
	if sm != "" && sm[0] == '/' {
		sm = sm[1:]
	}
	pos := strings.LastIndex(sm, "/")
	if pos == -1 {
		if trInfo != nil {
			trInfo.tr.LazyLog(&fmtStringer{"Malformed method name %q", []interface{}{sm}}, true)
			trInfo.tr.SetError()
		}
		if err := t.WriteStatus(stream, codes.InvalidArgument, fmt.Sprintf("malformed method name: %q", stream.Method())); err != nil {
			if trInfo != nil {
				trInfo.tr.LazyLog(&fmtStringer{"%v", []interface{}{err}}, true)
				trInfo.tr.SetError()
			}
			grpclog.Printf("grpc: Server.handleStream failed to write status: %v", err)
		}
		if trInfo != nil {
			trInfo.tr.Finish()
		}
		return
	}
	service := sm[:pos]
	method := sm[pos+1:]
	srv, ok := s.m[service]
	if !ok {
		if trInfo != nil {
			trInfo.tr.LazyLog(&fmtStringer{"Unknown service %v", []interface{}{service}}, true)
			trInfo.tr.SetError()
		}
		if err := t.WriteStatus(stream, codes.Unimplemented, fmt.Sprintf("unknown service %v", service)); err != nil {
			if trInfo != nil {
				trInfo.tr.LazyLog(&fmtStringer{"%v", []interface{}{err}}, true)
				trInfo.tr.SetError()
			}
			grpclog.Printf("grpc: Server.handleStream failed to write status: %v", err)
		}
		if trInfo != nil {
			trInfo.tr.Finish()
		}
		return
	}
	// Unary RPC or Streaming RPC?
	if md, ok := srv.md[method]; ok {
		s.processUnaryRPC(t, stream, srv, md, trInfo)
		return
	}
	if sd, ok := srv.sd[method]; ok {
		s.processStreamingRPC(t, stream, srv, sd, trInfo)
		return
	}
	if trInfo != nil {
		trInfo.tr.LazyLog(&fmtStringer{"Unknown method %v", []interface{}{method}}, true)
		trInfo.tr.SetError()
	}
	if err := t.WriteStatus(stream, codes.Unimplemented, fmt.Sprintf("unknown method %v", method)); err != nil {
		if trInfo != nil {
			trInfo.tr.LazyLog(&fmtStringer{"%v", []interface{}{err}}, true)
			trInfo.tr.SetError()
		}
		grpclog.Printf("grpc: Server.handleStream failed to write status: %v", err)
	}
	if trInfo != nil {
		trInfo.tr.Finish()
	}
}