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

Code before:
	_, err = stream.CloseAndRecv()
	if err == nil {
		t.Fatalf("stream.CloseAndRecv() = %v, want an error", err)
	}
	if status.Code(err) != codes.Internal {
		t.Fatalf("stream.CloseAndRecv() = %v, want error %s", err, codes.Internal)
	}
}

// Test to verify for client-streaming RPCs, when SendAndClose is called, the server
// should reset stream after sending the response message successfully.
// If server calls Recv after, those operations should fail.
func (s) TestClientStreamingCardinalityViolation_ServerHandlerRecvAfterSendAndClose(t *testing.T) {
	// TODO : https://github.com/grpc/grpc-go/issues/8119 - remove `t.Skip()`
	// after this is fixed.
	t.Skip()
	ss := &stubserver.StubServer{
		StreamingInputCallF: func(stream testgrpc.TestService_StreamingInputCallServer) error {
			sum := 0
			in, _ := stream.Recv()
			p := in.GetPayload().GetBody()
			sum += len(p)
			stream.SendAndClose(&testpb.StreamingInputCallResponse{
				AggregatedPayloadSize: int32(sum),
			})
			_, err := stream.Recv()
			return err
		},
	}
	if err := ss.Start(nil); err != nil {
		t.Fatalf("Error starting endpoint server: %v", 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)
	}
	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 == io.EOF {
			break
		}

Review comment (bug):
so, we should verify if `stream.Recv()` actually failed. Can do `logger.Fatalf()` if err is nil. Then just return as the stream is no longer available. See here https://github.com/grpc/grpc-go/blob/eb744dec5da784e7e2fc140ddaa6d2aa645b5371/stream_interfaces.go#L94

I don't think for this test case we need to return the error from handler

Code after:

	_, err = stream.CloseAndRecv()
	if err == nil {
		t.Fatalf("stream.CloseAndRecv() = %v, want an error", err)
	}
	if status.Code(err) != codes.Internal {
		t.Fatalf("stream.CloseAndRecv() = %v, want error %s", err, codes.Internal)
	}
}

// Tests that the server can continue to receive messages after calling SendAndClose. Although
// this is unexpected, we retain it for backward compatibility.
func (s) TestClientStreaming_ServerHandlerRecvAfterSendAndClose(t *testing.T) {
	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)
			}
			if resp, err := stream.Recv(); err != nil || !proto.Equal(resp, &testpb.StreamingInputCallRequest{}) {
				t.Errorf("stream.Recv() = %s, %v, want non-nil empty response, <nil>", resp, err)
			}
			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)
	}
	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)