Method serveStreams

server.go:1053–1091

(ctx context.Context, st transport.ServerTransport, rawConn net.Conn)

}

func (s *Server) serveStreams(ctx context.Context, st transport.ServerTransport, rawConn net.Conn) {
	ctx = transport.SetConnection(ctx, rawConn)
	ctx = peer.NewContext(ctx, st.Peer())
	if s.statsHandler != nil {
		ctx = s.statsHandler.TagConn(ctx, &stats.ConnTagInfo{
			RemoteAddr: st.Peer().Addr,
			LocalAddr:  st.Peer().LocalAddr,
		})
		s.statsHandler.HandleConn(ctx, &stats.ConnBegin{})
	}

	defer func() {
		st.Close(errors.New("finished serving streams for the server transport"))
		if s.statsHandler != nil {
			s.statsHandler.HandleConn(ctx, &stats.ConnEnd{})
		}
	}()

	streamQuota := newHandlerQuota(s.opts.maxConcurrentStreams)
	st.HandleStreams(ctx, func(stream *transport.ServerStream) {
		s.handlersWG.Add(1)
		streamQuota.acquire()
		f := func() {
			defer streamQuota.release()
			defer s.handlersWG.Done()
			s.handleStream(st, stream)
		}

		if s.opts.numServerWorkers > 0 {
			select {
			case s.serverWorkerChannel <- f:
				return
			default:
				// If all stream workers are busy, fallback to the default code path.
			}
		}
		go f()
	})
}

var _ http.Handler = (*Server)(nil)