Method TestLargeMessageWithDelayRead

internal/transport/transport_test.go:903–1002

(t *testing.T)

}

func (s) TestLargeMessageWithDelayRead(t *testing.T) {
	// Disable dynamic flow control.
	sc := &ServerConfig{
		BufferPool:            mem.DefaultBufferPool(),
		InitialWindowSize:     defaultWindowSize,
		InitialConnWindowSize: defaultWindowSize,
		StaticWindowSize:      true,
	}
	co := ConnectOptions{
		InitialWindowSize:     defaultWindowSize,
		InitialConnWindowSize: defaultWindowSize,
		StaticWindowSize:      true,
		BufferPool:            mem.DefaultBufferPool(),
	}
	server, ct, cancel := setUpWithOptions(t, 0, sc, delayRead, co)
	defer cancel()
	defer server.stop()
	defer ct.Close(fmt.Errorf("closed manually by test"))
	server.mu.Lock()
	ready := server.ready
	server.mu.Unlock()
	callHdr := &CallHdr{
		Host:   "localhost",
		Method: "foo.Large",
	}
	ctx, cancel := context.WithTimeout(context.Background(), time.Second*10)
	defer cancel()
	s, err := ct.NewStream(ctx, callHdr, nil)
	if err != nil {
		t.Fatalf("%v.NewStream(_, _) = _, %v, want _, <nil>", ct, err)
		return
	}
	// Wait for server's handler to be initialized
	select {
	case <-ready:
	case <-ctx.Done():
		t.Fatalf("Client timed out waiting for server handler to be initialized.")
	}
	server.mu.Lock()
	serviceHandler := server.h
	server.mu.Unlock()
	var (
		mu    sync.Mutex
		total int
	)
	s.wq.replenish = func(n int) {
		mu.Lock()
		total += n
		mu.Unlock()
		s.wq.realReplenish(n)
	}
	getTotal := func() int {
		mu.Lock()
		defer mu.Unlock()
		return total
	}
	done := make(chan struct{})
	defer close(done)