package io.scalecube.gateway.benchmarks;

import static io.scalecube.gateway.benchmarks.BenchmarksService.CLIENT_RECV_TIME;
import static io.scalecube.gateway.benchmarks.BenchmarksService.CLIENT_SEND_TIME;

import io.scalecube.benchmarks.BenchmarkSettings;
import io.scalecube.gateway.clientsdk.ClientMessage;
import io.scalecube.gateway.clientsdk.ReferenceCountUtil;
import java.time.Duration;
import java.util.Optional;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Scheduler;

public final class RequestOneScenario {

  private static final Logger LOGGER = LoggerFactory.getLogger(RequestOneScenario.class);

  private static final String QUALIFIER = "/benchmarks/one";

  private RequestOneScenario() {
    // Do not instantiate
  }

  /**
   * Runner function for benchmarks.
   *
   * @param args program arguments
   * @param benchmarkStateFactory producer function for {@link AbstractBenchmarkState}
   */
  public static void runWith(
      String[] args, Function<BenchmarkSettings, AbstractBenchmarkState<?>> benchmarkStateFactory) {

    int numOfThreads = Runtime.getRuntime().availableProcessors();
    Duration rampUpDuration = Duration.ofSeconds(numOfThreads);

    BenchmarkSettings settings =
        BenchmarkSettings.from(args)
            .injectors(numOfThreads)
            .messageRate(1) // workaround
            .rampUpDuration(rampUpDuration)
            .durationUnit(TimeUnit.MILLISECONDS)
            .build();

    AbstractBenchmarkState<?> benchmarkState = benchmarkStateFactory.apply(settings);

    benchmarkState.runWithRampUp(
        (rampUpTick, state) -> state.createClient(),
        state -> {
          LatencyHelper latencyHelper = new LatencyHelper(state);

          return client ->
              (executionTick, task) ->
                  Mono.defer(
                      () -> {
                        Scheduler taskScheduler = task.scheduler();
                        return client
                            .requestResponse(enrichRequest(), taskScheduler)
                            .map(RequestOneScenario::enrichResponse)
                            .doOnError(
                                th -> LOGGER.warn("Exception occured on requestResponse: " + th))
                            .doOnNext(
                                msg -> {
                                  Optional.ofNullable(msg.data())
                                      .ifPresent(ReferenceCountUtil::safestRelease);
                                  latencyHelper.calculate(msg);
                                })
                            .doOnTerminate(() -> taskScheduler.schedule(task));
                      });
        },
        (state, client) -> client.close());
  }

  private static ClientMessage enrichResponse(ClientMessage msg) {
    return ClientMessage.from(msg).header(CLIENT_RECV_TIME, System.currentTimeMillis()).build();
  }

  private static ClientMessage enrichRequest() {
    return ClientMessage.builder()
        .qualifier(QUALIFIER)
        .header(CLIENT_SEND_TIME, System.currentTimeMillis())
        .build();
  }
}

@@ -818,16 +818,61 @@
 /one%22;%0A%0A
+  private static final int MULT_FACTOR = 2;%0A%0A
   privat
@@ -892,24 +892,24 @@
 cenario() %7B%0A
-
     // Do no
@@ -1235,16 +1235,163 @@
 ory) %7B%0A%0A
+    int multFactor =%0A        Integer.parseInt(%0A            BenchmarkSettings.from(args).build().find(%22multFactor%22, String.valueOf(MULT_FACTOR)));%0A%0A
     int 
@@ -1445,24 +1445,71 @@
 ocessors();%0A
+    int injectors = numOfThreads * multFactor;%0A
     Duration
@@ -1654,27 +1654,24 @@
 jectors(
-numOfThread
+injector
 s)%0A     

Add removed RequestOneScenario.MULT_FACTOR; set to 2 by default