rename some inner types

Author: apolcyn

test/end2end_test.go

@@ -2361,7 +2361,7 @@ func testMetadataStreamingRPC(t *testing.T, e env) {
 func TestServerStreaming(t *testing.T) {
 	serverRespSizes := [][]int{
 		[]int{27182, 8, 1828, 45904},
-		[]int{2000000, 2000000, 2000000, 2000000, 2000000, 2000000, 2000000, 2000000},
+		[]int{(1 << 21), (1 << 21), (1 << 21), (1 << 21), (1 << 21), (1 << 21), (1 << 21), (1 << 21)},
 	}
 	defer leakCheck(t)()
 	for _, s := range serverRespSizes {
@@ -2537,7 +2537,7 @@ func TestClientStreaming(t *testing.T) {
 	defer leakCheck(t)()
 	clientReqSizes := [][]int{
 		[]int{27182, 8, 1828, 45904},
-		[]int{20000000, 2000000, 2000000, 2000000, 2000000, 2000000, 2000000, 2000000},
+		[]int{(1 << 21), (1 << 21), (1 << 21), (1 << 21), (1 << 21), (1 << 21), (1 << 21), (1 << 21)},
 	}
 	for _, s := range clientReqSizes {
 		for _, e := range listTestEnv() {
@@ -2559,7 +2559,6 @@ func testClientStreaming(t *testing.T, e env, clientReqSizes []int) {
 
 	var sum int
 	for _, s := range clientReqSizes {
-		grpclog.Println("sending another message of size " + string(s))
 		payload, err := newPayload(testpb.PayloadType_COMPRESSABLE, int32(s))
 		if err != nil {
 			t.Fatal(err)

transport/http2_client.go

@@ -300,13 +300,13 @@ func (t *http2Client) newStream(ctx context.Context, callHdr *CallHdr) *Stream {
 	// That means, s.ctx should be read-only. And s.ctx is done iff ctx is done.
 	// So we use the original context here instead of creating a copy.
 	s.ctx = ctx
-	windowReader := func(n uint32) {
+	consumeConnectionWindow := func(n uint32) {
 		t.updateWindow(s, n)
 	}
-	streamWindowPreconsumer := func(n uint32) {
+	preConsumeStreamWindow := func(n uint32) {
 		t.preConsumeWindowSpace(s, n)
 	}
-	s.setClientStreamReader(windowReader, streamWindowPreconsumer)
+	s.setClientStreamReader(consumeConnectionWindow, preConsumeStreamWindow)
 	return s
 }
 

transport/http2_server.go

@@ -296,13 +296,13 @@ func (t *http2Server) operateHeaders(frame *http2.MetaHeadersFrame, handle func(
 		}
 		t.stats.HandleRPC(s.ctx, inHeader)
 	}
-	windowReader := func(n uint32) {
+	consumeConnectionWindow := func(n uint32) {
 		t.updateWindow(s, n)
 	}
-	streamWindowPreconsumer := func(n uint32) {
+	preConsumeStreamWindow := func(n uint32) {
 		t.preConsumeWindowSpace(s, n)
 	}
-	s.setServerStreamReader(windowReader, streamWindowPreconsumer)
+	s.setServerStreamReader(consumeConnectionWindow, preConsumeStreamWindow)
 	handle(s)
 	return
 }

transport/transport.go

@@ -221,33 +221,33 @@ type Stream struct {
 	rstError http2.ErrCode
 }
 
-func (s *Stream) setClientStreamReader(windowReader func(uint32), streamWindowPreConsumer func(uint32)) {
+func (s *Stream) setClientStreamReader(consumeConnectionWindow func(uint32), preConsumeStreamWindow func(uint32)) {
 	if s.ctx == nil || s.goAway == nil || s.buf == nil {
 		grpclog.Fatalf("Uninitialized stream. s.ctx == nil: %v; s.goAway == nil: %v; s.buf == nil: %v", s.ctx == nil, s.goAway == nil, s.buf == nil)
 	}
-	s.setStreamReader(windowReader, streamWindowPreConsumer)
+	s.setStreamReader(consumeConnectionWindow, preConsumeStreamWindow)
 }
 
-func (s *Stream) setServerStreamReader(windowReader func(uint32), streamWindowPreConsumer func(uint32)) {
+func (s *Stream) setServerStreamReader(consumeConnectionWindow func(uint32), preConsumeStreamWindow func(uint32)) {
 	if s.ctx == nil || s.buf == nil {
 		grpclog.Fatalf("Uninitialized stream. s.ctx == nil: %v; s.buf == nil: %v", s.ctx == nil, s.buf == nil)
 	}
 	if s.goAway != nil {
 		grpclog.Fatalf("Unexpected initialization of s.goAway; got %v; want nil", s.goAway)
 	}
-	s.setStreamReader(windowReader, streamWindowPreConsumer)
+	s.setStreamReader(consumeConnectionWindow, preConsumeStreamWindow)
 }
 
-func (s *Stream) setStreamReader(windowReader func(uint32), streamWindowPreConsumer func(uint32)) {
-	s.streamReader = &streamWindowUpdaterReader{
-		streamWindowPreConsumer: streamWindowPreConsumer,
-		twr: &transportWindowUpdaterReader{
+func (s *Stream) setStreamReader(consumeConnectionWindow func(uint32), preConsumeStreamWindow func(uint32)) {
+	s.streamReader = &streamWindowPreconsumingReader{
+		preConsumeStreamWindow: preConsumeStreamWindow,
+		cwr: &connectionWindowConsumingReader{
 			dec: &recvBufferReader{
 				ctx:    s.ctx,
 				goAway: s.goAway,
 				recv:   s.buf,
 			},
-			windowReader: windowReader,
+			consumeConnectionWindow: consumeConnectionWindow,
 		},
 	}
 }
@@ -273,39 +273,39 @@ func (s *Stream) ReadFull(p []byte) (n int, err error) {
 	return io.ReadFull(s.streamReader, p)
 }
 
-type streamWindowUpdaterReader struct {
+type streamWindowPreconsumingReader struct {
 	// The handler to control the window update procedure for both this
 	// particular stream and the associated transport.
-	streamWindowPreConsumer func(uint32)
-	twr                     *transportWindowUpdaterReader
-	readFullErr             error
+	preConsumeStreamWindow func(uint32)
+	cwr                    *connectionWindowConsumingReader
+	readFullErr            error
 }
 
-func (r *streamWindowUpdaterReader) Read(p []byte) (n int, err error) {
+func (r *streamWindowPreconsumingReader) Read(p []byte) (n int, err error) {
 	if r.readFullErr != nil {
 		return 0, r.readFullErr
 	}
 	defer func() { r.readFullErr = err }()
 	readAmount := min(maxSingleStreamWindowUpdate, uint32(len(p)))
-	r.streamWindowPreConsumer(readAmount)
-	return io.ReadFull(r.twr, p[:readAmount])
+	r.preConsumeStreamWindow(readAmount)
+	return io.ReadFull(r.cwr, p[:readAmount])
 }
 
-type transportWindowUpdaterReader struct {
-	dec          io.Reader
-	windowReader func(uint32)
+type connectionWindowConsumingReader struct {
+	dec                     io.Reader
+	consumeConnectionWindow func(uint32)
 }
 
 // Read reads all the data available for this Stream from the transport and
 // passes them into the decoder, which converts them into a gRPC message stream.
 // The error is io.EOF when the stream is done or another non-nil error if
 // the stream broke.
-func (s *transportWindowUpdaterReader) Read(p []byte) (n int, err error) {
+func (s *connectionWindowConsumingReader) Read(p []byte) (n int, err error) {
 	n, err = s.dec.Read(p)
 	if err != nil {
 		return
 	}
-	s.windowReader(uint32(n))
+	s.consumeConnectionWindow(uint32(n))
 	return
 }
 

transport/transport_test.go

@@ -1112,7 +1112,7 @@ func TestClientWithMisbehavedServer(t *testing.T) {
 		t.Fatalf("Failed to write: %v", err)
 	}
 	// reflect to get the inner recvBufferReader, which reads without doing window updates
-	recvBufferReader := s.streamReader.(*io.LimitedReader).R.(*windowUpdatingReader).pr.dec
+	recvBufferReader := s.streamReader.(*streamWindowPreconsumingReader).cwr.dec
 	// Read without window update.
 	for {
 		p := make([]byte, http2MaxFrameLen)
@@ -1174,7 +1174,8 @@ func TestEncodingRequiredStatus(t *testing.T) {
 	}
 	p := make([]byte, http2MaxFrameLen)
 	// reflect to get the plain recvBufferReader from the stream's stream reader, which doesn't do window updates
-	if _, err := s.streamReader.(*io.LimitedReader).R.(*windowUpdatingReader).pr.dec.Read(p); err != io.EOF {
+	recvBufferReader := s.streamReader.(*streamWindowPreconsumingReader).cwr.dec
+	if _, err := recvBufferReader.Read(p); err != io.EOF {
 		t.Fatalf("Read got error %v, want %v", err, io.EOF)
 	}
 	if s.StatusCode() != encodingTestStatusCode || s.StatusDesc() != encodingTestStatusDesc {
@@ -1203,7 +1204,7 @@ func TestInvalidHeaderField(t *testing.T) {
 	}
 	p := make([]byte, http2MaxFrameLen)
 	// reflect to get the inner recvBufferReader, which reads without doing window updates
-	_, err = s.streamReader.(*io.LimitedReader).R.(*windowUpdatingReader).pr.dec.Read(p)
+	_, err = s.streamReader.(*streamWindowPreconsumingReader).cwr.dec.Read(p)
 	if se, ok := err.(StreamError); !ok || se.Code != codes.FailedPrecondition || !strings.Contains(err.Error(), expectedInvalidHeaderField) {
 		t.Fatalf("Read got error %v, want error with code %s and contains %q", err, codes.FailedPrecondition, expectedInvalidHeaderField)
 	}