Method Run

perf/perf.go:81–220

(ctx context.Context)

}

func (p *perf) Run(ctx context.Context) {
	slog.Info(
		"Starting Function stream perf client",
		slog.Any("config", p.config),
	)

	name := "perf-" + strconv.Itoa(rand.Int())
	var f adminclient.ModelFunction
	if p.config.Func != nil {
		f = *p.config.Func
	} else {
		f = adminclient.ModelFunction{
			Runtime: adminclient.ModelRuntimeConfig{
				Config: map[string]interface{}{
					common.RuntimeArchiveConfigKey: "./bin/example_basic.wasm",
				},
			},
			Source: utils.MakeMemorySourceTubeConfig("test-input-" + strconv.Itoa(rand.Int())),
			Sink:   *utils.MakeMemorySinkTubeConfig("test-output-" + strconv.Itoa(rand.Int())),
		}
	}
	f.Name = name

	queueFactory, err := p.tubeBuilder(ctx)
	if err != nil {
		slog.Error(
			"Failed to create Record Queue Factory",
			slog.Any("error", err),
		)
		os.Exit(1)
	}

	inputTopic, err := utils.GetInputTopics(&f)
	if err != nil {
		slog.Error(
			"Failed to get input topics",
			slog.Any("error", err),
		)
		os.Exit(1)

	}
	p.input, err = queueFactory.NewSinkTube(ctx, (&contube.SinkQueueConfig{
		Topic: inputTopic[0],
	}).ToConfigMap())
	if err != nil {
		slog.Error(
			"Failed to create Sink Perf Channel",
			slog.Any("error", err),
		)
		os.Exit(1)
	}

	outputTopic, err := utils.GetOutputTopic(&f)
	if err != nil {
		slog.Error(
			"Failed to get output topic",
			slog.Any("error", err),
		)