Function testFlowControlAccountCheck

internal/transport/transport_test.go:1879–2010

(t *testing.T, msgSize int, wc windowSizeConfig)

}

func testFlowControlAccountCheck(t *testing.T, msgSize int, wc windowSizeConfig) {
	sc := &ServerConfig{
		InitialWindowSize:     wc.serverStream,
		InitialConnWindowSize: wc.serverConn,
		StaticWindowSize:      true,
		BufferPool:            mem.DefaultBufferPool(),
	}
	co := ConnectOptions{
		InitialWindowSize:     wc.clientStream,
		InitialConnWindowSize: wc.clientConn,
		StaticWindowSize:      true,
		BufferPool:            mem.DefaultBufferPool(),
	}
	server, client, cancel := setUpWithOptions(t, 0, sc, pingpong, co)
	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
	})
	var st *http2Server
	server.mu.Lock()
	for k := range server.conns {
		st = k.(*http2Server)
	}
	server.mu.Unlock()

	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
	defer cancel()
	const numStreams = 5
	clientStreams := make([]*ClientStream, numStreams)
	for i := 0; i < numStreams; i++ {
		var err error
		clientStreams[i], err = client.NewStream(ctx, &CallHdr{}, nil)
		if err != nil {
			t.Fatalf("Failed to create stream. Err: %v", err)
		}
	}
	var wg sync.WaitGroup
	// For each stream send pingpong messages to the server.
	for _, stream := range clientStreams {
		wg.Add(1)
		go func(stream *ClientStream) {
			defer wg.Done()
			buf := make([]byte, msgSize+5)
			buf[0] = byte(0)
			binary.BigEndian.PutUint32(buf[1:], uint32(msgSize))
			opts := WriteOptions{}
			header := make([]byte, 5)
			for i := 1; i <= 5; i++ {
				if err := stream.Write(nil, newBufferSlice(buf), &opts); err != nil {
					t.Errorf("Error on client while writing message %v on stream %v: %v", i, stream.id, err)
					return