Method TestServerStreaming_ClientCallSendMsgTwice

test/end2end_test.go:3952–4019

Tests the behavior for server-side streaming when client calls SendMsg twice. Second call to SendMsg should fail with Internal error and result in closing the connection with a RST_STREAM.

(t *testing.T)

// Second call to SendMsg should fail with Internal error and result in closing
// the connection with a RST_STREAM.
func (s) TestServerStreaming_ClientCallSendMsgTwice(t *testing.T) {
	// To ensure initial call to server.recvMsg() made by the generated code is successfully
	// completed. Otherwise, if the client attempts to send a second request message, that
	// will trigger a RST_STREAM from the client due to the application violating the RPC's
	// protocol. The RST_STREAM could cause the server’s first RecvMsg to fail and will prevent
	// the method handler from being called.
	recvDoneOnServer := make(chan struct{})
	// To ensure goroutine for test does not end before RPC handler performs error
	// checking.
	handlerDone := make(chan struct{})
	ss := stubserver.StubServer{
		StreamingOutputCallF: func(_ *testpb.StreamingOutputCallRequest, stream testgrpc.TestService_StreamingOutputCallServer) error {
			close(recvDoneOnServer)
			// Block until the stream’s context is done. Second call to client.SendMsg
			// triggers a RST_STREAM which cancels the stream context on the server.
			<-stream.Context().Done()
			var err error
			waitCtx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
			defer cancel()
			for ; waitCtx.Err() == nil; <-time.After(time.Millisecond) {
				if err = stream.SendMsg(&testpb.StreamingOutputCallRequest{}); err != nil {
					break
				}
			}
			if waitCtx.Err() != nil {
				t.Fatalf("Context timed out waiting for stream.SendMsg() to return an error")
			}
			if status.Code(err) != codes.Canceled {
				t.Errorf("stream.SendMsg() = %v, want error %v", err, codes.Canceled)
			}
			close(handlerDone)
			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()
	cc, err := grpc.NewClient(ss.Address, grpc.WithTransportCredentials(local.NewCredentials()))
	if err != nil {
		t.Fatalf("grpc.NewClient(%q) failed unexpectedly: %v", ss.Address, err)
	}
	defer cc.Close()

	desc := &grpc.StreamDesc{
		StreamName:    "StreamingOutputCall",
		ServerStreams: true,
		ClientStreams: false,
	}

	stream, err := cc.NewStream(ctx, desc, "/grpc.testing.TestService/StreamingOutputCall")
	if err != nil {
		t.Fatalf("cc.NewStream() failed unexpectedly: %v", err)
	}