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

Code before:

	req := &testpb.StreamingInputCallRequest{Payload: payload}
	for {
		if err = stream.Send(req); err == io.EOF {
			break
		}
		if err != nil {
			t.Fatalf("Stream.Send(req) = %v, want <nil>", err)
		}
		select {
		case <-ctx.Done():
			t.Fatal("timed out waiting for error from server")
		default:
		}
	}
	_, err = stream.CloseAndRecv()
	if err != nil {
		t.Fatalf("stream.CloseAndRecv() = %v, want error", err)
	}
	if status.Code(err) != codes.OK {
		t.Fatalf("stream.CloseAndRecv() = %v, want error %s", err, codes.OK)
	}
}

// Tests the scenario where a client-streaming server sends an error after calling
// `SendAndClose()`. The error should be ignored by client because RPC should be
// closed when `SendAndClose()` is called.
func (s) TestClientStreamingCardinalityViolation_IgnoreServerHandlerErrorAfterSendAndClose(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),
			})
			return status.Errorf(codes.Unimplemented, "error for testing")
		},
	}
	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()


Review comment (suggestion):
```suggestion
// `SendAndClose()`. The error should be ignored by the client because the RPC should be
```

Please re-wrap.

Code after:
		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
	}()

	// Check that Recv() is blocked.
	select {
	case err := <-errCh:
		t.Fatalf("stream.RecvMsg(_) = %v returned unexpectedly", err)
	case <-time.After(defaultTestShortTimeout):
	}

	close(waitForReturn)

	// Recv() should return after the server handler returns.
	select {
	case err := <-errCh:
		if err != nil {
			t.Fatalf("stream.RecvMsg(_) = %v, want <nil>", err)
		}
	case <-time.After(defaultTestTimeout):
		t.Fatal("Timed out waiting for stream.RecvMsg(_) to return")
	}
}

// Tests the behavior where server handler returns an error after calling
// SendAndClose. It verifies the that client receives nil message and
// non-nil error.
func (s) TestClientStreaming_ReturnErrorAfterSendAndClose(t *testing.T) {