Method generateTraffic

perf/perf.go:222–272

(ctx context.Context, latencyCh chan int64, failureCount *int64)

}

func (p *perf) generateTraffic(ctx context.Context, latencyCh chan int64, failureCount *int64) {
	limiter := rate.NewLimiter(rate.Limit(p.config.RequestRate), int(p.config.RequestRate))

	count := 0
	for {
		if err := limiter.Wait(ctx); err != nil {
			return
		}
		c := count
		count++
		person := Person{Name: "rbt", Money: c, Expected: c + 1}
		jsonBytes, err := json.Marshal(person)
		if err != nil {
			slog.Error(
				"Failed to marshal Person",
				slog.Any("error", err),
			)
			os.Exit(1)
		}
		start := time.Now()
		if !common.SendToChannel(ctx, p.input, contube.NewRecordImpl(jsonBytes, func() {})) {
			return
		}
		go func() {
			e, ok := common.ReceiveFromChannel(ctx, p.output)
			if !ok {
				return
			}
			latencyCh <- time.Since(start).Microseconds()
			payload := e.GetPayload()
			e.Commit()
			var out Person
			err = json.Unmarshal(payload, &out)
			if err != nil {
				slog.Error(
					"Failed to unmarshal Person",
					slog.Any("error", err),
					slog.Any("payload", payload),
				)
				os.Exit(1)
			}
			if out.Money != out.Expected {
				slog.Error(
					"Unexpected value for money",
					slog.Any("money", out.Money),
				)
				atomic.AddInt64(failureCount, 1)
			}
		}()
	}
}