Function setupStream

benchmark/benchmain/main.go:488–521

(bf stats.Features, unconstrained bool)

}

func setupStream(bf stats.Features, unconstrained bool) ([][]testgrpc.BenchmarkService_StreamingCallClient, *testpb.SimpleRequest, rpcCleanupFunc) {
	clients, cleanup := makeClients(bf)

	streams := make([][]testgrpc.BenchmarkService_StreamingCallClient, bf.Connections)
	ctx := context.Background()
	if unconstrained {
		md := metadata.Pairs(benchmark.UnconstrainedStreamingHeader, "1", benchmark.UnconstrainedStreamingDelayHeader, bf.SleepBetweenRPCs.String())
		ctx = metadata.NewOutgoingContext(ctx, md)
	}
	if bf.EnablePreloader {
		md := metadata.Pairs(benchmark.PreloadMsgSizeHeader, strconv.Itoa(bf.RespSizeBytes), benchmark.UnconstrainedStreamingDelayHeader, bf.SleepBetweenRPCs.String())
		ctx = metadata.NewOutgoingContext(ctx, md)
	}
	for cn := 0; cn < bf.Connections; cn++ {
		tc := clients[cn]
		streams[cn] = make([]testgrpc.BenchmarkService_StreamingCallClient, bf.MaxConcurrentCalls)
		for pos := 0; pos < bf.MaxConcurrentCalls; pos++ {
			stream, err := tc.StreamingCall(ctx)
			if err != nil {
				logger.Fatalf("%v.StreamingCall(_) = _, %v", tc, err)
			}
			streams[cn][pos] = stream
		}
	}

	pl := benchmark.NewPayload(testpb.PayloadType_COMPRESSABLE, bf.ReqSizeBytes)
	req := &testpb.SimpleRequest{
		ResponseType: pl.Type,
		ResponseSize: int32(bf.RespSizeBytes),
		Payload:      pl,
	}

	return streams, req, cleanup
}

func prepareMessages(streams [][]testgrpc.BenchmarkService_StreamingCallClient, req *testpb.SimpleRequest) [][]*grpc.PreparedMsg {
	preparedMsg := make([][]*grpc.PreparedMsg, len(streams))