Repository: grpc/grpc-go
File: test/end2end_test.go (Go)

Code before:
	}
}

func testClientStreamingRecvAfterCloseError(t *testing.T, e env) {
	te := newTest(t, e)
	te.startServer(&testServer{security: e.security, recvAfterClose: true})
	defer te.tearDown()
	tc := testgrpc.NewTestServiceClient(te.clientConn())

	stream, err := tc.StreamingInputCall(te.ctx)
	if err != nil {
		t.Fatalf("%v.StreamingInputCall(_) = _, %v, want <nil>", tc, err)
	}
	payload, err := newPayload(testpb.PayloadType_COMPRESSABLE, 1)
	if err != nil {
		t.Fatal(err)
	}

	req := &testpb.StreamingInputCallRequest{
		Payload: payload,
	}

	for {
		if err = stream.Send(req); err == nil {
			continue
		}
		if status.Code(err) != codes.Internal {
			t.Fatalf("%v.CloseAndRecv() = %v, want error %s", stream, err, codes.Internal)
		}
	}
}

func (s) TestExceedMaxStreamsLimit(t *testing.T) {
	for _, e := range listTestEnv() {
		testExceedMaxStreamsLimit(t, e)
	}
}

func testExceedMaxStreamsLimit(t *testing.T, e env) {
	te := newTest(t, e)
	te.declareLogNoise(
		"http2Client.notifyError got notified that the client transport was broken",
		"Conn.resetTransport failed to create client transport",
		"grpc: the connection is closing",
	)
	te.maxStream = 1 // Only allows 1 live stream per server transport.
	te.startServer(&testServer{security: e.security})
	defer te.tearDown()

	cc := te.clientConn()
	tc := testgrpc.NewTestServiceClient(cc)

Review comment (suggestion):
If send returns `io.EOF`, you need to call `Recv()` to get the RPC status. See https://pkg.go.dev/google.golang.org/grpc#ClientStream

Code after:
	}
	if err := ss.Start(nil); err != nil {
		t.Fatal("Error starting server:", err)
	}
	defer ss.Stop()

	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
	defer cancel()
	stream, err := ss.Client.StreamingInputCall(ctx)
	if err != nil {
		t.Fatalf(".StreamingInputCall(_) = _, %v, want <nil>", err)
	}
	if err := stream.Send(&testpb.StreamingInputCallRequest{}); err != nil {
		t.Fatalf("stream.Send(_) = %v, want <nil>", err)
	}
	if resp, err := stream.CloseAndRecv(); err != nil || !proto.Equal(resp, &testpb.StreamingInputCallResponse{}) {
		t.Fatalf("stream.CloseSend() = %v , %v, want non-nil empty response, <nil>", resp, err)
	}
}

// Tests that Recv() on client streaming client blocks till the server handler
// returns even after calling SendAndClose from the server handler.
func (s) TestClientStreaming_RecvWaitsForServerHandlerRetrun(t *testing.T) {
	waitForReturn := make(chan struct{})
	ss := stubserver.StubServer{
		StreamingInputCallF: func(stream testgrpc.TestService_StreamingInputCallServer) error {
			if err := stream.SendAndClose(&testpb.StreamingInputCallResponse{}); err != nil {
				t.Errorf("stream.SendAndClose(_) = %v, want <nil>", err)
			}
			<-waitForReturn
			return nil
		},
	}
	if err := ss.Start(nil); err != nil {
		t.Fatal("Error starting server:", err)
	}
	defer ss.Stop()

	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
	defer cancel()
	stream, err := ss.Client.StreamingInputCall(ctx)
	if err != nil {
		t.Fatalf(".StreamingInputCall(_) = _, %v, want <nil>", err)
	}
	// Start Recv in a goroutine to test if it blocks until the server handler returns.
	errCh := make(chan error, 1)
	go func() {
		resp := new(testpb.StreamingInputCallResponse)
		err := stream.RecvMsg(resp)
		errCh <- err
	}()