services-gateway-benchmarks/src/main/java/io/scalecube/services/benchmarks/gateway/InfiniteStreamScenario.java
/**
   * Runner function for benchmarks.
   *
   * @param args program arguments
   * @param benchmarkStateFactory producer function for {@link AbstractBenchmarkState}
   */
Runner function for benchmarks.
@param args program arguments
@param benchmarkStateFactory producer function for  AbstractBenchmarkState
Runner function for benchmarks . @param args program arguments @param benchmarkStateFactory producer function for 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) 
            .rampUpDuration(rampUpDuration)
            .durationUnit(TimeUnit.MILLISECONDS)
            .build();
    AbstractBenchmarkState<?> benchmarkState = benchmarkStateFactory.apply(settings);
    benchmarkState.runWithRampUp(
        (rampUpTick, state) -> state.createClient(),
        state -> {
          LatencyHelper latencyHelper = new LatencyHelper(state);
          BenchmarkMeter serviceToClientMeter = state.meter("meter.service-to-client");
          Integer rateLimit = rateLimit(settings);
          ServiceMessage request = ServiceMessage.builder().qualifier(QUALIFIER).build();
          return client ->
              (executionTick, task) ->
                  Flux.defer(
                      () ->
                          client
                              .requestStream(request)
                              .map(InfiniteStreamScenario::enrichResponse)
                              .limitRate(rateLimit)
                              .doOnNext(
                                  message -> {
                                    serviceToClientMeter.mark();
                                    latencyHelper.calculate(message);
                                  })
                              .doOnError(
                                  th -> LOGGER.warn("Exception occured on requestStream: " + th)));
        },
        (state, client) -> {
          client.close();
          return client.onClose();
        });
  }
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 ) . rampUpDuration ( rampUpDuration ) . durationUnit ( TimeUnit . MILLISECONDS ) . build ( ) ; AbstractBenchmarkState < ? > benchmarkState = benchmarkStateFactory . apply ( settings ) ; benchmarkState . runWithRampUp ( ( rampUpTick , state ) -> state . createClient ( ) , state -> { LatencyHelper latencyHelper = new LatencyHelper ( state ) ; BenchmarkMeter serviceToClientMeter = state . meter ( "meter.service-to-client" ) ; Integer rateLimit = rateLimit ( settings ) ; ServiceMessage request = ServiceMessage . builder ( ) . qualifier ( QUALIFIER ) . build ( ) ; return client -> ( executionTick , task ) -> Flux . defer ( ( ) -> client . requestStream ( request ) . map ( InfiniteStreamScenario :: enrichResponse ) . limitRate ( rateLimit ) . doOnNext ( message -> { serviceToClientMeter . mark ( ) ; latencyHelper . calculate ( message ) ; } ) . doOnError ( th -> LOGGER . warn ( "Exception occured on requestStream: " + th ) ) ) ; } , ( state , client ) -> { client . close ( ) ; return client . onClose ( ) ; } ) ; }
Runner function for benchmarks.
Runner function for benchmarks .
// workaround