Function runPingPongTest

internal/transport/transport_test.go:2517–2563

This is a stress-test of flow control logic.

(t *testing.T, msgSize int)

// This is a stress-test of flow control logic.
func runPingPongTest(t *testing.T, msgSize int) {
	server, client, cancel := setUp(t, 0, pingpong)
	defer cancel()
	defer server.stop()
	defer client.Close(fmt.Errorf("closed manually by test"))
	waitWhileTrue(t, func() (bool, error) {
		server.mu.Lock()
		defer server.mu.Unlock()
		if len(server.conns) == 0 {
			return true, fmt.Errorf("timed out while waiting for server transport to be created")
		}
		return false, nil
	})
	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
	defer cancel()
	stream, err := client.NewStream(ctx, &CallHdr{}, nil)
	if err != nil {
		t.Fatalf("Failed to create stream. Err: %v", err)
	}
	msg := make([]byte, msgSize)
	outgoingHeader := make([]byte, 5)
	outgoingHeader[0] = byte(0)
	binary.BigEndian.PutUint32(outgoingHeader[1:], uint32(msgSize))
	opts := &WriteOptions{}
	incomingHeader := make([]byte, 5)

	ctx, cancel = context.WithTimeout(ctx, 10*time.Millisecond)
	defer cancel()
	for ctx.Err() == nil {
		if err := stream.Write(outgoingHeader, newBufferSlice(msg), opts); err != nil {
			t.Fatalf("Error on client while writing message. Err: %v", err)
		}
		if _, err := stream.readTo(incomingHeader); err != nil {
			t.Fatalf("Error on client while reading data header. Err: %v", err)
		}
		sz := binary.BigEndian.Uint32(incomingHeader[1:])
		recvMsg := make([]byte, int(sz))
		if _, err := stream.readTo(recvMsg); err != nil {
			t.Fatalf("Error on client while reading data. Err: %v", err)
		}
	}

	stream.Write(nil, nil, &WriteOptions{Last: true})
	if _, err := stream.readTo(incomingHeader); err != io.EOF {
		t.Fatalf("Client expected EOF from the server. Got: %v", err)
	}
}

type tableSizeLimit struct {
	mu     sync.Mutex